feat(ai)!: agents speak the LangChain create_agent contract

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "python.languageServer": "None"
+}

example/agents/.gitignore

@@ -7,3 +7,7 @@ storage
 node_modules
 /public/build
 /public/hot
+.ruff_cache
+.pytest_cache
+.ai
+.vite

example/agents/.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "python.languageServer": "None"
+}

example/agents/app/agents/agent.py

@@ -0,0 +1,123 @@
+import json
+
+from fastapi_startkit.ai import Middleware
+from fastapi_startkit.ai import state as ai_state
+from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
+from langchain_core.tools import BaseTool
+
+from app.agents.job_search_graph import USER_PROFILE
+from app.agents.remember import RememberMixin, ThreadAgent
+from app.agents.state import Context, RouterOutput
+from app.middleware.agent_logger import AgentLogger
+from app.models.message import Message
+from app.tools.job_search_tool import job_search_tool
+
+JOB_SEARCH_PROMPT = """You find jobs for a user. ALWAYS call job_search_tool exactly once —
+never reply with text only, and never ask clarifying questions.
+
+Derive role/skill/location keywords from the LATEST user message only (e.g. 'python
+developer remote'); it replaces any earlier search. Use earlier messages solely for
+preferences like location or remote. Ignore filler words like 'suggest', 'me', 'jobs'.
+If the latest message has no usable keywords, take them from the user's profile."""
+
+SUMMARIZER_PROMPT = """Write a short, friendly summary of these job listings for the user.
+Group by role type, mention company and location. If the list is empty, say so plainly.
+Use only what is given - do not invent jobs."""
+
+ROUTER_PROMPT = """You route a career assistant's queries.
+
+- job_search: the user wants job listings / openings / roles.
+- company_research: the user asks about a specific company.
+- chat: greetings (hi / hello), thanks, small talk, or anything conversational.
+
+Also decide which extra context the next node needs:
+- include_user_profile: the query is vague about role, skills or location, so the user's own profile helps.
+- include_last_job_search_response: the user is refining an earlier search.
+"""
+
+
+class JobSearchRouterAgent(RememberMixin, ThreadAgent):
+    """Classifies a query into a RouterOutput via structured output, answering
+    greetings inline through its `reply` field. Context: only the user input."""
+
+    model = "gemini-3.1-flash-lite"  # match the graph's model, not the lab default
+
+    def instructions(self) -> str:
+        return ROUTER_PROMPT
+
+    def schema(self) -> type:
+        return RouterOutput
+
+    def middleware(self) -> list[Middleware]:
+        return [AgentLogger()]
+
+    async def remember(self, state: dict) -> None:
+        # The router's output is routing state, except when it answers a greeting
+        # inline — that reply is conversation, so it goes on the thread.
+        decision = ai_state.structured(state)
+        if decision and decision.intent == "chat" and decision.reply:
+            await self.remember_row("text", {"text": decision.reply}, state)
+
+
+class JobSummarizerAgent(RememberMixin, ThreadAgent):
+    """Summarizes a job list into prose. Context: only the tool response it is
+    prompted with — no history, no profile, no tools."""
+
+    model = "gemini-3.1-flash-lite"  # match the graph's model, not the lab default
+
+    def instructions(self) -> str:
+        return SUMMARIZER_PROMPT
+
+    def middleware(self) -> list[Middleware]:
+        return [AgentLogger()]
+
+
+class JobSearchAgent(RememberMixin, ThreadAgent):
+    """Plain single-LLM agent (like ChatAgent) that owns its own context: the thread's
+    past user messages plus the user profile, injected only when the router asked for
+    it via `contexts`. messages() is async — the runner awaits it.
+    """
+
+    # The model must search, never chat: text-only replies here would be discarded
+    # by job_search_node (it only reads tool_events).
+    tool_choice = "any"
+
+    @property
+    def contexts(self) -> list[Context]:
+        return self.state.get("contexts") or []
+
+    async def messages(self) -> list[BaseMessage]:
+        messages: list[BaseMessage] = []
+
+        if Context.INCLUDE_USER_PROFILE in self.contexts:
+            messages.append(SystemMessage(content=f"User profile: {json.dumps(USER_PROFILE)}"))
+
+        rows = list(await Message.where("thread_id", self.thread_id).where("role", "user").order_by("id").get())
+        # The current turn's user row is persisted before the run; the runner appends
+        # the live query itself, so drop it from history to avoid sending it twice.
+        if rows and rows[-1].data.get("text") == self.state.get("query"):
+            rows = rows[:-1]
+        messages.extend(HumanMessage(content=row.data.get("text", "")) for row in rows)
+
+        if Context.INCLUDE_LAST_JOB_SEARCH_RESPONSE in self.contexts and (last := await self._last_job_search()):
+            messages.append(AIMessage(content=f"Previous job search results: {json.dumps(last.data)}"))
+
+        return messages
+
+    async def _last_job_search(self) -> Message | None:
+        return (
+            await Message.where("thread_id", self.thread_id)
+            .where("role", "ai")
+            .where("type", "tool_response")
+            .order_by("id", "desc")
+            .first()
+        )
+
+    def instructions(self) -> str:
+        return JOB_SEARCH_PROMPT
+
+    def middleware(self) -> list[Middleware]:
+        return [AgentLogger()]
+
+    def tools(self) -> list[BaseTool]:
+        return [job_search_tool]

