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).
This commit is contained in:
@@ -0,0 +1,65 @@
|
|||||||
|
"""LangGraph implementation of the backend-neutral chat contract(作业 2)。
|
||||||
|
|
||||||
|
实现 ``ChatBackend`` Protocol,把图执行结果翻译为公共契约:
|
||||||
|
- ``complete()`` 返回 ``ChatResult``,``content`` 保留
|
||||||
|
``<state>XXXX</state>`` 前缀(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): 实现该方法。最低要求:
|
||||||
|
- 先产出含完整 ``<state>...</state>`` 前缀的 ``TextDelta``,
|
||||||
|
再产出正文片段(保证 API 层的 stage 先于正文约束);
|
||||||
|
- 本阶段允许“完整答案切片段”,提交说明须标注
|
||||||
|
“尚未接入模型增量输出”(作业 8 改为真实增量)。
|
||||||
|
"""
|
||||||
|
result = await self.complete(chat_input)
|
||||||
|
prefix, _, body = result.content.partition("</state>")
|
||||||
|
yield TextDelta(prefix + "</state>")
|
||||||
|
if body:
|
||||||
|
yield TextDelta(body)
|
||||||
@@ -15,9 +15,21 @@ class Config:
|
|||||||
FASTGPT_BASE_URL = os.getenv("ANALYSIS_SERVICE_URL")
|
FASTGPT_BASE_URL = os.getenv("ANALYSIS_SERVICE_URL")
|
||||||
FASTGPT_APP_ID = os.getenv("APP_ID")
|
FASTGPT_APP_ID = os.getenv("APP_ID")
|
||||||
|
|
||||||
|
# Chat backend switch(作业 2)
|
||||||
|
# "fastgpt" 保持现有行为;"langgraph" 使用内存图 fake 回复。
|
||||||
|
AGENT_BACKEND = os.getenv("AGENT_BACKEND", "fastgpt")
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def validate(cls):
|
def validate(cls):
|
||||||
"""Validate required configuration."""
|
"""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 = [
|
required = [
|
||||||
("FASTGPT_BASE_URL (ANALYSIS_SERVICE_URL)", cls.FASTGPT_BASE_URL),
|
("FASTGPT_BASE_URL (ANALYSIS_SERVICE_URL)", cls.FASTGPT_BASE_URL),
|
||||||
("FASTGPT_API_KEY (ANALYSIS_AUTH_TOKEN)", cls.FASTGPT_API_KEY),
|
("FASTGPT_API_KEY (ANALYSIS_AUTH_TOKEN)", cls.FASTGPT_API_KEY),
|
||||||
|
|||||||
+18
-11
@@ -2,8 +2,10 @@
|
|||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
from fastapi import FastAPI
|
from fastapi import FastAPI
|
||||||
from fastgpt_client import AsyncChatClient
|
from fastgpt_client import AsyncChatClient
|
||||||
|
from ..agent.graph import build_graph
|
||||||
from ..backends.chat import ChatBackend
|
from ..backends.chat import ChatBackend
|
||||||
from ..backends.fastgpt import FastGPTBackend
|
from ..backends.fastgpt import FastGPTBackend
|
||||||
|
from ..backends.langgraph import LangGraphBackend
|
||||||
from .config import Config
|
from .config import Config
|
||||||
|
|
||||||
|
|
||||||
@@ -13,20 +15,25 @@ _chat_backend: ChatBackend | None = None
|
|||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def lifespan(app: FastAPI):
|
async def lifespan(app: FastAPI):
|
||||||
"""Manage FastGPT client lifecycle."""
|
"""Manage chat backend resource lifecycle."""
|
||||||
global _chat_backend, _fastgpt_client
|
global _chat_backend, _fastgpt_client
|
||||||
Config.validate()
|
Config.validate()
|
||||||
|
|
||||||
_fastgpt_client = AsyncChatClient(
|
if Config.AGENT_BACKEND == "langgraph":
|
||||||
api_key=Config.FASTGPT_API_KEY,
|
# 作业 2:LangGraph fake 模式不要求 FastGPT 密钥。
|
||||||
base_url=Config.FASTGPT_BASE_URL,
|
# TODO(作业 2): 若语音仍启用,需在此单独初始化并校验 FastGPT 配置。
|
||||||
timeout=60.0,
|
_chat_backend = LangGraphBackend(build_graph())
|
||||||
max_retries=3,
|
else:
|
||||||
retry_delay=1.0,
|
_fastgpt_client = AsyncChatClient(
|
||||||
enable_logging=Config.DEBUG,
|
api_key=Config.FASTGPT_API_KEY,
|
||||||
)
|
base_url=Config.FASTGPT_BASE_URL,
|
||||||
await _fastgpt_client.__aenter__()
|
timeout=60.0,
|
||||||
_chat_backend = FastGPTBackend(_fastgpt_client)
|
max_retries=3,
|
||||||
|
retry_delay=1.0,
|
||||||
|
enable_logging=Config.DEBUG,
|
||||||
|
)
|
||||||
|
await _fastgpt_client.__aenter__()
|
||||||
|
_chat_backend = FastGPTBackend(_fastgpt_client)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
yield
|
yield
|
||||||
|
|||||||
@@ -0,0 +1,62 @@
|
|||||||
|
"""作业 2 验收测试:LangGraph 后端适配器。
|
||||||
|
|
||||||
|
通过标准(对应迁移计划作业 2):
|
||||||
|
- ``complete()`` 返回带 ``<state>1002</state>`` 前缀的 ``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("<state>1002</state>")
|
||||||
|
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("<state>1002</state>")
|
||||||
|
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
|
||||||
Reference in New Issue
Block a user