llama-cloud==0.1.7
llama-index==0.12.14
llama-index-agent-openai==0.4.1
llama-index-cli==0.4.0
llama-index-core==0.12.14
llama-index-embeddings-fastembed==0.3.0
llama-index-embeddings-huggingface==0.3.1
llama-index-embeddings-openai==0.3.1
llama-index-embeddings-openai-like==0.1.0
llama-index-indices-managed-llama-cloud==0.6.3
llama-index-legacy==0.9.48
llama-index-llms-anyscale==0.1.4
llama-index-llms-fireworks==0.1.5
llama-index-llms-huggingface==0.2.0
llama-index-llms-langchain==0.1.4
llama-index-llms-ollama==0.5.0
llama-index-llms-openai==0.3.13
llama-index-llms-openai-like==0.3.4
llama-index-multi-modal-llms-openai==0.4.1
llama-index-program-openai==0.3.1
llama-index-question-gen-openai==0.3.0
llama-index-readers-file==0.4.1
llama-index-readers-llama-parse==0.4.0
llama-index-utils-workflow==0.3.0
llama-index-vector-stores-chroma==0.4.1
llama-index-vector-stores-milvus==0.5.0

Code Snippet
# --- Import Statements ---
from __future__ import annotations
import os, ast, asyncio
from typing import List, Dict, Any, Union
import requests
from sqlalchemy import create_engine, text, inspect
from llama_index.core.workflow import (
    Workflow, step, StartEvent, StopEvent, Event, Context
)
from llama_index.core.tools import FunctionTool
from llama_index.core import Settings
from llama_index.core.prompts import PromptTemplate
from llama_index.llms.openai_like import OpenAILike

# --- Configure Gemini LLM ---
Settings.llm = OpenAILike(
    model="gemma3:latest",
    api_base="http://localhost:11434/v1",
    api_key="ollama",
    context_window=8000,
    is_chat_model=True,
    is_function_calling_model=True,
)

# --- Download DB ---
DB_PATH = "Chinook.db"
if not os.path.exists(DB_PATH):
    url = "https://storage.googleapis.com/benchmarks-artifacts/chinook/Chinook.db"
    with open(DB_PATH, "wb") as f:
        f.write(requests.get(url).content)

engine = create_engine(f"sqlite:///{DB_PATH}")

# --- Tool Functions ---
# (list_tables, schema, query, query_checker)
# ... (Omitted here for brevity — same as shared)

# --- Custom Event Classes ---
class ToolCallEvent(Event):
    tool_calls: List[Dict[str, Any]]

class ObservationEvent(Event):
    observations: List[str]

# --- ReAct Prompt ---
REACT_PROMPT = PromptTemplate("""
You are an agent designed to answer questions about a SQL database using the ReAct pattern.
...
""")

# --- SQLAgentWorkflow ---
class SQLAgentWorkflow(Workflow):

    async def _agent_common(self, scratchpad, question) -> Union[ToolCallEvent, StopEvent]:
        # (Tool parsing + ReAct logic)
        ...

    @step
    async def agent_start_step(self, ctx: Context, ev: StartEvent) -> Union[ToolCallEvent, StopEvent]:
        question = getattr(ev, "question", None) or getattr(ev, "kwargs", {}).get("question")
        if question is None:
            raise ValueError("No 'question' provided.")
        await ctx.set("question", question)
        scratchpad = f"Question: {question}\n"
        await ctx.set("scratchpad", scratchpad)
        return await self._agent_common(scratchpad, question)

    @step
    async def agent_continue_step(self, ctx: Context, ev: ObservationEvent) -> Union[ToolCallEvent, StopEvent]:
        scratchpad = await ctx.get("scratchpad", default="")
        for obs in ev.observations:
            scratchpad += f"Observation: {obs}\n"
        await ctx.set("scratchpad", scratchpad)
        question = await ctx.get("question")
        return await self._agent_common(scratchpad, question)

    @step
    async def tool_executor_step(self, ctx: Context, ev: ToolCallEvent) -> ObservationEvent:
        observations = []
        for call in ev.tool_calls:
            tool = next((t for t in tools if t.metadata.name == call["name"]), None)
            if tool:
                result = await asyncio.to_thread(tool.call, **call["args"])
                observations.append(str(result))
            else:
                observations.append(f"Tool '{call['name']}' not found.")
        return ObservationEvent(observations=observations)

# --- Runner ---
async def main():
    wf = SQLAgentWorkflow(timeout=180, verbose=True)
    result = await wf.run(question="Show the first 3 rows from the Artist table.")
    print(result)

if __name__ == "__main__":
    asyncio.run(main())