example/agents/app/agents/chat.py

@@ -1,16 +1,15 @@
-from typing import Callable
-
 from fastapi_startkit.ai import Agent, Middleware
+from langchain_core.tools import BaseTool
 
 from app.middleware.agent_logger import AgentLogger
 from app.tools.job_search_tool import job_search_tool
 
 
-class RouterAgent(Agent):
+class ChatAgent(Agent):
     def middleware(self) -> list[Middleware]:
         return [AgentLogger()]
 
-    def tools(self) -> list[Callable]:
+    def tools(self) -> list[BaseTool]:
         return [job_search_tool]
 
     def instructions(self) -> str:

example/agents/app/agents/graph_agent.py

@@ -0,0 +1,30 @@
+from fastapi_startkit.ai import GraphAgent, GraphRunner, Middleware
+from fastapi_startkit.ai.graph import AgentState
+from langchain_core.tools import BaseTool
+from langgraph.graph import END, START, StateGraph
+from langgraph.types import Checkpointer
+
+from app.middleware.agent_logger import AgentLogger
+from app.tools.job_search_tool import job_search_tool
+
+
+class SalesAgent(GraphAgent):
+    def middleware(self) -> list[Middleware]:
+        return [AgentLogger()]
+
+    def tools(self) -> list[BaseTool]:
+        return [job_search_tool]
+
+    async def graph(self, runner: GraphRunner) -> StateGraph:
+        from fastapi_startkit.application import app
+
+        checkpointer: Checkpointer = await app().make("checkpointer")
+
+        return (
+            StateGraph(AgentState)
+            .add_node("llm", runner.llm)
+            .add_node("tools", runner.call_tools)
+            .add_edge(START, "llm")
+            .add_conditional_edges("llm", runner.route, ["tools", END])
+            .add_edge("tools", "llm")
+        ).compile(checkpointer=checkpointer)

example/agents/app/agents/job_search_graph.py

@@ -0,0 +1,143 @@
+import json
+from typing import cast
+
+from fastapi_startkit.ai import state as ai_state
+from langchain_core.runnables import RunnableConfig
+from langgraph.graph import END, START, StateGraph
+from langgraph.graph.state import CompiledStateGraph
+
+from app.agents.state import InputState, OutputState, OverallState, RouterOutput
+
+# Stand-in for a real profile lookup; the router decides when it's needed.
+USER_PROFILE = {
+    "name": "Alice",
+    "title": "Python Developer",
+    "location": "Remote",
+    "skills": ["Python", "FastAPI", "PostgreSQL"],
+}
+
+
+async def router_node(state: InputState, config: RunnableConfig) -> dict:
+    """Context: only the user input. No history, no profile, no tool output.
+
+    Handles greetings itself: for the 'chat' intent it answers inline via
+    RouterOutput.reply, so no extra node/LLM call is needed.
+    """
+    from app.agents.agent import JobSearchRouterAgent
+
+    response = await JobSearchRouterAgent(state, config).prompt(state["query"])
+    decision = cast(
+        RouterOutput, ai_state.structured(response) or RouterOutput.model_validate_json(ai_state.text(response))
+    )
+
+    out: dict = {"intent": decision.intent, "contexts": decision.contexts}
+    if decision.intent == "chat":
+        out |= {"type": "text", "data": decision.reply, "data_type": "string"}
+    return out
+
+
+def route(state: OverallState) -> str:
+    # chat is already answered inline by the router; company_research isn't built yet.
+    return "job_search" if state["intent"] == "job_search" else END
+
+
+# --- job search ------------------------------------------------------------
+
+
+async def job_search_node(state: OverallState, config: RunnableConfig) -> dict:
+    """Context: the thread's history + the user profile — both owned by JobSearchAgent.
+
+    Delegates the tool-calling loop to JobSearchAgent.prompt(), then pulls the raw job
+    rows out of the response's tool_events (the summarizer node turns them into prose).
+    Also reports the calls it made — {"name", "args"} — so they can be shown/persisted.
+    """
+    from app.agents.agent import JobSearchAgent
+
+    response = await JobSearchAgent(state, config).prompt(state["query"])
+
+    jobs: list[dict] = []
+    calls: list[dict] = []
+    for event in ai_state.tool_events(response):
+        if event["name"] == "job_search_tool":
+            jobs.extend(json.loads(event["content"]))
+            calls.append({"name": event["name"], "args": event["args"]})
+
+    return {"type": "tool_response", "data": jobs, "data_type": "json", "tool_calls": calls}
+
+
+# --- ask user (human-in-the-loop) ------------------------------------------
+
+
+def ask_user_node(state: OverallState) -> dict:
+    """No jobs found: pause the graph and ask the frontend to refine the search.
+
+    interrupt() freezes the run (state is persisted by the checkpointer) and
+    surfaces this payload to the caller. The graph resumes when the caller sends
+    Command(resume=<reply>), and interrupt() returns that reply here.
+    """
+    from langgraph.types import interrupt
+
+    reply = interrupt(
+        {
+            "type": "question",
+            "reason": "no_jobs",
+            "message": f"No jobs matched '{state['query']}'. Try different keywords, or send blank to stop.",
+        }
+    )
+
+    return {"query": str(reply).strip(), "data": None}
+
+
+def after_ask(state: OverallState) -> str:
+    return "job_search" if state["query"] else "job_summarizer"
+
+
+# --- job summarizer --------------------------------------------------------
+
+
+async def job_summarizer_node(state: OverallState, config: RunnableConfig) -> dict:
+    """Context: only the tool response. Never sees the query or the profile."""
+    if not state["data"] and state["query"]:
+        # ask_user still wants a refined query; don't spend a call summarizing nothing
+        return {}
+
+    from app.agents.agent import JobSummarizerAgent
+
+    response = await JobSummarizerAgent(state, config).prompt(json.dumps(state["data"] or []))
+
+    return {"type": "text", "data": ai_state.text(response), "data_type": "string"}
+
+
+def ask_or_finish(state: OverallState) -> str:
+    return "ask_user" if not state["data"] and state["query"] else END
+
+
+def build() -> StateGraph:
+    builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
+
+    builder.add_node("router", router_node)
+    builder.add_node("job_search", job_search_node)
+    builder.add_node("ask_user", ask_user_node)
+    builder.add_node("job_summarizer", job_summarizer_node)
+
+    builder.add_edge(START, "router")
+    builder.add_conditional_edges("router", route, ["job_search", END])
+    builder.add_edge("job_search", "job_summarizer")
+    builder.add_conditional_edges("job_summarizer", ask_or_finish, ["ask_user", END])
+    builder.add_conditional_edges("ask_user", after_ask, ["job_search", "job_summarizer"])
+
+    return builder
+
+
+_graph: CompiledStateGraph | None = None
+
+
+async def job_graph() -> CompiledStateGraph:
+    """Compile once, backed by the container's checkpointer (needed for interrupts)."""
+    global _graph
+    if _graph is None:
+        from fastapi_startkit.application import app
+
+        checkpointer = await app().make("checkpointer")
+        _graph = build().compile(checkpointer=checkpointer)
+    return _graph

