"""LangGraph implementation of the backend-neutral chat contract(作业 2)。
实现 ``ChatBackend`` Protocol,把图执行结果翻译为公共契约:
- ``complete()`` 返回 ``ChatResult``,``content`` 保留
``XXXX`` 前缀(API 层负责解析);
- ``stream()`` 产生 ``TextDelta`` / ``FormUpdate``;初期允许把固定完整答案
切成若干片段,真实增量输出在作业 8 接入。
通过 ``build_graph()`` 编译的图构造;会话用
``configurable.thread_id = chat_input.session_id`` 绑定。
对应测试:test/backends/test_langgraph_backend.py
"""
from collections.abc import AsyncIterator
from ..agent.graph import GraphDeps, build_graph, workflow_fake_llm
from ..agent.llm import LLMAuthError, LLMRateLimitError
from .chat import (
ChatBackend,
ChatBackendAuthenticationError,
ChatBackendRateLimitError,
ChatInput,
ChatResult,
ChatStreamEvent,
FormUpdate,
TextDelta,
)
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()
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)
return ChatResult(
content = result["response_text"],
status_code = None,
form_update = result.get("turn_patch") or {}
)
async def _complete_turn(self, chat_input: ChatInput) -> dict:
"""执行一轮图,把 LLM 层错误翻译为后端中立错误。
错误码等价(对照 FastGPT 后端):
LLMAuthError -> ChatBackendAuthenticationError(API 层 401);
LLMRateLimitError -> ChatBackendRateLimitError(API 层 429)。
"""
graph_input: dict = {
"session_id": chat_input.session_id,
"input_text": chat_input.text,
"client_mode": chat_input.client_mode,
}
# 作业 6:应用服务注入的最新业务字段覆盖本轮输入,
# 使 set_info 写入后本轮 chat 立即可见(输入覆盖 checkpoint 同键)。
if chat_input.injected_fields:
graph_input.update(chat_input.injected_fields)
try:
return await self._graph.ainvoke(
graph_input,
config={"configurable": {"thread_id": chat_input.session_id}},
context=GraphDeps(llm=self._llm),
)
except LLMAuthError as exc:
raise ChatBackendAuthenticationError(str(exc)) from exc
except LLMRateLimitError as exc:
raise ChatBackendRateLimitError(str(exc)) from exc
async def stream(self, chat_input: ChatInput) -> AsyncIterator[ChatStreamEvent]:
"""流式:产生中立流事件。
TODO(作业 2): 实现该方法。最低要求:
- 先产出含完整 ``...`` 前缀的 ``TextDelta``,
再产出正文片段(保证 API 层的 stage 先于正文约束);
- 本阶段允许“完整答案切片段”,提交说明须标注
“尚未接入模型增量输出”(作业 8 改为真实增量)。
"""
result = await self.complete(chat_input)
prefix, sep, body = result.content.partition("")
if not sep:
yield TextDelta(result.content)
return
yield TextDelta(prefix + "")
if result.form_update:
yield FormUpdate(result.form_update)
if body:
yield TextDelta(body)