File size: 7,459 Bytes
81e5fe7 0e5fdb5 81e5fe7 ea9848d 81e5fe7 5a60e93 81e5fe7 0721bb4 81e5fe7 d65c41d 81e5fe7 0e5fdb5 81e5fe7 5a60e93 b68816f c089593 81e5fe7 0e5fdb5 5a60e93 b68816f c089593 5a60e93 81e5fe7 0e5fdb5 81e5fe7 ea9848d 0721bb4 ea9848d 0721bb4 81e5fe7 0721bb4 81e5fe7 0721bb4 81e5fe7 0721bb4 81e5fe7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 | """Assembler — single LLM call at the end of the slow path.
Reads the `RunState` (all `TaskResult`s) + `BusinessContext` and produces an
`AssembledOutput` { chat_answer, analysis_record }. Owns all language/output: prose,
markdown tables, citations, and merging structured + unstructured results.
The model authors only the *narrative* (`AssemblerNarrative`); this service copies
the structured pass-through (`results_snapshot`, `tasks_run`) and metadata from the
`RunState` so the record stays a faithful source of truth (§8.3, INV-4).
Chain construction mirrors `agents/planner/service.py`.
See AGENT_ARCHITECTURE_CONTEXT_new.md §7.5.
"""
from __future__ import annotations
from datetime import UTC, datetime
from pathlib import Path
from langchain_core.messages import SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable
from langchain_openai import AzureChatOpenAI
from src.middlewares.logging import get_logger
from ..language import detect_reply_language
from ..planner.contracts import BusinessContext
from .errors import AssemblerError
from .prompt import _MIN_TRIM_CHARS, _trim_lists, build_assembler_prompt
from .schemas import (
AnalysisRecord,
AssembledOutput,
AssemblerNarrative,
RunAssessment,
RunState,
TaskResult,
TaskSummary,
)
logger = get_logger("assembler")
_PROMPT_PATH = (
Path(__file__).resolve().parent.parent.parent / "config" / "prompts" / "assembler.md"
)
def _load_prompt_text() -> str:
return _PROMPT_PATH.read_text(encoding="utf-8")
def _build_default_chain() -> Runnable:
from src.config.settings import settings
llm = AzureChatOpenAI(
azure_deployment=settings.azureai_deployment_name_54m,
openai_api_version=settings.azureai_api_version_54m,
azure_endpoint=settings.azureai_endpoint_url_54m,
api_key=settings.azureai_api_key_54m,
temperature=0,
)
prompt = ChatPromptTemplate.from_messages(
[
SystemMessage(content=_load_prompt_text()),
("human", "{human_content}"),
]
)
return prompt | llm.with_structured_output(AssemblerNarrative)
_default_chain: Runnable | None = None
def _get_default_chain() -> Runnable:
global _default_chain
if _default_chain is None:
_default_chain = _build_default_chain()
return _default_chain
class Assembler:
"""Wraps the single Assembler LLM call. Inject `structured_chain` for tests."""
def __init__(self, structured_chain: Runnable | None = None) -> None:
self._chain = structured_chain
def _ensure_chain(self) -> Runnable:
if self._chain is None:
self._chain = _get_default_chain()
return self._chain
async def assemble(
self,
run_state: RunState,
context: BusinessContext,
question: str | None = None,
reply_language: str | None = None,
callbacks: list | None = None,
assessment: RunAssessment | None = None,
domain_terms: str | None = None,
document_context: str | None = None,
) -> AssembledOutput:
chain = self._ensure_chain()
# `reply_language` is detected upstream from the ORIGINAL user message. Fall back
# to `question` only if not provided — but note `question` is the router's
# rewritten_query, which may be normalized to English, so the caller should pass it.
if reply_language is None:
reply_language = detect_reply_language([], message=question)
human_content = build_assembler_prompt(
run_state, context, question, reply_language, assessment=assessment,
domain_terms=domain_terms, document_context=document_context,
)
try:
if callbacks:
narrative: AssemblerNarrative = await chain.ainvoke(
{"human_content": human_content}, config={"callbacks": callbacks}
)
else:
narrative = await chain.ainvoke({"human_content": human_content})
except Exception as exc: # surface as a typed error for the caller
raise AssemblerError(f"assembler call failed: {exc}") from exc
record = _build_record(narrative, run_state)
logger.info(
"analysis assembled",
plan_id=run_state.plan_id,
business_context_id=run_state.business_context_id,
n_tasks=len(run_state.results),
reply_language=reply_language,
)
return AssembledOutput(chat_answer=narrative.chat_answer, analysis_record=record)
# Persisted records cap raw `table` rows from data-access tools (retrieve_data can
# return up to the 10k LIMIT): the report never renders raw rows, so storing them all
# would bloat every record's jsonb.
#
# `stats` values are trimmed too (D-9, 2026-09-23). The note that stood here called
# analyze_* outputs "small" — not always: a merge or pareto over thousands of rows
# returns thousands of entries, persisted whole. They get the SAME list bound the
# assembler prompt uses (`prompt._trim_lists`, 20 items + a "+N more" marker) and only
# above the same `_MIN_TRIM_CHARS`, so every self-bounded tool is stored untouched.
# Scalars and dict keys survive, and `meta` is never touched — auto-charts live in
# `meta["charts"]` and the report reads them from here.
_SNAPSHOT_ROW_SAMPLE = 10
def _trim_for_snapshot(result: TaskResult) -> TaskResult:
trimmed = []
changed = False
for out in result.outputs:
if out.kind == "table" and out.rows is not None and len(out.rows) > _SNAPSHOT_ROW_SAMPLE:
changed = True
trimmed.append(
out.model_copy(
update={
"rows": out.rows[:_SNAPSHOT_ROW_SAMPLE],
"meta": {**out.meta, "total_rows": len(out.rows), "rows_truncated": True},
}
)
)
elif out.kind == "stats" and len(str(out.value)) > _MIN_TRIM_CHARS:
changed = True
trimmed.append(
out.model_copy(
update={
"value": _trim_lists(out.value, 0),
"meta": {**out.meta, "value_truncated": True},
}
)
)
else:
trimmed.append(out)
return result.model_copy(update={"outputs": trimmed}) if changed else result
def _build_record(narrative: AssemblerNarrative, run_state: RunState) -> AnalysisRecord:
tasks_run = [
TaskSummary(
task_id=task_id,
stage=result.stage,
objective=result.objective,
status=result.status,
tools_used=[o.tool for o in result.outputs],
)
for task_id, result in run_state.results.items()
]
results_snapshot = {
task_id: _trim_for_snapshot(result) for task_id, result in run_state.results.items()
}
return AnalysisRecord(
goal_restated=narrative.goal_restated,
findings=narrative.findings,
caveats=narrative.caveats,
data_used=narrative.data_used,
open_questions=narrative.open_questions,
tasks_run=tasks_run,
results_snapshot=results_snapshot,
plan_id=run_state.plan_id,
business_context_id=run_state.business_context_id,
created_at=datetime.now(UTC),
)
|