example/agents/app/agents/remember.py

@@ -0,0 +1,60 @@
+import json
+
+from fastapi_startkit.ai import Agent
+from fastapi_startkit.ai import state as ai_state
+from langchain_core.runnables import RunnableConfig
+
+from app.agents.state import OverallState
+from app.models.message import Message
+from app.models.thread import Thread
+
+
+class ThreadAgent(Agent):
+    """An agent bound to a conversation thread via the graph's (state, config)."""
+
+    def __init__(self, state: OverallState | None = None, config: RunnableConfig | None = None):
+        super().__init__()
+        self.state: dict = dict(state or {})
+        configurable = (config or {}).get("configurable") or {}
+        self.thread_id: str = configurable.get("thread_id", "")
+        self.turn_id: str | None = configurable.get("turn_id")
+
+
+class RememberMixin:
+    """Persists what the agent did — tool calls, tool responses and text replies —
+    onto its thread, right after each prompt(). No thread_id -> remembers nothing.
+    """
+
+    async def prompt(self, message: str, **kwargs) -> dict:
+        state = await super().prompt(message, **kwargs)
+        await self.remember(state)
+        return state
+
+    async def remember(self, state: dict) -> None:
+        events = ai_state.tool_events(state)
+        for event in events:
+            await self.remember_row("tool_call", {"name": event["name"], "args": event["args"]}, state)
+            await self.remember_row("tool_response", json.loads(event["content"]), state)
+
+        # A plain text reply; structured output is state, not conversation.
+        text = ai_state.text(state)
+        if text and not events and ai_state.structured(state) is None:
+            await self.remember_row("text", {"text": text}, state)
+
+    async def remember_row(self, type_: str, data: dict | list, state: dict) -> None:
+        if not self.thread_id:  # not bound to a conversation -> nothing to remember
+            return
+        await Thread.first_or_create({"id": self.thread_id}, {"id": self.thread_id})
+        row: dict = {
+            "thread_id": self.thread_id,
+            "role": "ai",
+            "type": type_,
+            "data": data,
+            "data_type": "string" if type_ == "text" else "json",
+            "run_id": self.turn_id,
+            "meta": {"agent": type(self).__name__, **({"model": self.model} if self.model else {})},
+        }
+        usage = ai_state.usage(state)
+        if usage["input"] or usage["output"]:
+            row["usage"] = usage
+        await Message.create(row)