From 949f3a62d511ead0b978bf76c645dc55a357dee8 Mon Sep 17 00:00:00 2001 From: Stav Ponte Date: Tue, 20 Oct 2026 17:21:49 +0300 Subject: [PATCH 1/2] feat(deep-agent): specialised subagents and Langfuse-managed prompts - subagents.py: schema-scout, value-resolver, result-verifier with explicit tool allow-lists, own prompts, skills, context/date middleware, tool-call cap and optional smaller model (DEEP_AGENT_SUBAGENT_MODEL) - prompt_registry.py: deep-agent/ text prompts fetched from Langfuse by label (DEEP_AGENT_PROMPT_LABEL) with packaged .md fallback; PromptRefreshMiddleware serves label changes without a redeploy - run config links generations to the served prompt version (langfuse_prompt metadata) and records prompt_source/prompt_version - executed_queries reducer de-duplicates records (subagent state round trips re-sent the parent's history) - scripts/sync_deep_agent_prompts.py pushes packaged prompts to Langfuse Closes #48 Co-Authored-By: Claude Opus 5.5 --- agent/scripts/sync_deep_agent_prompts.py | 43 ++++ agent/src/agent/config.py | 6 + agent/src/deep_agent/agent.py | 19 +- agent/src/deep_agent/context.py | 20 +- agent/src/deep_agent/observability.py | 8 + agent/src/deep_agent/prompt_registry.py | 118 +++++++++++ .../src/deep_agent/prompts/result-verifier.md | 11 + agent/src/deep_agent/prompts/schema-scout.md | 11 + agent/src/deep_agent/prompts/system.md | 2 +- .../src/deep_agent/prompts/value-resolver.md | 11 + agent/src/deep_agent/subagents.py | 75 +++++++ agent/tests/deep_agent/conftest.py | 8 + .../deep_agent/test_subagents_prompts.py | 191 ++++++++++++++++++ 13 files changed, 514 insertions(+), 9 deletions(-) create mode 100644 agent/scripts/sync_deep_agent_prompts.py create mode 100644 agent/src/deep_agent/prompt_registry.py create mode 100644 agent/src/deep_agent/prompts/result-verifier.md create mode 100644 agent/src/deep_agent/prompts/schema-scout.md create mode 100644 agent/src/deep_agent/prompts/value-resolver.md create mode 100644 agent/src/deep_agent/subagents.py create mode 100644 agent/tests/deep_agent/test_subagents_prompts.py diff --git a/agent/scripts/sync_deep_agent_prompts.py b/agent/scripts/sync_deep_agent_prompts.py new file mode 100644 index 0000000..33d051a --- /dev/null +++ b/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/` 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() diff --git a/agent/src/agent/config.py b/agent/src/agent/config.py index e4b9e56..d5bdd10 100644 --- a/agent/src/agent/config.py +++ b/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/ 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 = "" diff --git a/agent/src/deep_agent/agent.py b/agent/src/deep_agent/agent.py index 120dd6d..45a6e8d 100644 --- a/agent/src/deep_agent/agent.py +++ b/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, diff --git a/agent/src/deep_agent/context.py b/agent/src/deep_agent/context.py index c075e64..143d241 100644 --- a/agent/src/deep_agent/context.py +++ b/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) diff --git a/agent/src/deep_agent/observability.py b/agent/src/deep_agent/observability.py index 43acc9c..8874fd0 100644 --- a/agent/src/deep_agent/observability.py +++ b/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}, diff --git a/agent/src/deep_agent/prompt_registry.py b/agent/src/deep_agent/prompt_registry.py new file mode 100644 index 0000000..f72f132 --- /dev/null +++ b/agent/src/deep_agent/prompt_registry.py @@ -0,0 +1,118 @@ +""" +Prompt management for the Deep Agent. + +Prompts are Langfuse text prompts named `deep-agent/` (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/.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 +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) + + +def load_prompt(key: str) -> LoadedPrompt: + local = local_prompt(key) + if not settings.DEEP_AGENT_LANGFUSE_PROMPTS: + 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 LoadedPrompt(key, local, "local") + if getattr(client, "is_fallback", False): + return LoadedPrompt(key, local, "local") + 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 + 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)) diff --git a/agent/src/deep_agent/prompts/result-verifier.md b/agent/src/deep_agent/prompts/result-verifier.md new file mode 100644 index 0000000..27a8216 --- /dev/null +++ b/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. diff --git a/agent/src/deep_agent/prompts/schema-scout.md b/agent/src/deep_agent/prompts/schema-scout.md new file mode 100644 index 0000000..932df4b --- /dev/null +++ b/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. diff --git a/agent/src/deep_agent/prompts/system.md b/agent/src/deep_agent/prompts/system.md index b620d9e..1577da4 100644 --- a/agent/src/deep_agent/prompts/system.md +++ b/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 diff --git a/agent/src/deep_agent/prompts/value-resolver.md b/agent/src/deep_agent/prompts/value-resolver.md new file mode 100644 index 0000000..8dddafa --- /dev/null +++ b/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. diff --git a/agent/src/deep_agent/subagents.py b/agent/src/deep_agent/subagents.py new file mode 100644 index 0000000..981d3df --- /dev/null +++ b/agent/src/deep_agent/subagents.py @@ -0,0 +1,75 @@ +""" +Specialised subagents the Deep Agent delegates to with the `task` tool. + +Each has a narrow tool allow-list (fewer schemas per call, no off-task tools), its own +Langfuse-managed prompt, the request-context middleware (connection, date), a tool-call cap, +and optionally a smaller model (DEEP_AGENT_SUBAGENT_MODEL). They return compact briefs, so +the main agent's context holds conclusions instead of raw catalog / profile dumps. +""" + +from langchain.agents.middleware import ToolCallLimitMiddleware +from langchain_core.language_models import BaseChatModel + +from agent.config import settings +from deep_agent import tools as t +from deep_agent.context import DeepAgentContextMiddleware +from deep_agent.limits import BuiltinToolFilterMiddleware +from deep_agent.prompt_registry import PromptRefreshMiddleware, load_prompt +from deep_agent.resources import SKILLS_ROUTE + +SUBAGENT_TOOLS = { + "schema-scout": [ + t.search_metadata, t.get_database_tables, t.get_table_columns, t.get_table_profile, + t.get_column_profile, t.get_table_relationships, + ], + "value-resolver": [ + t.search_column_values, t.get_column_distinct_values, t.get_column_profile, t.search_business_terms, + t.resolve_location_to_wkt, t.get_sample_rows, + ], + "result-verifier": [t.execute_trino_query, t.get_sample_rows], +} + +SUBAGENT_DESCRIPTIONS = { + "schema-scout": ( + "Finds the tables, columns, join keys and date-column types relevant to a question across a large catalog. " + "Give it the user's question (and candidate tables if known). Returns a compact brief, never raw dumps." + ), + "value-resolver": ( + "Resolves several filter values at once to exact stored values (by semantic type), place names to WKT " + "literals and relative dates to concrete bounds. Give it the question and the list of (table, column, term)." + ), + "result-verifier": ( + "Independently checks a finished JOIN/aggregation query for grain, fan-out, completeness and magnitude. " + "Give it the question, the SQL and a summary of the result. Returns PASS/FAIL with evidence." + ), +} + + +def _subagent_model(main_model: BaseChatModel) -> BaseChatModel | None: + if not settings.DEEP_AGENT_SUBAGENT_MODEL: + return None # inherit the main agent's model + return main_model.model_copy(update={"model_name": settings.DEEP_AGENT_SUBAGENT_MODEL}) + + +def build_subagents(main_model: BaseChatModel) -> list[dict]: + model = _subagent_model(main_model) + subagents = [] + for name, tools in SUBAGENT_TOOLS.items(): + prompt = load_prompt(name).text + spec = { + "name": name, + "description": SUBAGENT_DESCRIPTIONS[name], + "system_prompt": prompt, + "tools": tools, + "skills": [SKILLS_ROUTE], + "middleware": [ + DeepAgentContextMiddleware(), + PromptRefreshMiddleware(name, prompt), + ToolCallLimitMiddleware(run_limit=settings.DEEP_AGENT_SUBAGENT_MAX_TOOL_CALLS), + BuiltinToolFilterMiddleware(), + ], + } + if model is not None: + spec["model"] = model + subagents.append(spec) + return subagents diff --git a/agent/tests/deep_agent/conftest.py b/agent/tests/deep_agent/conftest.py index 86a9726..b670e11 100644 --- a/agent/tests/deep_agent/conftest.py +++ b/agent/tests/deep_agent/conftest.py @@ -58,3 +58,11 @@ def _make(policy: Callable[[list[BaseMessage]], AIMessage]) -> PolicyChatModel: return model return _make + + +@pytest.fixture(autouse=True) +def _packaged_prompts(monkeypatch): + """Unit tests use the packaged prompts; Langfuse prompt fetching is tested explicitly.""" + from agent.config import settings + + monkeypatch.setattr(settings, "DEEP_AGENT_LANGFUSE_PROMPTS", False) diff --git a/agent/tests/deep_agent/test_subagents_prompts.py b/agent/tests/deep_agent/test_subagents_prompts.py new file mode 100644 index 0000000..6bfee5d --- /dev/null +++ b/agent/tests/deep_agent/test_subagents_prompts.py @@ -0,0 +1,191 @@ +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from langchain_core.messages import AIMessage + +from agent.config import settings +from deep_agent import observability, prompt_registry, tools +from deep_agent.agent import create_deep_agent +from deep_agent.context import DeepAgentContext, merge_executions +from deep_agent.prompt_registry import PROMPT_KEYS, load_prompt, local_prompt +from deep_agent.subagents import SUBAGENT_TOOLS, build_subagents +from tests.deep_agent.conftest import system_text + +HARNESS = {"read_file", "ls", "write_todos"} + + +def _role(messages) -> str: + text = system_text(messages) + for name in SUBAGENT_TOOLS: + if f"You are {name}" in text: + return name + return "main" + + +def _delegating_policy(subagent: str, subagent_call: dict): + """Main agent delegates once to `subagent`, which makes one tool call and reports back.""" + def policy(messages): + role = _role(messages) + tool_msgs = [m for m in messages if m.type == "tool"] + if role == "main": + if not tool_msgs: + return AIMessage("", tool_calls=[{"name": "task", "id": "t1", "args": { + "subagent_type": subagent, "description": "resolve 'צרפת' in flights_table.launch_country"}}]) + return AIMessage("done") + if not tool_msgs: + return AIMessage("", tool_calls=[{**subagent_call, "id": "s1"}]) + return AIMessage("launch_country = 'France'") + return policy + + +@pytest.mark.asyncio +async def test_subagent_gets_only_its_tools_context_and_date(policy_model): + model = policy_model(_delegating_policy( + "value-resolver", {"name": "search_column_values", "args": {"search_term": "צרפת", "column": "launch_country"}})) + client = MagicMock(search_column_values=AsyncMock(return_value={"values": [{"value": "France"}]})) + agent = create_deep_agent(model=model) + with patch.object(tools, "get_jeen_metadata_client", return_value=client): + result = await agent.ainvoke({"messages": [("user", "כמה טיסות מצרפת")]}, + {"configurable": {"thread_id": "sub"}}, context=DeepAgentContext(connection_id=7)) + + # The subagent's model call offered exactly its allow-list (+ harness file/todo tools). + sub_calls = [i for i, msgs in enumerate(model.seen) if _role(msgs) == "value-resolver"] + offered = set(model.bound_tool_names[sub_calls[0]]) + assert offered - HARNESS == {tool.name for tool in SUBAGENT_TOOLS["value-resolver"]} + assert "execute_trino_query" not in offered and "task" not in offered + + # Request context propagated into the subagent's tools; date middleware ran there too. + assert client.search_column_values.await_args.kwargs["connection_id"] == 7 + first_human = next(m for m in model.seen[sub_calls[0]] if m.type == "human") + assert first_human.text.startswith("[Today's date:") + + task_result = next(m for m in result["messages"] if m.type == "tool" and m.name == "task") + assert "France" in task_result.text + + +@pytest.mark.asyncio +async def test_main_agent_task_tool_advertises_the_subagents(scripted_model): + model = scripted_model(AIMessage("done")) + await create_deep_agent(model=model).ainvoke({"messages": [("user", "q")]}, {"configurable": {"thread_id": "adv"}}) + assert "task" in model.bound_tool_names[0] + prompt = system_text(model.seen[0]) + for name in SUBAGENT_TOOLS: + assert f"`{name}`" in prompt + + +def test_subagent_specs(): + specs = {s["name"]: s for s in build_subagents(MagicMock())} + assert set(specs) == {"schema-scout", "value-resolver", "result-verifier"} + assert [t.name for t in specs["result-verifier"]["tools"]] == ["execute_trino_query", "get_sample_rows"] + assert all("model" not in s for s in specs.values()) # inherit main model by default + assert all(s["system_prompt"] == local_prompt(s["name"]) for s in specs.values()) + + +def test_subagent_model_override(monkeypatch): + from deep_agent.agent import _default_llm + + monkeypatch.setattr(settings, "DEEP_AGENT_SUBAGENT_MODEL", "small-model") + specs = build_subagents(_default_llm()) + assert {s["model"].model_name for s in specs} == {"small-model"} + + +def test_merge_executions_is_idempotent_for_subagent_round_trips(): + a = {"query_id": "q1", "executed_sql": "SELECT 1"} + b = {"query_id": "q2", "executed_sql": "SELECT 2"} + # Subagent returns parent's history plus one new record. + assert merge_executions([a], [a, b]) == [a, b] + assert merge_executions(None, [a, a]) == [a] + + +# ── prompt registry ────────────────────────────────────────────────────────── + +def test_every_prompt_has_a_packaged_fallback(): + for key in PROMPT_KEYS: + assert local_prompt(key).strip() + + +def test_disabled_langfuse_uses_packaged_prompt(): + p = load_prompt("system") + assert p.source == "local" and p.text == local_prompt("system") and p.client is None + + +@pytest.fixture +def langfuse_enabled(monkeypatch): + monkeypatch.setattr(settings, "DEEP_AGENT_LANGFUSE_PROMPTS", True) + + +def _served(text, version=3, fallback=False): + return SimpleNamespace(prompt=text, version=version, is_fallback=fallback) + + +def test_langfuse_prompt_by_label(langfuse_enabled, monkeypatch): + monkeypatch.setattr(settings, "DEEP_AGENT_PROMPT_LABEL", "staging") + fake = MagicMock(get_prompt=MagicMock(return_value=_served("served v3"))) + with patch("agent.langfuse_client.langfuse_client", fake): + p = load_prompt("value-resolver") + assert (p.source, p.text, p.version) == ("langfuse", "served v3", 3) + args, kwargs = fake.get_prompt.call_args + assert args == ("deep-agent/value-resolver",) + assert kwargs["label"] == "staging" and kwargs["fallback"] == local_prompt("value-resolver") + + +@pytest.mark.parametrize("behaviour", ["fallback", "raises"]) +def test_langfuse_unavailable_falls_back(langfuse_enabled, behaviour): + fake = MagicMock() + if behaviour == "fallback": + fake.get_prompt.return_value = _served(local_prompt("system"), version=None, fallback=True) + else: + fake.get_prompt.side_effect = RuntimeError("404 prompt not found") + with patch("agent.langfuse_client.langfuse_client", fake): + p = load_prompt("system") + assert p.source == "local" and p.text == local_prompt("system") + + +@pytest.mark.asyncio +async def test_prompt_label_change_takes_effect_without_rebuild(scripted_model, langfuse_enabled): + served = {"text": local_prompt("system")} + fake = MagicMock(get_prompt=MagicMock(side_effect=lambda *a, **k: _served(served["text"]))) + model = scripted_model(AIMessage("one"), AIMessage("two")) + with patch("agent.langfuse_client.langfuse_client", fake): + agent = create_deep_agent(model=model) + await agent.ainvoke({"messages": [("user", "a")]}, {"configurable": {"thread_id": "p1"}}) + served["text"] = "You are the NEW deep analyst prompt v4." + await agent.ainvoke({"messages": [("user", "b")]}, {"configurable": {"thread_id": "p2"}}) + + first, second = system_text(model.seen[0]), system_text(model.seen[1]) + assert first.startswith(local_prompt("system")[:80]) + assert second.startswith("You are the NEW deep analyst prompt v4.") + assert "Skills System" in second # harness / skills sections after our prompt are preserved + assert local_prompt("system")[:80] not in second + + +def test_run_config_links_generations_to_the_prompt_version(langfuse_enabled): + client = _served("x", version=7) + with patch.object(observability, "_langfuse_handler", return_value=object()), \ + patch.object(prompt_registry, "load_prompt", + return_value=prompt_registry.LoadedPrompt("system", "x", "langfuse", 7, client)): + config = observability.build_run_config("t", tracing=True) + assert config["metadata"]["langfuse_prompt"] is client + assert config["metadata"]["prompt_version"] == 7 + assert config["metadata"]["prompt_source"] == "langfuse" + + +def test_sync_script_pushes_every_prompt_with_label(): + import importlib.util + from pathlib import Path + + path = Path(__file__).resolve().parents[2] / "scripts" / "sync_deep_agent_prompts.py" + spec = importlib.util.spec_from_file_location("sync_prompts", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + fake = MagicMock() + pushed = module.sync(fake, label="staging", dry_run=False) + assert pushed == [f"deep-agent/{k}" for k in PROMPT_KEYS] + for call, key in zip(fake.create_prompt.call_args_list, PROMPT_KEYS): + assert call.kwargs["name"] == f"deep-agent/{key}" + assert call.kwargs["prompt"] == local_prompt(key) + assert call.kwargs["labels"] == ["staging"] and call.kwargs["type"] == "text" + assert module.sync(MagicMock(), label="staging", dry_run=True) == pushed + json.dumps(pushed) From adcdce9f6cf7818445301e38886e09d8da635565 Mon Sep 17 00:00:00 2001 From: Stav Ponte Date: Tue, 20 Oct 2026 17:39:41 +0300 Subject: [PATCH 2/2] fix(deep-agent): cache Langfuse prompt misses so a missing prompt isn't refetched per model call The Langfuse SDK caches fetched prompts but not misses; PromptRefreshMiddleware calls load_prompt on every model call, so an absent prompt (404 -> fallback) cost a network round trip on each call. Misses are now remembered for DEEP_AGENT_PROMPT_CACHE_TTL_S. Co-Authored-By: Claude Opus 5.5 --- agent/src/deep_agent/prompt_registry.py | 18 +++++++++++++++-- .../deep_agent/test_subagents_prompts.py | 20 +++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/agent/src/deep_agent/prompt_registry.py b/agent/src/deep_agent/prompt_registry.py index f72f132..f60cc8f 100644 --- a/agent/src/deep_agent/prompt_registry.py +++ b/agent/src/deep_agent/prompt_registry.py @@ -13,6 +13,7 @@ """ import logging +import time from collections.abc import Awaitable, Callable from dataclasses import dataclass from pathlib import Path @@ -46,10 +47,22 @@ class LoadedPrompt: 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 @@ -64,9 +77,10 @@ def load_prompt(key: str) -> LoadedPrompt: ) except Exception as exc: logger.warning("Langfuse prompt %s unavailable (%s); using packaged prompt", langfuse_name(key), exc) - return LoadedPrompt(key, local, "local") + return _miss(key, local) if getattr(client, "is_fallback", False): - return LoadedPrompt(key, local, "local") + return _miss(key, local) + _misses.pop(key, None) return LoadedPrompt(key, client.prompt, "langfuse", getattr(client, "version", None), client) diff --git a/agent/tests/deep_agent/test_subagents_prompts.py b/agent/tests/deep_agent/test_subagents_prompts.py index 6bfee5d..ce649a3 100644 --- a/agent/tests/deep_agent/test_subagents_prompts.py +++ b/agent/tests/deep_agent/test_subagents_prompts.py @@ -114,6 +114,9 @@ def test_disabled_langfuse_uses_packaged_prompt(): @pytest.fixture def langfuse_enabled(monkeypatch): monkeypatch.setattr(settings, "DEEP_AGENT_LANGFUSE_PROMPTS", True) + prompt_registry._misses.clear() + yield + prompt_registry._misses.clear() def _served(text, version=3, fallback=False): @@ -189,3 +192,20 @@ def test_sync_script_pushes_every_prompt_with_label(): assert call.kwargs["labels"] == ["staging"] and call.kwargs["type"] == "text" assert module.sync(MagicMock(), label="staging", dry_run=True) == pushed json.dumps(pushed) + + +def test_missing_prompt_is_not_refetched_on_every_call(langfuse_enabled): + """A 404 / fallback must not cost a Langfuse round trip on every model call.""" + fake = MagicMock(get_prompt=MagicMock(return_value=_served(local_prompt("system"), version=None, fallback=True))) + with patch("agent.langfuse_client.langfuse_client", fake): + for _ in range(5): + assert load_prompt("system").source == "local" + assert fake.get_prompt.call_count == 1 + + +def test_miss_expires_after_ttl(langfuse_enabled, monkeypatch): + fake = MagicMock(get_prompt=MagicMock(side_effect=[RuntimeError("down"), _served("now served", version=2)])) + with patch("agent.langfuse_client.langfuse_client", fake): + assert load_prompt("system").source == "local" + prompt_registry._misses["system"] = 0.0 # TTL elapsed + assert load_prompt("system").text == "now served"