feat(agent): use native parallel nodes and early reply streaming

src/agent/collect.py

@@ -92,6 +92,23 @@ async def validated_reply(
     或关键字替模型选择动作。非法输出不写回图状态；修复仍失败则抛出
     不含模型原文的错误。
     """
+    async def invoke(payload):
+        return await _validated_reply(**payload)
+
+    # 原生 chain-end 事件表示整条回复已经通过校验；它早于字段提取/业务提交。
+    from langchain_core.runnables import RunnableLambda
+
+    return await RunnableLambda(invoke).ainvoke(
+        dict(llm=llm, system_prompt=system_prompt, user_message=user_message,
+             current=current, validator=validator, history=history,
+             reply_validator=reply_validator),
+        config={"run_name": "validated_reply", "tags": ["reply"]},
+    )
+
+
+async def _validated_reply(
+    llm, system_prompt, user_message, current, *, validator, history, reply_validator,
+):
     for attempt in range(2):
         prompt = system_prompt
         if attempt:
@@ -273,15 +290,15 @@ def _bookkeeping(state: AgentState) -> tuple[int, int]:
     return no_reply_count, state.get("clarify_count") or 0
 
 
-async def run_collect(state: AgentState, llm: LLMService, *, now: datetime | None = None) -> dict:
-    """采集节点核心：一轮对话的 LLM 接入（参考实现）。
+def collect_time(state: AgentState, now: datetime | None = None) -> datetime:
+    if now is None and state.get("turn_observed_at"):
+        return datetime.fromisoformat(state["turn_observed_at"])
+    return observation_time(now)
 
-    流程：一次模型调用 → 前缀校验（不合规修复一次）→
-    白名单守卫（非法提议修复一次）→ 字段候选落库 → 返回状态增量。
-    任何一步修复后仍不合规都抛 CollectReplyError：不提交业务状态、
-    错误信息不携带原始模型内容。
-    """
-    moment = observation_time(now)
+
+async def collect_reply(state: AgentState, llm: LLMService, *, now: datetime | None = None) -> dict:
+    """只生成并校验回复；不等待独立的字段抽取分支。"""
+    moment = collect_time(state, now)
     evidence = time_evidence(state, moment)
     system_prompt = render_prompt(COLLECTION, state, now=moment) + context_prompt(state, time_context=evidence)
     user_message = _build_user_message(state, now=moment)
@@ -290,11 +307,6 @@ async def run_collect(state: AgentState, llm: LLMService, *, now: datetime | Non
         llm, system_prompt, user_message, current, history=conversation_history(state),
         reply_validator=lambda code, body: valid_time_reply(code, body, evidence),
     )
-
-
-    # 4) 字段提取与表单差量（提取调用在对话调用之外，见 _extract_turn_fields）
-    extracted = await _extract_turn_fields(state, llm, now=moment, time_context=evidence)
-    accident_fields, patch = _merge_form(state, extracted)
     no_reply_count, clarify_count = _bookkeeping(state)
 
     # 5) 状态增量：response_text 保留完整前缀文本，API 层负责解析；
@@ -311,8 +323,30 @@ async def run_collect(state: AgentState, llm: LLMService, *, now: datetime | Non
         "response_text": raw,
         "stage_code": stage,
         "raw_status_code": stage,
-        "accident_fields": accident_fields,
-        "turn_patch": patch,
         "dialogue": dialogue,
         "dialogue_timestamps": timestamps,
     }
+
+
+async def collect_extraction(state: AgentState, llm: LLMService) -> dict:
+    """与回复读取同一份轮次快照，返回独立的候选结果通道。"""
+    moment = collect_time(state)
+    extracted = await _extract_turn_fields(
+        state, llm, now=moment, time_context=time_evidence(state, moment),
+    )
+    return {"extracted_accident_fields": extracted}
+
+
+def merge_collect(state: AgentState) -> dict:
+    form, patch = _merge_form(state, state.get("extracted_accident_fields") or {})
+    return {"accident_fields": form, "turn_patch": patch, "extracted_accident_fields": {}}
+
+
+async def run_collect(state: AgentState, llm: LLMService, *, now: datetime | None = None) -> dict:
+    """独立调用入口；图内通过两个原生节点并行执行回复和提取。"""
+    moment = observation_time(now)
+    snapshot = {**state, "turn_observed_at": moment.isoformat()}
+    reply = await collect_reply(snapshot, llm)
+    extraction = await collect_extraction(snapshot, llm)
+    merged = merge_collect({**snapshot, **reply, **extraction})
+    return {**reply, **merged}

src/agent/graph.py

@@ -8,6 +8,8 @@
 - 节点间大切换由 ``route_by_stage`` 条件边按 ``stage_code`` 分发
   （对照 Condition#2/#7/#9 等）；一轮 HTTP 请求 = 一次图执行，
   跨轮记忆由 checkpoint 承载；
+- 采集阶段回复与字段提取由原生条件分支同时调度，汇合后合并差量；
+  两个分支读取同一轮时间与状态，各自写独立通道；
 - 节点内阶段码一律由 LLM 提议（提示词即程序），代码只校验
   （前缀格式、白名单、clientMode 出口）与落库；
 - 依赖注入走 LangGraph 原生机制：``context_schema=GraphDeps``，
@@ -18,7 +20,8 @@
 图结构::
 
     START → normalize_input → route ─┬→ welcome（首轮固定问候）→ END
-                                     ├→ collect ───────────→ END
+                                     ├→ collect ──────────┐
+                                     │  extract_accident_fields ─┴→ merge_collect → END
                                      ├→ single_photo ──────→ END
                                      ├→ dual_photo ────────→ END
                                      ├→ single_confirm ────→ END
@@ -40,7 +43,7 @@ from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.graph import END, START, StateGraph
 from langgraph.runtime import Runtime
 
-from .collect import run_collect
+from .collect import collect_reply, collect_extraction, collect_time, merge_collect
 from .confirm import run_dual_confirm, run_single_confirm
 from .events import parse_client_event
 from .llm import LLMService, workflow_fake_llm
@@ -74,6 +77,10 @@ def normalize_input(state: AgentState) -> dict:
         "input_text": cleaned,
         "client_event": event.value if event else "",
         "turn_patch": {},
+        "extracted_accident_fields": {},
+        "turn_observed_at": (
+            collect_time({}) if route_by_stage(state) == "collect" else observation_time()
+        ).isoformat(),
     }
 
 
@@ -135,7 +142,16 @@ def welcome(state: AgentState) -> dict:
 
 async def collect_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict:
     """事故信息收集节点（工作流「事故信息收集」chatNode）。"""
-    return await run_collect(state, runtime.context.llm)
+    return await collect_reply(state, runtime.context.llm)
+
+
+async def extract_accident_fields(state: AgentState, runtime: Runtime[GraphDeps]) -> dict:
+    return await collect_extraction(state, runtime.context.llm)
+
+
+def route_turn(state: AgentState) -> str | list[str]:
+    target = route_by_stage(state)
+    return ["collect", "extract_accident_fields"] if target == "collect" else target
 
 
 async def single_photo_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict:
@@ -163,7 +179,7 @@ def build_graph(checkpointer=None):
 
     参数:
         checkpointer: 自定义 checkpointer；缺省 ``InMemorySaver``，
-            作业 7 替换为 ``AsyncPostgresSaver``。
+            持久化模式由运行时注入 SQLite/PostgreSQL saver。
 
     LLM 依赖不在此处绑定——调用时通过
     ``config={"configurable": {"thread_id": <会话>}}`` 区分会话，
@@ -173,6 +189,8 @@ def build_graph(checkpointer=None):
     builder.add_node("normalize_input", normalize_input)
     builder.add_node("welcome", welcome)
     builder.add_node("collect", collect_node)
+    builder.add_node("extract_accident_fields", extract_accident_fields)
+    builder.add_node("merge_collect", merge_collect)
     builder.add_node("single_photo", single_photo_node)
     builder.add_node("dual_photo", dual_photo_node)
     builder.add_node("single_confirm", single_confirm_node)
@@ -181,10 +199,11 @@ def build_graph(checkpointer=None):
     builder.add_edge(START, "normalize_input")
     builder.add_conditional_edges(
         "normalize_input",
-        route_by_stage,
+        route_turn,
         {
             "welcome": "welcome",
             "collect": "collect",
+            "extract_accident_fields": "extract_accident_fields",
             "single_photo": "single_photo",
             "dual_photo": "dual_photo",
             "single_confirm": "single_confirm",
@@ -192,8 +211,10 @@ def build_graph(checkpointer=None):
             END: END,
         },
     )
+    builder.add_edge(["collect", "extract_accident_fields"], "merge_collect")
+    builder.add_edge("merge_collect", END)
     for name in (
-        "welcome", "collect", "single_photo", "dual_photo",
+        "welcome", "single_photo", "dual_photo",
         "single_confirm", "dual_confirm",
     ):
         builder.add_edge(name, END)

src/agent/llm.py

@@ -201,6 +201,8 @@ class OpenAILLMService:
                     response = await client.ainvoke(
                         messages,
                         config={"run_name": "llm.complete", "metadata": {"llm_attempt": attempt + 1}},
+                        # astream_events 的 callback 不能把抽取调用隐式改成流式。
+                        stream=False,
                     )
                 content = self._text(response)
                 if not content:

src/agent/reply_stream.py

@@ -0,0 +1,76 @@
+"""LangChain 原生 Runnable 文本流：在输出前验证阶段前缀。"""
+
+from contextlib import aclosing
+
+from langchain_core.runnables import RunnableGenerator
+
+from .collect import STATE_PREFIX_RE, parse_state_reply
+
+REPLY_TOKENS_RUN = "validated_reply_tokens"
+REPLY_RUN = "validated_reply"
+
+
+class ValidatedStreamingLLM:
+    """为现有 LLMService 添加受控流；抽取调用仍走 complete。
+
+    流事件由 RunnableGenerator 原生生成，不自行创建任务或事件队列。
+    每个请求独立实例，避免并行请求共享前缀/重试状态。
+    """
+
+    def __init__(self, inner):
+        self._inner = inner
+        self.response_started = False
+
+    async def complete(self, system_prompt, user_message, *, history=None):
+        return await self._inner.complete(system_prompt, user_message, history=history)
+
+    async def complete_validated(self, system_prompt, user_message, *, history=None, stage_validator):
+        raw = ""
+
+        async def generate(inputs):
+            nonlocal raw
+            # RunnableGenerator 接收上游输入流；每次调用恰好一个 payload。
+            async for _ in inputs:
+                stream_tokens = getattr(self._inner, "stream_tokens", None)
+                if stream_tokens is None:
+                    raw = await self.complete(system_prompt, user_message, history=history)
+                    try:
+                        code, _ = parse_state_reply(raw)
+                    except ValueError:
+                        return
+                    if stage_validator(code):
+                        self.response_started = True
+                        yield raw
+                    return
+
+                chunks = []
+                prefix = ""
+                accepted = False
+                try:
+                    async with aclosing(stream_tokens(system_prompt, user_message, history=history)) as source:
+                        async for token in source:
+                            chunks.append(token)
+                            if not accepted:
+                                prefix += token
+                                match = STATE_PREFIX_RE.match(prefix)
+                                if match is None:
+                                    if len(prefix) > 1024 or "</state>" in prefix:
+                                        return
+                                    continue
+                                if not stage_validator(match.group(1)):
+                                    return
+                                accepted = True
+                                self.response_started = True
+                                yield prefix
+                            else:
+                                yield token
+                finally:
+                    raw = "".join(chunks)
+
+        runnable = RunnableGenerator(generate).with_config(
+            run_name=REPLY_TOKENS_RUN, tags=["reply_tokens"],
+        )
+        async with aclosing(runnable.astream(None)) as source:
+            async for _ in source:
+                pass  # 原生 astream_events 消费这些 chain-stream 事件。
+        return raw

src/agent/state.py

@@ -40,6 +40,8 @@ class AgentState(TypedDict):
     clarify_count: int          # 澄清计数（阈值按阶段配置，作业 4-9）
     no_reply_count: int         # 连续无回复计数（两次 → 0004）
     turn_patch: dict            # 本轮输出字段差量；每轮开始清空（作业 3-7）
+    turn_observed_at: str       # 并行分支共用的服务器参考时间
+    extracted_accident_fields: dict  # 抽取分支临时通道；汇合后清空
     state_version: int          # 业务版本（作业 7 并发冲突检测）
 
     # ---- 对话上下文（作业 4/5） ----

src/api/endpoints.py

@@ -11,6 +11,7 @@ from ..backends.chat import (
     ChatBackendRateLimitError,
     ChatInput,
     FormUpdate,
+    ReplyFinished,
     TextDelta,
 )
 from ..core.fastgpt_client import (
@@ -309,6 +310,17 @@ async def chat(
                                 f"ttfb_ms={(time.perf_counter() - stream_started_at) * 1000:.1f}"
                             )
 
+                        if isinstance(event, ReplyFinished):
+                            # 回复已独立通过校验；抽取/提交可能仍在进行。
+                            # 最后一句立即刷新，对外不新增事件，done 仍等待提交。
+                            if state_code_found:
+                                text = clean_later_state_tags("", final=True)
+                                for text_event in build_text_delta_events(text):
+                                    yield text_event
+                            for text_event in flush_text_chunker_events():
+                                yield text_event
+                            continue
+
                         if isinstance(event, FormUpdate) and not module_form_sent:
                             if event.data:
                                 form_update_payload = event.data

src/backends/chat.py

@@ -40,7 +40,12 @@ class FormUpdate:
     data: Any
 
 
-ChatStreamEvent = TextDelta | FormUpdate
+@dataclass(frozen=True)
+class ReplyFinished:
+    """完整回复已经校验；用于刷新文本缓冲，不表示业务提交成功。"""
+
+
+ChatStreamEvent = TextDelta | FormUpdate | ReplyFinished
 
 
 @dataclass(frozen=True)

src/backends/langgraph.py

@@ -1,27 +1,19 @@
-"""LangGraph implementation of the backend-neutral chat contract（作业 2）。
+"""LangGraph 后端：原生 astream_events 驱动增量回复与结束信号。
 
-实现 ``ChatBackend`` Protocol，把图执行结果翻译为公共契约：
-- ``complete()`` 返回 ``ChatResult``，``content`` 保留
-  ``<state>XXXX</state>`` 前缀（API 层负责解析）；
-- ``stream()`` 产生 ``TextDelta`` / ``FormUpdate``。
-  作业 8 起为真实增量：图内对话调用的模型 token 逐条透出，
-  字段提取等辅助调用经标记旁路（见 _TokenPassThroughLLM）。
-
-通过 ``build_graph()`` 编译的图构造；会话用
-``configurable.thread_id = chat_input.session_id`` 绑定。
-
-对应测试：test/backends/test_langgraph_backend.py、
-test/backends/test_langgraph_streaming.py
+图调度、并行分支和事件生命周期交给 LangGraph/LangChain。
+只转发经过前缀校验的回复 Runnable 文本，抽取模型的原生事件不作为正文。
+业务提交成功后才公开 FormUpdate；公开 SSE done 仍由 API 层产生。
 """
 
-import asyncio
 import inspect
-from contextlib import suppress
-from collections.abc import AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterator
+from contextlib import aclosing
+
+from langchain_core.runnables import RunnableLambda
 
-from ..agent.collect import STATE_PREFIX_RE, parse_state_reply
 from ..agent.graph import GraphDeps, build_graph, workflow_fake_llm
 from ..agent.llm import LLMAuthError, LLMRateLimitError, LLMService
+from ..agent.reply_stream import REPLY_RUN, REPLY_TOKENS_RUN, ValidatedStreamingLLM
 from .chat import (
     ChatBackend,
     ChatBackendAuthenticationError,
@@ -30,77 +22,15 @@ from .chat import (
     ChatResult,
     ChatStreamEvent,
     FormUpdate,
+    ReplyFinished,
     TextDelta,
 )
 
-# 流式桥接的队列哨兵：图任务完成后入队，消费端据此收尾
-_DONE = object()
-
-
-class _TokenPassThroughLLM:
-    """先验证阶段前缀；已经输出的回复不能重生成后拼接。"""
-
-    def __init__(self, inner: LLMService, sink: Callable[[str], Awaitable[None]]):
-        self._inner = inner
-        self._sink = sink
-        self.response_started = False
-
-    async def complete(self, system_prompt, user_message, *, history=None):
-        # 提取等未经过 validated_reply 的辅助调用始终静默。
-        return await self._inner.complete(system_prompt, user_message, history=history)
-
-    async def complete_validated(
-        self, system_prompt, user_message, *, history=None, stage_validator,
-    ) -> str:
-        stream_tokens = getattr(self._inner, "stream_tokens", None)
-        if stream_tokens is None:
-            raw = await self.complete(system_prompt, user_message, history=history)
-            try:
-                code, _ = parse_state_reply(raw)
-            except ValueError:
-                return raw
-            if stage_validator(code):
-                self.response_started = True
-                await self._sink(raw)
-            return raw
-
-        chunks: list[str] = []
-        prefix = ""
-        accepted = False
-        source = stream_tokens(system_prompt, user_message, history=history)
-        try:
-            async for token in source:
-                chunks.append(token)
-                if not accepted:
-                    prefix += token
-                    match = STATE_PREFIX_RE.match(prefix)
-                    if match is None:
-                        if len(prefix) > 1024 or "</state>" in prefix:
-                            return "".join(chunks)
-                        continue
-                    if not stage_validator(match.group(1)):
-                        return "".join(chunks)
-                    accepted = True
-                    self.response_started = True
-                    await self._sink(prefix)
-                else:
-                    await self._sink(token)
-            return "".join(chunks)
-        finally:
-            close = getattr(source, "aclose", None)
-            if close is not None:
-                await close()
+TURN_RUN = "accident_turn"
 
 
 class LangGraphBackend(ChatBackend):
-    """Adapt a compiled LangGraph to the neutral chat contract."""
-
     def __init__(self, graph=None, llm=None):
-        """参数:
-        graph: 编译后的图；缺省调用 ``build_graph()``。
-        llm: LLM 服务（运行时依赖）；缺省为离线 workflow fake。
-            生产环境由 lifespan 注入 ``build_llm_service()`` 的实例。
-        """
         self._graph = graph if graph is not None else build_graph()
         self._llm = llm if llm is not None else workflow_fake_llm()
 
@@ -109,17 +39,6 @@ class LangGraphBackend(ChatBackend):
         await delete_checkpoint(self._graph, session_id)
 
     async def complete(self, chat_input: ChatInput) -> ChatResult:
-        """非流式：执行一轮图并翻译结果。
-
-        TODO(作业 2): 实现该方法。要求：
-        - 用 ``ainvoke`` 执行图，``thread_id`` 取
-          ``chat_input.session_id``；
-        - 初始 state 至少包含 ``session_id``、``input_text``；
-        - 返回 ``ChatResult(content=<响应文本（含 state 前缀）>,
-          status_code=None, form_update={})``；
-        - 本阶段不处理 ``need_form_update`` 的真实表单差量
-          （作业 6 起实现）。
-        """
         result = await self._complete_turn(chat_input)
         patch = result.get("turn_patch") or {}
         return ChatResult(
@@ -131,17 +50,7 @@ class LangGraphBackend(ChatBackend):
     async def _complete_turn(
         self, chat_input: ChatInput, llm: LLMService | None = None,
     ) -> dict:
-        """执行一轮图，把 LLM 层错误翻译为后端中立错误。
-
-        llm：运行时 LLM 依赖；缺省用构造注入的实例。stream 路径传入
-        token 透传包装器（_TokenPassThroughLLM）。
-
-        错误码等价（对照 FastGPT 后端）：
-        LLMAuthError -> ChatBackendAuthenticationError（API 层 401）；
-        LLMRateLimitError -> ChatBackendRateLimitError（API 层 429）。
-        """
-        graph_input: dict = dict(chat_input.injected_fields or {})
-        # 恢复快照中的上一轮 input_text 不得覆盖本次请求。
+        graph_input = dict(chat_input.injected_fields or {})
         graph_input.update({
             "session_id": chat_input.session_id,
             "input_text": chat_input.text,
@@ -153,7 +62,6 @@ class LangGraphBackend(ChatBackend):
                 config={"configurable": {"thread_id": chat_input.session_id}},
                 context=GraphDeps(llm=llm or self._llm),
             )
-            # 即使消费者在图完成后断开，业务 repository 仍与 checkpoint 一致。
             if chat_input.on_commit is not None:
                 pending = chat_input.on_commit(result.get("turn_patch") or {}, state=result)
                 if inspect.isawaitable(pending):
@@ -165,58 +73,43 @@ class LangGraphBackend(ChatBackend):
             raise ChatBackendRateLimitError(str(exc)) from exc
 
     async def stream(self, chat_input: ChatInput) -> AsyncIterator[ChatStreamEvent]:
-        """流式（作业 8 真实增量）：模型 token 逐条透出。
+        llm = ValidatedStreamingLLM(self._llm)
 
-        结构：后台任务执行一轮图（LLM 经 _TokenPassThroughLLM 把
-        对话 token 推入队列），本生成器从队列消费并产出
-        ``TextDelta``；图完成后产出 ``FormUpdate``（差量）。
-        - API 层的缓冲解析（state 标签跨片段拼接、stage 事件优先）
-          本就支持任意分片，无需改动；
-        - 生成器被关闭（客户端断开）时取消图任务，停止 LLM 调用；
-        - 图任务异常在消费端重新抛出（错误映射沿用 _complete_turn）。
-        """
-        queue: asyncio.Queue = asyncio.Queue(maxsize=32)
-        token_count = 0
+        async def run_turn(request):
+            return await self._complete_turn(request, llm=llm)
 
-        async def sink(token: str) -> None:
-            nonlocal token_count
-            token_count += 1
-            await queue.put(token)
+        turn = RunnableLambda(run_turn).with_config(run_name=TURN_RUN)
+        text_started = False
+        reply_finished = False
+        result = None
+        # 底层负责关闭时取消并等待任务，避免业务锁释放后残留图执行。
+        source = turn.astream_events(
+            chat_input, version="v2",
+            include_names=[TURN_RUN, REPLY_RUN, REPLY_TOKENS_RUN],
+        )
+        async with aclosing(source):
+            async for event in source:
+                name = event["name"]
+                kind = event["event"]
+                if name == REPLY_TOKENS_RUN and kind == "on_chain_stream":
+                    text = event["data"]["chunk"]
+                    if text:
+                        text_started = True
+                        yield TextDelta(text)
+                elif name == REPLY_RUN and kind == "on_chain_end":
+                    # validated_reply 正常结束，完整回复才算通过校验。
+                    if not reply_finished:
+                        reply_finished = True
+                        yield ReplyFinished()
+                elif name == TURN_RUN and kind == "on_chain_end" and not event["parent_ids"]:
+                    result = event["data"]["output"]
 
-        async def run_turn() -> dict:
-            try:
-                result = await self._complete_turn(
-                    chat_input, llm=_TokenPassThroughLLM(self._llm, sink)
-                )
-            except asyncio.CancelledError:
-                raise
-            except Exception as exc:
-                await queue.put(exc)
-                return {}
-            await queue.put(_DONE)
-            return result
-
-        task = asyncio.ensure_future(run_turn())
-        try:
-            while True:
-                item = await queue.get()
-                if item is _DONE:
-                    break
-                if isinstance(item, BaseException):
-                    raise item
-                yield TextDelta(item)
-            result = await task
-            if token_count == 0:
-                # 本回合无 LLM 调用（welcome 等固定应答轮）：
-                # sink 无 token 可推，回退为完整文本切片（旧行为）
-                text = result.get("response_text") or ""
-                if text:
-                    yield TextDelta(text)
-            form_update = result.get("turn_patch") or {}
-            if form_update and chat_input.need_form_update:
-                yield FormUpdate(form_update)
-        finally:
-            if not task.done():
-                task.cancel()
-            with suppress(asyncio.CancelledError):
-                await task
+        if result is None:
+            raise RuntimeError("Graph stream ended without a committed result")
+        if not text_started:
+            text = result.get("response_text") or ""
+            if text:
+                yield TextDelta(text)
+        patch = result.get("turn_patch") or {}
+        if patch and chat_input.need_form_update:
+            yield FormUpdate(patch)