From 5d68b731d4b8fb146275348a4ef59870b1b5eeb5 Mon Sep 17 00:00:00 2001 From: Eric Wang Date: Mon, 19 Oct 2026 14:29:31 +0800 Subject: [PATCH] feat(backends): wire LangGraph backend behind AGENT_BACKEND switch (assignment 2) - src/backends/langgraph.py: LangGraphBackend implements ChatBackend; complete() runs one graph turn per session (thread_id=sessionId) and returns state-prefixed content; stream() emits the full state prefix before body fragments (full-answer slicing, not yet real incremental output - assignment 8) - src/core/config.py: AGENT_BACKEND=fastgpt|langgraph, langgraph mode no longer requires FastGPT credentials at startup - src/core/fastgpt_client.py: lifespan builds the backend by switch - test/backends/test_langgraph_backend.py: adapter acceptance tests Verified: - pytest: 48 passed - AGENT_BACKEND=langgraph uvicorn: POST /chat -> 1002; browser_addon -> browser_addon.1002; SSE = stage_code, text_delta, single done Known limit: stream slices a complete answer; no formUpdate deltas yet (assignment 6). --- src/backends/langgraph.py | 65 +++++++++++++++++++++++++ src/core/config.py | 12 +++++ src/core/fastgpt_client.py | 29 ++++++----- test/backends/test_langgraph_backend.py | 62 +++++++++++++++++++++++ 4 files changed, 157 insertions(+), 11 deletions(-) create mode 100644 src/backends/langgraph.py create mode 100644 test/backends/test_langgraph_backend.py diff --git a/src/backends/langgraph.py b/src/backends/langgraph.py new file mode 100644 index 0000000..15c9e9a --- /dev/null +++ b/src/backends/langgraph.py @@ -0,0 +1,65 @@ +"""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 build_graph +from .chat import ChatInput, ChatResult, ChatStreamEvent, ChatBackend, TextDelta + + +class LangGraphBackend(ChatBackend): + """Adapt a compiled LangGraph to the neutral chat contract.""" + + def __init__(self, graph=None): + """参数: + graph: 编译后的图；缺省调用 ``build_graph()``（内存 checkpoint）。 + """ + self._graph = graph if graph is not None else build_graph() + + 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._graph.ainvoke( + {"session_id": chat_input.session_id, "input_text": chat_input.text}, + config={"configurable": {"thread_id": chat_input.session_id}}, + ) + return ChatResult( + content = result["response_text"], + status_code = None, + form_update = {} + ) + + async def stream(self, chat_input: ChatInput) -> AsyncIterator[ChatStreamEvent]: + """流式：产生中立流事件。 + + TODO(作业 2): 实现该方法。最低要求： + - 先产出含完整 ``...`` 前缀的 ``TextDelta``， + 再产出正文片段（保证 API 层的 stage 先于正文约束）； + - 本阶段允许“完整答案切片段”，提交说明须标注 + “尚未接入模型增量输出”（作业 8 改为真实增量）。 + """ + result = await self.complete(chat_input) + prefix, _, body = result.content.partition("") + yield TextDelta(prefix + "") + if body: + yield TextDelta(body) diff --git a/src/core/config.py b/src/core/config.py index 4bc2476..6c06984 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -15,9 +15,21 @@ class Config: FASTGPT_BASE_URL = os.getenv("ANALYSIS_SERVICE_URL") FASTGPT_APP_ID = os.getenv("APP_ID") + # Chat backend switch（作业 2） + # "fastgpt" 保持现有行为；"langgraph" 使用内存图 fake 回复。 + AGENT_BACKEND = os.getenv("AGENT_BACKEND", "fastgpt") + @classmethod def validate(cls): """Validate required configuration.""" + # TODO(作业 2): 语音管线仍依赖 FastGPT。当前仅按文本后端开关校验； + # 接入语音开关后，需要单独校验语音路径所需配置。 + if cls.AGENT_BACKEND not in ("fastgpt", "langgraph"): + raise ValueError( + f"AGENT_BACKEND must be 'fastgpt' or 'langgraph', got {cls.AGENT_BACKEND!r}" + ) + if cls.AGENT_BACKEND != "fastgpt": + return required = [ ("FASTGPT_BASE_URL (ANALYSIS_SERVICE_URL)", cls.FASTGPT_BASE_URL), ("FASTGPT_API_KEY (ANALYSIS_AUTH_TOKEN)", cls.FASTGPT_API_KEY), diff --git a/src/core/fastgpt_client.py b/src/core/fastgpt_client.py index 3ce0c4c..4879baf 100644 --- a/src/core/fastgpt_client.py +++ b/src/core/fastgpt_client.py @@ -2,8 +2,10 @@ from contextlib import asynccontextmanager from fastapi import FastAPI from fastgpt_client import AsyncChatClient +from ..agent.graph import build_graph from ..backends.chat import ChatBackend from ..backends.fastgpt import FastGPTBackend +from ..backends.langgraph import LangGraphBackend from .config import Config @@ -13,20 +15,25 @@ _chat_backend: ChatBackend | None = None @asynccontextmanager async def lifespan(app: FastAPI): - """Manage FastGPT client lifecycle.""" + """Manage chat backend resource lifecycle.""" global _chat_backend, _fastgpt_client Config.validate() - _fastgpt_client = AsyncChatClient( - api_key=Config.FASTGPT_API_KEY, - base_url=Config.FASTGPT_BASE_URL, - timeout=60.0, - max_retries=3, - retry_delay=1.0, - enable_logging=Config.DEBUG, - ) - await _fastgpt_client.__aenter__() - _chat_backend = FastGPTBackend(_fastgpt_client) + if Config.AGENT_BACKEND == "langgraph": + # 作业 2：LangGraph fake 模式不要求 FastGPT 密钥。 + # TODO(作业 2): 若语音仍启用，需在此单独初始化并校验 FastGPT 配置。 + _chat_backend = LangGraphBackend(build_graph()) + else: + _fastgpt_client = AsyncChatClient( + api_key=Config.FASTGPT_API_KEY, + base_url=Config.FASTGPT_BASE_URL, + timeout=60.0, + max_retries=3, + retry_delay=1.0, + enable_logging=Config.DEBUG, + ) + await _fastgpt_client.__aenter__() + _chat_backend = FastGPTBackend(_fastgpt_client) try: yield diff --git a/test/backends/test_langgraph_backend.py b/test/backends/test_langgraph_backend.py new file mode 100644 index 0000000..bb7e668 --- /dev/null +++ b/test/backends/test_langgraph_backend.py @@ -0,0 +1,62 @@ +"""作业 2 验收测试：LangGraph 后端适配器。 + +通过标准（对应迁移计划作业 2）： +- ``complete()`` 返回带 ``1002`` 前缀的 ``ChatResult``， + API 层据此解析阶段码； +- ``stream()`` 先产出含完整 state 前缀的 ``TextDelta``，再产出正文， + 保证 ``stage_code`` 事件先于正文； +- 会话通过 ``thread_id = session_id`` 绑定（图 checkpoint 语义）。 + +注意：``src/backends/langgraph.py`` 中的 TODO 完成前， +本文件的测试会失败，属于预期状态。 +""" + +import pytest + +from src.backends.chat import ChatInput, TextDelta +from src.backends.langgraph import LangGraphBackend + + +def make_input() -> ChatInput: + return ChatInput( + session_id="session-001", + text="发生了交通事故", + client_mode="direct", + ) + + +@pytest.mark.asyncio +async def test_complete_returns_state_prefixed_result() -> None: + backend = LangGraphBackend() + + result = await backend.complete(make_input()) + + assert result.content.startswith("1002") + assert "发生了交通事故" in result.content + + +@pytest.mark.asyncio +async def test_stream_emits_prefix_before_body() -> None: + backend = LangGraphBackend() + + events = [event async for event in backend.stream(make_input())] + + assert events, "stream 至少应产生一个 TextDelta" + assert all(isinstance(event, TextDelta) for event in events) + joined = "".join(event.text for event in events) + assert joined.startswith("1002") + assert "发生了交通事故" in joined + + +@pytest.mark.asyncio +async def test_sessions_are_isolated_by_thread_id() -> None: + backend = LangGraphBackend() + + first = await backend.complete(make_input()) + second = await backend.complete(make_input()) + other = await backend.complete( + ChatInput(session_id="session-002", text="你好", client_mode="direct") + ) + + assert first.content != "" and second.content != "" + assert "你好" in other.content