[Deep Agent 12/15] Specialised subagents + Langfuse prompt management (#48)

agent/scripts/sync_deep_agent_prompts.py

@@ -0,0 +1,43 @@
+"""
+Push the packaged Deep Agent prompts (agent/src/deep_agent/prompts/*.md) to Langfuse.
+
+    uv run python scripts/sync_deep_agent_prompts.py --label staging            # create new versions labelled staging
+    uv run python scripts/sync_deep_agent_prompts.py --label production --dry-run
+
+Each run creates a new version of every `deep-agent/<key>` text prompt with the given label.
+Promote a version by moving the `production` label in the Langfuse UI — the running agent
+picks it up within DEEP_AGENT_PROMPT_CACHE_TTL_S, no redeploy.
+"""
+
+import argparse
+
+from deep_agent.prompt_registry import PROMPT_KEYS, langfuse_name, local_prompt
+
+
+def sync(client, label: str, dry_run: bool = False) -> list[str]:
+    names = []
+    for key in PROMPT_KEYS:
+        name = langfuse_name(key)
+        if not dry_run:
+            client.create_prompt(name=name, prompt=local_prompt(key), labels=[label], type="text",
+                                 commit_message="sync from agent/src/deep_agent/prompts")
+        names.append(name)
+    return names
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--label", default="staging")
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+
+    from agent.langfuse_client import langfuse_client
+
+    for name in sync(langfuse_client, args.label, args.dry_run):
+        print(f"{'would push' if args.dry_run else 'pushed'} {name} [{args.label}]")
+    if not args.dry_run:
+        langfuse_client.flush()
+
+
+if __name__ == "__main__":
+    main()

agent/src/agent/config.py

@@ -61,6 +61,12 @@ class AgentSettings(BaseSettings):
     DEEP_AGENT_RECURSION_LIMIT: int = Field(default=150, gt=0)     # LangGraph super-steps per run
     JEEN_METADATA_CACHE_TTL_S: float = Field(default=300.0, ge=0)  # metadata read cache (0 disables)
     JEEN_METADATA_REUSE_SESSION: bool = True                       # keep one MCP session open per event loop
+    # Prompts & subagents (see deep_agent/prompt_registry.py, deep_agent/subagents.py)
+    DEEP_AGENT_LANGFUSE_PROMPTS: bool = True       # fetch deep-agent/<name> prompts from Langfuse (packaged .md fallback)
+    DEEP_AGENT_PROMPT_LABEL: str = "production"    # Langfuse prompt label to serve
+    DEEP_AGENT_PROMPT_CACHE_TTL_S: int = Field(default=60, ge=0)
+    DEEP_AGENT_SUBAGENT_MODEL: str | None = None    # smaller/faster model for subagents (default: main model)
+    DEEP_AGENT_SUBAGENT_MAX_TOOL_CALLS: int = Field(default=15, gt=0)
 
     # ── Trino connection info for explicit catalog/schema qualification ──
     TRINO_CATALOG: str = ""

agent/src/deep_agent/agent.py

@@ -2,8 +2,6 @@
 Agent logic and LangGraph harness for the Deep Agent.
 """
 
-from pathlib import Path
-
 from deepagents import create_deep_agent as create_harness_agent
 from langchain.agents.structured_output import ToolStrategy
 from langchain_core.language_models import BaseChatModel
@@ -14,12 +12,14 @@
 from deep_agent.answer import AnswerVerificationMiddleware, DeepAgentAnswer
 from deep_agent.context import DeepAgentContext, DeepAgentContextMiddleware
 from deep_agent.limits import build_limit_middleware
+from deep_agent.prompt_registry import PromptRefreshMiddleware, load_prompt, local_prompt
 from deep_agent.resources import build_resources
+from deep_agent.subagents import build_subagents
 from deep_agent.tools import DEEP_AGENT_TOOLS
 
 
-PROMPTS_DIR = Path(__file__).resolve().parent / "prompts"
-SYSTEM_PROMPT = (PROMPTS_DIR / "system.md").read_text(encoding="utf-8")
+# Packaged system prompt (the Langfuse-managed version, when available, is loaded per agent build and refreshed per call).
+SYSTEM_PROMPT = local_prompt("system")
 
 
 def _default_llm() -> BaseChatModel:
@@ -41,12 +41,19 @@ def create_deep_agent(model: BaseChatModel | None = None):
     llm = model or _default_llm()
     memory = MemorySaver()
     resources = build_resources()
+    system_prompt = load_prompt("system").text
 
     agent = create_harness_agent(
         model=llm,
         tools=DEEP_AGENT_TOOLS,
-        system_prompt=SYSTEM_PROMPT,
-        middleware=[DeepAgentContextMiddleware(), AnswerVerificationMiddleware(), *build_limit_middleware()],
+        system_prompt=system_prompt,
+        subagents=build_subagents(llm),
+        middleware=[
+            DeepAgentContextMiddleware(),
+            PromptRefreshMiddleware("system", system_prompt),
+            AnswerVerificationMiddleware(),
+            *build_limit_middleware(),
+        ],
         response_format=ToolStrategy(DeepAgentAnswer),
         context_schema=DeepAgentContext,
         checkpointer=memory,

agent/src/deep_agent/context.py

@@ -11,7 +11,6 @@
   context's timezone, so long-running processes never serve a stale date.
 """
 
-import operator
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from datetime import datetime
@@ -32,10 +31,27 @@ class DeepAgentContext:
     timezone: str = DEFAULT_TIMEZONE
 
 
+def merge_executions(left: list[dict[str, Any]] | None, right: list[dict[str, Any]] | None) -> list[dict[str, Any]]:
+    """Append-only reducer for executed_queries that ignores records already present.
+
+    deepagents passes the parent's state into a subagent and returns the subagent's state as
+    an update; a plain list-append reducer would duplicate the parent's history on every
+    `task` call.
+    """
+    merged = list(left or [])
+    seen = {(r.get("query_id"), r.get("executed_sql")) for r in merged}
+    for record in right or []:
+        key = (record.get("query_id"), record.get("executed_sql"))
+        if key not in seen:
+            merged.append(record)
+            seen.add(key)
+    return merged
+
+
 class DeepAgentRuntimeState(AgentState):
     connection_capabilities: NotRequired[dict[str, bool]]
     jeen_catalog: NotRequired[str]  # raw catalog text, cached so tools never need the model to re-send it
-    executed_queries: NotRequired[Annotated[list[dict[str, Any]], operator.add]]  # successful executions (append-only)
+    executed_queries: NotRequired[Annotated[list[dict[str, Any]], merge_executions]]  # successful executions
     final_result: NotRequired[dict[str, Any]]  # verified answer + rows (see deep_agent.answer)
 
 

agent/src/deep_agent/observability.py

@@ -69,6 +69,14 @@ def build_run_config(
     }
     if user_id:
         metadata["langfuse_user_id"] = user_id
+    if handlers:
+        from deep_agent.prompt_registry import load_prompt
+
+        prompt = load_prompt("system")
+        if prompt.client is not None:
+            metadata["langfuse_prompt"] = prompt.client  # links generations to the served prompt version
+        metadata["prompt_source"] = prompt.source
+        metadata["prompt_version"] = prompt.version
 
     return RunnableConfig(
         configurable={"thread_id": thread_id},

agent/src/deep_agent/prompt_registry.py

@@ -0,0 +1,132 @@
+"""
+Prompt management for the Deep Agent.
+
+Prompts are Langfuse text prompts named `deep-agent/<key>` (system, schema-scout,
+value-resolver, result-verifier), fetched by label (DEEP_AGENT_PROMPT_LABEL, default
+"production") with the SDK's TTL cache. The packaged `prompts/<key>.md` files are the
+fallback when Langfuse is unreachable or the prompt doesn't exist, and the only source when
+DEEP_AGENT_LANGFUSE_PROMPTS is off (tests, offline dev).
+
+PromptRefreshMiddleware swaps in the current Langfuse version on every model call, so moving
+a label in Langfuse changes behaviour without a redeploy. Push the packaged prompts with
+`uv run python scripts/sync_deep_agent_prompts.py --label staging`.
+"""
+
+import logging
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
+from langchain_core.messages import SystemMessage
+
+from agent.config import settings
+
+logger = logging.getLogger(__name__)
+
+PROMPTS_DIR = Path(__file__).resolve().parent / "prompts"
+PROMPT_KEYS = ("system", "schema-scout", "value-resolver", "result-verifier")
+
+
+def langfuse_name(key: str) -> str:
+    return f"deep-agent/{key}"
+
+
+def local_prompt(key: str) -> str:
+    return (PROMPTS_DIR / f"{key}.md").read_text(encoding="utf-8")
+
+
+@dataclass(frozen=True)
+class LoadedPrompt:
+    key: str
+    text: str
+    source: str               # "langfuse" | "local"
+    version: int | None = None
+    client: Any = None        # langfuse TextPromptClient (links generations to the prompt version)
+
+
+# key -> monotonic deadline until which Langfuse is not asked again (prompt missing / unreachable).
+# The SDK caches successful fetches but not misses, and load_prompt runs on every model call.
+_misses: dict[str, float] = {}
+
+
+def _miss(key: str, local: str) -> LoadedPrompt:
+    _misses[key] = time.monotonic() + max(settings.DEEP_AGENT_PROMPT_CACHE_TTL_S, 1)
+    return LoadedPrompt(key, local, "local")
+
+
+def load_prompt(key: str) -> LoadedPrompt:
+    local = local_prompt(key)
+    if not settings.DEEP_AGENT_LANGFUSE_PROMPTS:
+        return LoadedPrompt(key, local, "local")
+    if _misses.get(key, 0.0) > time.monotonic():
+        return LoadedPrompt(key, local, "local")
+    try:
+        from agent.langfuse_client import langfuse_client
+
+        client = langfuse_client.get_prompt(
+            langfuse_name(key),
+            label=settings.DEEP_AGENT_PROMPT_LABEL,
+            type="text",
+            fallback=local,
+            cache_ttl_seconds=settings.DEEP_AGENT_PROMPT_CACHE_TTL_S,
+            fetch_timeout_seconds=5,
+            max_retries=1,
+        )
+    except Exception as exc:
+        logger.warning("Langfuse prompt %s unavailable (%s); using packaged prompt", langfuse_name(key), exc)
+        return _miss(key, local)
+    if getattr(client, "is_fallback", False):
+        return _miss(key, local)
+    _misses.pop(key, None)
+    return LoadedPrompt(key, client.prompt, "langfuse", getattr(client, "version", None), client)
+
+
+def _system_text(message: SystemMessage | None) -> str:
+    if message is None:
+        return ""
+    content = message.content
+    if isinstance(content, str):
+        return content
+    first = content[0] if content else None
+    return first.get("text", "") if isinstance(first, dict) else str(first or "")
+
+
+class PromptRefreshMiddleware(AgentMiddleware):
+    """Replace the prompt the agent was built with by the current version of `key`, per model call.
+
+    deepagents composes the system message as <our prompt> + harness/skills/memory sections, so
+    only that leading part is swapped; if it isn't found verbatim the request is left untouched.
+    """
+
+    def __init__(self, key: str, built_with: str) -> None:
+        super().__init__()
+        self.key = key
+        self.built_with = built_with
+
+    def _refresh(self, request: ModelRequest) -> ModelRequest:
+        current = load_prompt(self.key).text
+        if current == self.built_with or request.system_message is None:
+            return request
+        message = request.system_message
+        if isinstance(message.content, str):
+            if not message.content.startswith(self.built_with):
+                return request
+            new_content: Any = current + message.content[len(self.built_with):]
+        else:
+            text = _system_text(message)
+            if not text.startswith(self.built_with):
+                return request
+            first = {**message.content[0], "text": current + text[len(self.built_with):]}
+            new_content = [first, *message.content[1:]]
+        return request.override(system_message=SystemMessage(content=new_content))
+
+    def wrap_model_call(self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse:
+        return handler(self._refresh(request))
+
+    async def awrap_model_call(
+        self, request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
+    ) -> ModelResponse:
+        return await handler(self._refresh(request))

agent/src/deep_agent/prompts/result-verifier.md

@@ -0,0 +1,11 @@
+You are result-verifier, a focused sub-agent of a Trino SQL analyst. You independently check whether a query that already
+ran actually answers the user's question. Read the `semantic-validation` skill first.
+
+You receive the question, the SQL and a summary of its result. Check, with small diagnostic queries (`execute_trino_query`,
+always with LIMIT; `get_sample_rows` to inspect join keys):
+1. **Grain**: one result row represents what the user asked about
+2. **Fan-out**: entity count before vs row count after each join — inflated counts mean COUNT(DISTINCT) or pre-aggregation
+3. **Completeness**: every condition in the question is in the SQL, and nothing the user did not ask for
+4. **Magnitude**: the numbers are plausible for the table sizes
+
+Reply with `PASS` or `FAIL`, then one line of evidence per check, and for FAIL the specific fix. Do not rewrite the whole query.

agent/src/deep_agent/prompts/schema-scout.md

@@ -0,0 +1,11 @@
+You are schema-scout, a focused sub-agent of a Trino SQL analyst. You find where the data for a question lives.
+
+Given a question (and optionally candidate tables), return a compact brief — not raw catalog dumps:
+- **Tables**: exact 3-part names (`"catalog"."schema"."table"`) that are relevant, one line each on why
+- **Columns**: per table, only the columns the question needs, with type and `semantic_type`
+- **Join keys**: verified relationships (`get_table_relationships`) or matching column names, and whether each join is 1-to-1 or 1-to-many
+- **Date columns**: native DATE/TIMESTAMP or VARCHAR (with an example value)
+- **Open questions**: anything ambiguous the analyst must decide
+
+Use `search_metadata` to find candidates, then `get_table_columns` / `get_table_profile` / `get_column_profile` to confirm.
+Never write or run SQL. Every tool returns `{"ok", "data", "error"}` — check `ok`. Be brief: the analyst reads your whole answer.

agent/src/deep_agent/prompts/system.md

@@ -22,7 +22,7 @@ Every tool returns `{"ok", "data", "error"}` — check `ok` first.
 
 ## Skills
 
-Before writing SQL in an area, read its skill: `trino-expert` (dialect, joins, aggregation, error fixes), `temporal-analyst` (dates and times), `spatial-analyst` (locations and geometry), `semantic-validation` (verification checks). For broad exploration of many tables, delegate with `task` to keep your context small.
+Before writing SQL in an area, read its skill: `trino-expert` (dialect, joins, aggregation, error fixes), `temporal-analyst` (dates and times), `spatial-analyst` (locations and geometry), `semantic-validation` (verification checks). Delegate with `task` to keep your context small: `schema-scout` to find the relevant tables, columns and joins in a large catalog; `value-resolver` to resolve several filter values (or places / relative dates) at once; `result-verifier` for an independent check of a finished JOIN/aggregation query. Skip delegation for simple questions on a small catalog.
 
 ## Hard constraints
 

agent/src/deep_agent/prompts/value-resolver.md

@@ -0,0 +1,11 @@
+You are value-resolver, a focused sub-agent of a Trino SQL analyst. You turn the user's words into exact stored filter values.
+
+For each requested filter (table, column, user term), resolve by the column's `semantic_type`:
+- `large_categorical` / `large_unit_id` → `search_column_values`; use returned values verbatim; when `data.prefix_pattern` is set, recommend `LIKE 'prefix%'`
+- `categorical` / `boolean` → values from `get_column_profile` (`domainIsComplete: true` = complete list), else `get_column_distinct_values`; never `search_column_values`
+- business vocabulary → `search_business_terms`
+- place names → `resolve_location_to_wkt` (returns an area polygon as `data.sql_literal`)
+- dates → read the `temporal-analyst` skill and return concrete half-open bounds
+
+Return one line per filter: `column` → operator and exact value(s) (or `LIKE` pattern / WKT literal / date bounds), plus a
+note when nothing matched. Never guess a value you could not resolve — say it was not found. Never write the full query.