diff --git a/backend/app/adapters/base.py b/backend/app/adapters/base.py index bf0a209..9a09982 100644 --- a/backend/app/adapters/base.py +++ b/backend/app/adapters/base.py @@ -99,12 +99,25 @@ async def emit_token_delta( step_index: int, delta: str, role: str = "assistant", + part: str = "text", **data: Any, ) -> None: - """Stream an incremental token chunk to SSE subscribers (no DB write).""" + """Stream an incremental token chunk to SSE subscribers (no DB write). + + ``part`` discriminates fine-grained streams: ``text`` (default) for the + visible reply, ``reasoning`` for model thinking / chain-of-thought. + Reasoning deltas are still ephemeral; persist the finished block with + ``emit_message(..., kind="reasoning")``. + """ await self.emit( "token.delta", - {"step_index": step_index, "delta": delta, "role": role, **data}, + { + "step_index": step_index, + "delta": delta, + "role": role, + "part": part, + **data, + }, ) async def emit_message( diff --git a/backend/app/adapters/langgraph_adapter.py b/backend/app/adapters/langgraph_adapter.py index 821faf1..03084ee 100644 --- a/backend/app/adapters/langgraph_adapter.py +++ b/backend/app/adapters/langgraph_adapter.py @@ -18,6 +18,7 @@ }, "system_prompt": "You are a helpful coordinator.", "stream_tokens": true, // emit token.delta SSE; defer tokens to step.updated + "stream_reasoning": false, // mock / providers: also stream part=reasoning blocks "tools": ["echo"], // builtin or MCP keys (mcp/{server}/{tool}) "mcp_servers": [ // optional MCP stdio/SSE/HTTP servers { @@ -156,6 +157,7 @@ def __init__(self, output: dict[str, Any]) -> None: @dataclass class ModelResponse: content: str + reasoning: str = "" tool_calls: list[dict[str, Any]] = field(default_factory=list) tokens_in: int = 0 tokens_out: int = 0 @@ -593,6 +595,7 @@ async def handler(state: dict[str, Any]) -> dict[str, Any]: total_out = 0 total_cost = 0.0 final_reply = "" + final_reasoning = "" had_recoverable_feedback = False policy = run_state.tool_error_policy last_response: ModelResponse | None = None @@ -698,6 +701,7 @@ async def _run_one_tool(tc: dict[str, Any]) -> ToolOutcome: continue final_reply = response.content or "" + final_reasoning = response.reasoning or "" messages.append({"role": "assistant", "content": final_reply}) break else: @@ -710,12 +714,13 @@ async def _run_one_tool(tc: dict[str, Any]) -> ToolOutcome: messages.append({"role": "assistant", "content": final_reply}) latency_ms = int((time.monotonic() - started) * 1000) - used_model = ( - (last_response.model if last_response else None) or model + cost_usd = estimate_cost_usd(model, total_in, total_out) + await _emit_assistant_messages( + ctx, + step_index=step_idx, + content=final_reply, + reasoning=final_reasoning, ) - routing_meta = last_response.routing if last_response else None - cost_usd = round(total_cost, 6) - await ctx.emit_message(role="assistant", content=final_reply, step_index=step_idx) await ctx.emit_step_updated( index=step_idx, tokens_in=total_in, @@ -827,7 +832,12 @@ async def handler(state: dict[str, Any]) -> dict[str, Any]: model=used_model, model_routing=response.routing, ) - await ctx.emit_message(role="assistant", content=reply, step_index=step_idx) + await _emit_assistant_messages( + ctx, + step_index=step_idx, + content=reply, + reasoning=response.reasoning, + ) await ctx.emit_step_completed( index=step_idx, node=spec.id, @@ -997,6 +1007,7 @@ async def _invoke_model_once( message = data["choices"][0]["message"] tool_calls = _parse_openai_tool_calls(message.get("tool_calls")) content = message.get("content") or "" + reasoning = _extract_reasoning_text(message) usage = data.get("usage") or {} prompt_text = json.dumps(messages, default=str) tokens_in = int( @@ -1004,10 +1015,15 @@ async def _invoke_model_once( ) tokens_out = int( usage.get("completion_tokens") - or estimate_tokens(content or json.dumps(tool_calls, default=str)) + or estimate_tokens( + (content or "") + + reasoning + + (json.dumps(tool_calls, default=str) if tool_calls else "") + ) ) return ModelResponse( content=content, + reasoning=reasoning, tool_calls=tool_calls, tokens_in=tokens_in, tokens_out=tokens_out, @@ -1064,12 +1080,24 @@ async def _invoke_mock( else: reply = f"[mock:{model}]{suffix} {user_input}" tokens_in = estimate_tokens(json.dumps(messages, default=str)) - tokens_out = estimate_tokens(reply) + reasoning = "" + if bool(ctx.agent_config.get("stream_reasoning")): + preview = user_input[:48].replace("\n", " ") + reasoning = f"[think:{model}] weigh options for: {preview}" + tokens_out = estimate_tokens(reply + reasoning) if stream_tokens: + if reasoning: + for chunk in _chunk_text(reasoning): + await ctx.emit_token_delta( + step_index=step_index, delta=chunk, part="reasoning" + ) for chunk in _chunk_text(reply): - await ctx.emit_token_delta(step_index=step_index, delta=chunk) + await ctx.emit_token_delta( + step_index=step_index, delta=chunk, part="text" + ) return ModelResponse( content=reply, + reasoning=reasoning, tokens_in=tokens_in, tokens_out=tokens_out, model=model, @@ -1090,6 +1118,7 @@ async def _invoke_openai_streaming( "stream_options": {"include_usage": True}, } parts: list[str] = [] + reasoning_parts: list[str] = [] tokens_in = 0 tokens_out = 0 @@ -1119,20 +1148,34 @@ async def _invoke_openai_streaming( ) for choice in chunk.get("choices") or []: delta = choice.get("delta") or {} + reasoning_piece = _extract_reasoning_text(delta) + if reasoning_piece: + reasoning_parts.append(reasoning_piece) + await ctx.emit_token_delta( + step_index=step_index, + delta=reasoning_piece, + part="reasoning", + ) content = delta.get("content") if content: parts.append(content) await ctx.emit_token_delta( - step_index=step_index, delta=content + step_index=step_index, + delta=content, + part="text", ) reply = "".join(parts) + reasoning = "".join(reasoning_parts) if not tokens_in: tokens_in = estimate_tokens(str(payload.get("messages", ""))) if not tokens_out: - tokens_out = estimate_tokens(reply) + tokens_out = estimate_tokens(reply + reasoning) return ModelResponse( - content=reply, tokens_in=tokens_in, tokens_out=tokens_out + content=reply, + reasoning=reasoning, + tokens_in=tokens_in, + tokens_out=tokens_out, ) @@ -1325,3 +1368,36 @@ def _initial_graph_state(ctx: AdapterContext) -> dict[str, Any]: def _chunk_text(text: str, *, size: int = 8) -> list[str]: """Split text into small chunks for mock streaming.""" return [text[i : i + size] for i in range(0, len(text), size)] or [text] + + +def _extract_reasoning_text(payload: dict[str, Any] | None) -> str: + """Pull reasoning / thinking text from OpenAI-compatible message or delta. + + Providers disagree on the field name (``reasoning_content``, ``reasoning``, + Anthropic-style ``thinking``). Empty / missing values are ignored. + """ + if not payload: + return "" + for key in ("reasoning_content", "reasoning", "thinking"): + value = payload.get(key) + if isinstance(value, str) and value: + return value + return "" + + +async def _emit_assistant_messages( + ctx: AdapterContext, + *, + step_index: int, + content: str, + reasoning: str = "", +) -> None: + """Persist optional reasoning block, then the visible assistant reply.""" + if reasoning.strip(): + await ctx.emit_message( + role="assistant", + content=reasoning, + step_index=step_index, + kind="reasoning", + ) + await ctx.emit_message(role="assistant", content=content, step_index=step_index) diff --git a/backend/app/services/thread_service.py b/backend/app/services/thread_service.py index 44c4002..921bb51 100644 --- a/backend/app/services/thread_service.py +++ b/backend/app/services/thread_service.py @@ -217,9 +217,11 @@ async def load_thread_window( ) -> list[dict[str, Any]]: """Return OpenAI-style chat dicts for prior thread turns, window-trimmed. - Seeds only complete user/assistant turns (no system / prompt_echo / tool). - Cross-run tool chains are incomplete and break model APIs, so they are - omitted from L1. Used by the worker when constructing ``AdapterContext``. + Seeds only complete user/assistant turns (no system / prompt_echo / + reasoning / tool). Cross-run tool chains are incomplete and break model + APIs, so they are omitted from L1. Reasoning blocks stay in the Run + transcript for the console but are not replayed into the next prompt. + Used by the worker when constructing ``AdapterContext``. """ max_messages = get_settings().thread_messages_max # Over-fetch slightly so role filtering still fills the cap. @@ -246,7 +248,7 @@ async def load_thread_window( messages: list[dict[str, Any]] = [] for msg, _run in result.all(): extra = msg.extra or {} - if extra.get("kind") == "prompt_echo": + if extra.get("kind") in ("prompt_echo", "reasoning"): continue payload = message_row_to_dict(msg) # Drop incomplete tool-call metadata from prior runs. diff --git a/backend/tests/test_langgraph_adapter.py b/backend/tests/test_langgraph_adapter.py index 20c47e4..1ca7833 100644 --- a/backend/tests/test_langgraph_adapter.py +++ b/backend/tests/test_langgraph_adapter.py @@ -294,6 +294,7 @@ async def test_langgraph_streams_token_deltas_and_defers_step_tokens(): ] assert len(token_events) > 0 assert all(e["step_index"] == 0 for e in token_events) + assert all(e.get("part", "text") == "text" for e in token_events) reply = (result.output or {}).get("reply", "") assert "".join(e["delta"] for e in token_events) == reply @@ -315,6 +316,49 @@ async def test_langgraph_streams_token_deltas_and_defers_step_tokens(): assert completed[0]["tokens_out"] == updated[0]["tokens_out"] +@pytest.mark.asyncio +async def test_langgraph_streams_and_persists_reasoning_blocks(): + adapter = LangGraphAdapter() + ctx = _RecordingContext() + ctx.agent_config = { + "model": "openai/gpt-4o-mini", + "stream_tokens": True, + "stream_reasoning": True, + } + result = await adapter.run(ctx) + assert result.status == RunStatus.SUCCEEDED + + reasoning_deltas = [ + data + for event, data in ctx.events + if event == "token.delta" and data.get("part") == "reasoning" + ] + text_deltas = [ + data + for event, data in ctx.events + if event == "token.delta" and data.get("part", "text") == "text" + ] + assert reasoning_deltas + assert text_deltas + reply = (result.output or {}).get("reply", "") + assert "".join(e["delta"] for e in text_deltas) == reply + + reasoning_msgs = [ + m + for m in _message_events(ctx.events) + if (m.get("extra") or {}).get("kind") == "reasoning" + ] + assert len(reasoning_msgs) == 1 + assert "".join(e["delta"] for e in reasoning_deltas) == reasoning_msgs[0]["content"] + + assistant = [ + m + for m in _message_events(ctx.events) + if m["role"] == "assistant" and (m.get("extra") or {}).get("kind") != "reasoning" + ] + assert assistant[-1]["content"] == reply + + @pytest.mark.asyncio async def test_langgraph_stream_tokens_disabled(): adapter = LangGraphAdapter() diff --git a/backend/tests/test_threads.py b/backend/tests/test_threads.py index b676762..f514975 100644 --- a/backend/tests/test_threads.py +++ b/backend/tests/test_threads.py @@ -243,7 +243,7 @@ async def test_thread_message_pagination(client): @pytest.mark.asyncio -async def test_load_thread_window_skips_tool_and_prompt_echo(): +async def test_load_thread_window_skips_tool_prompt_echo_and_reasoning(): from app.db.base import Base from app.db.session import SessionLocal, engine from app.models import Agent, Message, Run, Thread @@ -301,7 +301,14 @@ async def test_load_thread_window_skips_tool_and_prompt_echo(): content='{"ok":true}', tool_call_id="c1", ), - Message(run_id=run.id, index=4, role="assistant", content="done"), + Message( + run_id=run.id, + index=4, + role="assistant", + content="secret chain of thought", + extra={"kind": "reasoning"}, + ), + Message(run_id=run.id, index=5, role="assistant", content="done"), ] ) await session.commit() diff --git a/docs/api-contract.md b/docs/api-contract.md index 2bad5a7..79719d4 100644 --- a/docs/api-contract.md +++ b/docs/api-contract.md @@ -381,9 +381,15 @@ The supported `type` values are: `token.delta` is SSE-only (not persisted). Payload: ```json -{ "step_index": 0, "delta": "Hel", "role": "assistant" } +{ "step_index": 0, "delta": "Hel", "role": "assistant", "part": "text" } ``` +`part` discriminates fine-grained streams: `"text"` (default, visible reply) +or `"reasoning"` (model thinking / chain-of-thought). Finished reasoning is +persisted as a `message.created` with `extra.kind = "reasoning"` so the +console can show a collapsible block after reconnect; live deltas remain +ephemeral. + `step.updated` flushes deferred metrics on a running step (tokens, latency) before `step.completed`: @@ -592,11 +598,15 @@ ops dashboards. name: string | null; content: string; tool_call_id: string | null; - extra: Record; + extra: Record; // kind?: "prompt_echo" | "reasoning" | … created_at: string; } ``` +`extra.kind = "reasoning"` marks a persisted reasoning / thinking block +(content is the full chain-of-thought). These rows stay on the Run transcript +for the console but are excluded from Thread L1 window seeding. + ### `MessagePage` ```ts diff --git a/docs/architecture.md b/docs/architecture.md index 1805e8a..16372d9 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -226,30 +226,27 @@ When `feedback` is enabled and recoverable observations exhaust `max_tool_rounds` without a final model reply, the run fails with `tool_recovery_exhausted: reached max_tool_rounds=N`. -## Model routing / fallback +## Fine-grained streaming (reasoning blocks) -LLM provider selection is a **runtime** concern (`backend/app/runtime/model_router.py`), -not adapter-private logic. Agent config may set a primary `model` plus optional -fallbacks: +`token.delta` carries an optional `part` field: -```json -{ - "model": "openai/gpt-4o-mini", - "model_routing": { - "fallbacks": ["openai/gpt-4o"], - "max_attempts_per_model": 1, - "retry_on": ["timeout", "rate_limit", "server_error", "connection"] - } -} -``` +| `part` | Meaning | +| --- | --- | +| `text` (default) | Visible assistant reply chunk | +| `reasoning` | Model thinking / chain-of-thought chunk | + +Deltas stay SSE-only (no Redis replay log, no DB row). When the model finishes, +LangGraph persists a `message.created` with `extra.kind = "reasoning"` (full +text) before the visible assistant message. OpenAI-compatible providers that +emit `reasoning_content` / `reasoning` / `thinking` on deltas are mapped +automatically; mock mode uses `stream_reasoning: true` in agent config. + +The console folds reasoning in Messages and applies live `token.delta` drafts +so operators see thinking as it streams. Thread L1 seeding skips reasoning +rows (same as `prompt_echo`) so chain-of-thought is not replayed into the next +prompt. -`fallback_models` at the agent-config root is accepted as shorthand for -`model_routing.fallbacks`. Transient provider errors (HTTP 429/5xx, timeouts, -connection failures) advance to the next attempt or model; client errors (other -4xx) fail immediately. LangGraph records the winning model and attempt trail on -`step.updated` / `step.completed` (`model`, `model_routing`) and emits -`model_routing_fallback` / `model_routing_exhausted` log events. Metrics: -`agentflow.llm.routing_fallbacks`, `agentflow.llm.routing_exhausted`. +Multimodal attachment persistence is still outstanding. ## Unit tests diff --git a/docs/data-model.md b/docs/data-model.md index 1df05ee..b2e0717 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -143,6 +143,9 @@ stateDiagram-v2 - **`Message.step_id` is optional** so an adapter can attach a message to a specific node tick when it makes sense, while keeping the run-level ordering authoritative. +- **`Message.extra.kind = "reasoning"`** persists a finished thinking block + (content = full reasoning text). Live chunks use SSE `token.delta` with + `part: "reasoning"` and are not written to Redis replay or Postgres. - **`Thread` groups Runs for L1 short memory.** `Run.thread_id` is optional; when set, the worker seeds `AdapterContext.thread_messages` from prior runs in the same thread (window-trimmed). Messages remain Run-scoped rows. diff --git a/docs/plan.md b/docs/plan.md index ec2f326..8aaaa2f 100644 --- a/docs/plan.md +++ b/docs/plan.md @@ -79,7 +79,7 @@ - [ ] Agent 记忆服务（M1 Thread ✅；M2/M3 情景摘要 + 语义事实 + 文档 RAG 待做） - [x] 模型路由 / fallback 策略 - [ ] 定时与批量 Run -- [ ] 细粒度流式：推理块、多模态附件（DB 持久化） +- [x] 细粒度流式：推理块（`token.delta.part` + `Message.extra.kind=reasoning`）；多模态附件（DB 持久化）待做 ## Agent Memory 专项 diff --git a/frontend/components/EventStream.tsx b/frontend/components/EventStream.tsx index a7fb671..bb7597a 100644 --- a/frontend/components/EventStream.tsx +++ b/frontend/components/EventStream.tsx @@ -25,6 +25,12 @@ function formatEventData(type: string, data: Record): string { if (stateKeys) parts.push(`state keys: ${stateKeys}`); return parts.join(" · "); } + if (type === "token.delta") { + const part = data.part === "reasoning" ? "reasoning" : "text"; + const delta = typeof data.delta === "string" ? data.delta : ""; + const preview = delta.length > 40 ? `${delta.slice(0, 40)}…` : delta; + return `part=${part} · ${JSON.stringify(preview)}`; + } return JSON.stringify(data); } diff --git a/frontend/components/MessagesPanel.tsx b/frontend/components/MessagesPanel.tsx index 9ba9599..e403dfa 100644 --- a/frontend/components/MessagesPanel.tsx +++ b/frontend/components/MessagesPanel.tsx @@ -16,6 +16,16 @@ function isPromptEcho(message: Message): boolean { return message.extra?.kind === "prompt_echo"; } +function isReasoning(message: Message): boolean { + const kind = message.extra?.kind; + return kind === "reasoning" || kind === "streaming_reasoning"; +} + +function isStreaming(message: Message): boolean { + const kind = message.extra?.kind; + return kind === "streaming" || kind === "streaming_reasoning"; +} + function mergeMessages(...groups: Message[][]): Message[] { const byIndex = new Map(); for (const group of groups) { @@ -26,36 +36,42 @@ function mergeMessages(...groups: Message[][]): Message[] { return [...byIndex.values()].sort((a, b) => a.index - b.index); } -function PromptEchoRow({ - message, +function CollapsibleBlock({ + title, + meta, + content, expanded, onToggle, + dashed = false, }: { - message: Message; + title: string; + meta?: string | null; + content: string; expanded: boolean; onToggle: () => void; + dashed?: boolean; }) { - const node = - typeof message.extra.node === "string" ? message.extra.node : null; - return ( -
  • +
  • {expanded ? ( -
    - {message.content} -
    +
    {content}
    ) : null}
    • ); @@ -78,7 +94,7 @@ export function MessagesPanel({ const [nextCursor, setNextCursor] = useState(null); const [hasMore, setHasMore] = useState(messagesTruncated); const [loading, setLoading] = useState(false); - const [expandedEchoes, setExpandedEchoes] = useState>( + const [expanded, setExpanded] = useState>( () => new Set(), ); @@ -86,7 +102,7 @@ export function MessagesPanel({ setOlder([]); setNextCursor(null); setHasMore(messagesTruncated); - setExpandedEchoes(new Set()); + setExpanded(new Set()); }, [runId, messagesTruncated]); const displayed = useMemo( @@ -98,6 +114,10 @@ export function MessagesPanel({ () => displayed.filter(isPromptEcho).length, [displayed], ); + const reasoningCount = useMemo( + () => displayed.filter(isReasoning).length, + [displayed], + ); async function loadOlder() { if (loading || !hasMore) return; @@ -115,13 +135,13 @@ export function MessagesPanel({ } } - function toggleEcho(index: number) { - setExpandedEchoes((prev) => { + function toggle(key: string) { + setExpanded((prev) => { const next = new Set(prev); - if (next.has(index)) { - next.delete(index); + if (next.has(key)) { + next.delete(key); } else { - next.add(index); + next.add(key); } return next; }); @@ -137,6 +157,9 @@ export function MessagesPanel({ {promptEchoCount > 0 ? ` · ${promptEchoCount} prompt echo${promptEchoCount === 1 ? "" : "es"} collapsed` : ""} + {reasoningCount > 0 + ? ` · ${reasoningCount} reasoning block${reasoningCount === 1 ? "" : "s"}` + : ""} {hasMore ? " · older available" : ""} ) : null} @@ -154,21 +177,51 @@ export function MessagesPanel({ ) : null}
      - {displayed.map((message) => - isPromptEcho(message) ? ( - toggleEcho(message.index)} - /> - ) : ( + {displayed.map((message) => { + const key = message.id; + if (isPromptEcho(message)) { + const node = + typeof message.extra.node === "string" ? message.extra.node : null; + return ( + toggle(key)} + dashed + /> + ); + } + if (isReasoning(message)) { + return ( + toggle(key)} + /> + ); + } + return (
    1. {message.role} + {isStreaming(message) ? " · streaming…" : ""} {message.name ? ` · ${message.name}` : ""} {message.step_id ? ( @@ -187,8 +240,8 @@ export function MessagesPanel({
      {message.content}
      2. - ), - )} + ); + })}
    {displayed.length === 0 ? ( diff --git a/frontend/lib/run-events.ts b/frontend/lib/run-events.ts index 9836b75..b47a52b 100644 --- a/frontend/lib/run-events.ts +++ b/frontend/lib/run-events.ts @@ -181,6 +181,71 @@ function appendMessage(messages: Message[], data: Record, at: s return [...messages, message].sort((a, b) => a.index - b.index); } +function streamingDraftId(stepIndex: number, part: "text" | "reasoning"): string { + return `sse-stream-${stepIndex}-${part}`; +} + +function appendTokenDelta( + messages: Message[], + data: Record, + at: string, +): Message[] { + const stepIndex = data.step_index; + if (typeof stepIndex !== "number") return messages; + const delta = typeof data.delta === "string" ? data.delta : ""; + if (!delta) return messages; + + const part = data.part === "reasoning" ? "reasoning" : "text"; + const draftId = streamingDraftId(stepIndex, part); + const existing = messages.find((message) => message.id === draftId); + if (existing) { + return messages.map((message) => + message.id === draftId + ? { ...message, content: message.content + delta, created_at: at } + : message, + ); + } + + const index = + messages.length > 0 + ? Math.max(...messages.map((message) => message.index)) + 1 + : 0; + + const draft: Message = { + id: draftId, + index, + step_id: null, + role: "assistant", + name: null, + content: delta, + tool_call_id: null, + extra: { + kind: part === "reasoning" ? "streaming_reasoning" : "streaming", + step_index: stepIndex, + }, + created_at: at, + }; + return [...messages, draft].sort((a, b) => a.index - b.index); +} + +function dropStreamingDrafts( + messages: Message[], + data: Record, +): Message[] { + const stepIndex = + typeof data.step_index === "number" ? data.step_index : null; + if (stepIndex == null) return messages; + + const kind = asRecord(data.extra).kind; + const dropIds = new Set(); + if (kind === "reasoning") { + dropIds.add(streamingDraftId(stepIndex, "reasoning")); + } else { + dropIds.add(streamingDraftId(stepIndex, "text")); + } + return messages.filter((message) => !dropIds.has(message.id)); +} + function appendToolCall(step: Step, data: Record): ToolCall { const callId = typeof data.call_id === "string" && data.call_id @@ -411,10 +476,19 @@ export function applyRunEvent(run: Run, event: RunEvent): Run { usage: aggregateUsageFromSteps(steps), }; } + case "token.delta": + return { + ...next, + messages: appendTokenDelta(run.messages, data, at), + }; case "message.created": return { ...next, - messages: appendMessage(run.messages, data, at), + messages: appendMessage( + dropStreamingDrafts(run.messages, data), + data, + at, + ), }; case "tool_call.started": { const stepIndex = data.step_index;