基于 LangGraph 构建多 Agent 论文阅读系统

Sanqi_111

611人浏览 · 2026-04-12 18:53:26

Sanqi_111 · 2026-04-12 18:53:26 发布

文章目录

背景与动机

一、为什么是状态机，而不是函数调用链

二、状态设计：一个文件走天下

三、图的构建：add_edge 和 add_conditional_edges

四、Router：让 LLM 自己判断该走哪条路

五、双 LLM 后端兼容：给迁移留条后路

总结

背景与动机

项目最早是在 Dify 上搭的。那套方案的好处很直接——可视化拖拽，不写代码就能串起来一个多 Agent 工作流。前两周跑 demo 时一切都挺好，直到我在小组会上演示"切换论文重新深度阅读"这个场景时，workflow 编辑器前后卡了接近四十秒才把新的节点串起来，会议室里气氛尴尬。那次

演示之后我开始反思，这套架构到底适不适合我们的场景。列了一下发现的问题：

起步得拉十一个容器。本地机子开发时每次 `docker compose up` 要等两三分钟，Weaviate 那个 JVM 实例尤其吃资源。调一个 prompt 改一行字也得经历一整轮重启。

工作流逻辑存在 Dify 自己的 Postgres 里。改动无法进 Git，code review 时组员只能对着我的截图看——这跟我们整个项目的版本管理理念是冲突的。

想加一个新的节点类型（比如后面要做的幻觉检测节点），Dify 的插件机制要求把代码塞进一个单独的 plugin daemon 容器里，开发体验和写一个 Python 函数完全不是一回事。

流式输出能跑，但跨节点传递 token 需要手动处理 SSE 事件粘包，踩过两次坑。

最关键一点是，我们规划里后面要做 Self-RAG——一个"生成→验证→不达标就回退重写"的带环工作流。Dify 的 DAG 模型不支持带环的条件边，这等于直接卡死了后续规划。

换就换吧。选型阶段对比了 LangGraph、CrewAI 和自己手写状态机。CrewAI 更偏 role-play 风格，抽象层级太高；手写状态机灵活但得自己写持久化和可视化。LangGraph 的定位正好——它就是一个 StateGraph + 一堆 add_node / add_edge，没有多余的约束。

一、为什么是状态机，而不是函数调用链

一开始我倾向于把 Agent 系统做成一串函数调用：router → 选 agent → 跑 agent → 返回。但这套思路到了深度阅读 Agent 就撑不住了。深度阅读要跑三个阶段（信息抽取、批判分析、综合建议），每个阶段都要拿到前一阶段的完整结果作为上下文；之后还要经过幻觉检测，不合格的话要回到某个阶段重写。

如果用函数调用链来表达这个流程，我得手动维护一个叫 context 的 dict 在各函数之间传来传去，还得用循环加标志位来处理重试。代码会变成下面这种我最不想看到的样子：

# 反例：手写的重试逻辑

context = {"extraction": None, "analysis": None, "synthesis": None}

for attempt in range(3):

    context["extraction"] = run_extraction(paper, context)

    context["analysis"] = run_analysis(paper, context)

    context["synthesis"] = run_synthesis(paper, context)

    verdict = check_hallucination(context["synthesis"], paper)

    if verdict == "pass":

        break

    # 下一轮重试要保留 extraction / analysis 还是全部重来？

这种代码到第二个if分支就开始失控，每加一条重试规则就要动五处地方。LangGraph 的状态机模型在这里就显出价值：把"在哪里"和"做什么"分开——节点只管读状态、写状态，流转逻辑交给边。重试就是一条从 hallucination_check 指向 synthesis 的条件边，写在图定义里，一眼能看明白。

二、状态设计：一个文件走天下

LangGraph 用一个全局 State贯穿整个图。这个状态定义是整个系统的"契约"，所有节点都围绕它读写。我们的 PaperReadingState 最终定型如下：

class PaperReadingState(TypedDict, total=False):

    # 外部输入

    user_query: str

    paper_text: str

    selected_text: Optional[str]

    conversation_history: List[Dict[str, str]]

    paper_metadata: Dict[str, Any]        # 携带 forced_intent、doc_id 等



    # 控制字段（节点间协作用）

    intent: str

    current_agent: str

    agent_mode: Literal["streaming", "blocking"]



    # 输出字段

    agent_response: str

    stream_chunks: List[str]              # 流式 token 缓冲

    accumulated_results: Dict[str, str]   # 跨阶段结果累积



    # 幻觉检测相关

    hallucination_check: Optional[Literal["pass", "retry"]]

    halluc_iteration: int                 # 当前重试次数

    max_halluc_iterations: int            # 重试上限

```

一个容易踩的坑：TypedDict默认所有字段都是必填的，节点返回时哪怕只想更新一个字段，也得把整个字典填全。我把 total=False加上之后，节点就可以返回 {"intent": "deep_read"}这样的增量更新，LangGraph 内部会帮你 merge 到主状态。这个细节文档里藏得比较深，是调了半天才在 issue 里翻到的。

另一个设计决定是把 accumulated_results做成 dict 而不是 list。三个阶段的结果分别挂在 ["extraction"]、["analysis"]、["synthesis"] 键上，分析节点想读抽取结果直接 state["accumulated_results"]["extraction"]即可，不用关心下标。这在后面写 prompt 模板时省了很多力气。

三、图的构建：add_edge 和 add_conditional_edges

图定义函数叫 build_graph_v2，名字里带 v2 是因为第一版的图结构没考虑重试分支，被我自己推翻重做过一次。下面是核心代码：

def build_graph_v2(llm_client: Any, rag_service: Any) -> CompiledGraph:

    g = StateGraph(PaperReadingState)



    # 节点注册——用 lambda 把依赖注入进去

    g.add_node("router", lambda s: router_node(s, llm_client))

    g.add_node("qa_node", lambda s: qa_node(s, llm_client))

    g.add_node("critical_node", lambda s: critical_node(s, llm_client))

    g.add_node("extraction", lambda s: extraction_node(s, llm_client))

    g.add_node("analysis", lambda s: analysis_node(s, llm_client))

    g.add_node("synthesis", lambda s: synthesis_node(s, llm_client))

    g.add_node("hallucination_check",

               lambda s: hallucination_check_node(s, llm_client, rag_service))

    g.add_node("questioner", lambda s: questioner_node(s, llm_client))



    g.set_entry_point("router")



    # 第一次分叉：按 intent 路由到对应 agent

    g.add_conditional_edges("router", lambda s: s["intent"], {

        "chat":      "qa_node",

        "extract":   "qa_node",

        "recommend": "qa_node",

        "critical":  "critical_node",

        "deep_read": "extraction",

        # translate / literature_review 省略

    })



    # 深度阅读线性段

    g.add_edge("extraction", "analysis")

    g.add_edge("analysis",   "synthesis")

    g.add_edge("synthesis",  "hallucination_check")



    # 第二次分叉：幻觉检测后决定重试还是放行

    g.add_conditional_edges(

        "hallucination_check",

        _hallucination_route_unified,   # 返回 "retry" 或 "pass"

        {"retry": "synthesis", "pass": "questioner"}

    )



    g.add_edge("questioner", END)

    return g.compile()

这里有两个细节值得单独拎出来说。

第一个是依赖注入用 lambda包一层。LangGraph 的节点函数签名要求是 (state) -> dict，但我们的节点实际上还需要 llm_client 和 rag_service。直接把它们做成全局变量？不行，测试时换 mock 就得打猴补丁。用 functools.partial？可以但不够直观。最后用 lambda s: router_node(s, llm_client) 这种形式最清爽——图定义时把依赖捕获进闭包，节点函数本身保持纯净。

第二个是 add_conditional_edges 的签名里有三个参数：源节点、路由函数、分支字典。路由函数的返回值必须是分支字典里的 key 之一，否则运行时会报一个不太友好的 KeyError。为了避免这个坑，我把所有路由函数都统一收在 hallucination_route_unified 这种命名里，并且在函数体里强制断言返回值只能是 "pass" 或 "retry"：

def _hallucination_route_unified(state: PaperReadingState) -> str:

    result = state.get("hallucination_check")

    iteration = state.get("halluc_iteration", 0)

    max_iter = state.get("max_halluc_iterations", 2)



    if result == "pass":

        return "pass"

    if result == "retry" and iteration < max_iter:

        return "retry"

    # 兜底：重试超限或字段缺失也走 pass，避免死循环

    return "pass"

最后那行兜底是血的教训。第一版没写这一行，结果某次 DeepSeek 返回的格式偏差让 hallucination_check 留空，图进入了无限重试，占满了 API 额度才发现。从那以后我的原则是：LangGraph 里任何条件边都必须有兜底出口，不然一个偶然的字段缺失就能把整个服务拖垮。

四、Router：让 LLM 自己判断该走哪条路

Router 节点是系统的入口。它的职责很简单：读用户输入，决定接下来该跑哪个 Agent。我最开始想用关键词匹配（"帮我翻译" → translate、"评估" → critical），写了二十多行 if-else 之后放弃了——用户的表达方式太灵活，规则写不完。

换成 LLM 分类后代码反而更短：

INTENT_LIST = ["chat", "critical", "deep_read", "extract",

               "translate", "recommend", "literature_review"]



def router_node(state: PaperReadingState, llm_client: Any) -> dict:

    # 前端直接指定 intent 时跳过 LLM 调用（省 token、省时间）

    forced = state.get("paper_metadata", {}).get("forced_intent")

    if forced in INTENT_LIST:

        return {"intent": forced, "current_agent": forced}



    template = load_prompt("smart_router.md")

    prompt = template.format(

        user_query=state["user_query"],

        selected_text=state.get("selected_text") or "（无）",

    )

    full_text, _ = call_llm_blocking(llm_client, prompt)

    intent = _parse_intent(full_text)

    return {"intent": intent, "current_agent": intent}

两点值得说：

1. forced_intent是给前端"点按钮就走特定 Agent"用的。用户在聊天界面点了"批判分析"按钮，前端会在请求里带上 `paper_metadata.forced_intent = "critical"`，router 检测到之后直接跳过 LLM 分类。这一招让点按钮场景的响应时间从大约 800ms 降到了 50ms——省掉的那一次 LLM 调用是真金白银。

2. _parse_intent做了宽松解析。DeepSeek 有时候会在 JSON 前后加一段"以下是分类结果："的话，或者把 JSON 外面包一层 markdown 代码块。parse 函数做了三层 fallback：先按纯 JSON 解析；失败则正则抓 "intent":\s*"(\w+)"；再失败就在文本里找 7 个意图关键词的第一个匹配。这个宽容度在实际运行里救过不少次。

Prompt 模板 smart_router.md 里最关键的一段是：

请将下面的用户查询分类到以下 7 种意图之一（严格只输出 JSON）：

- chat: 一般问答、事实查询、概念解释

- critical: 要求批判评估、指出问题、分析方法论局限

- deep_read: 要求对整篇论文做全面系统的解读

- extract: 要求提取关键数据、方法、结果等结构化信息

- translate: 要求翻译论文或其中段落

- recommend: 要求根据研究方向推荐相关论文

- literature_review: 要求综合多篇论文做综述

输出格式：{"intent": "<意图名>", "reason": "<10 字内理由>"}

"严格只输出 JSON" 这一行是加了之后才稳定的，之前用 DeepSeek-chat 经常输出自然语言解释。加上 reason 字段是个小心机——强迫模型"先想一下再决定"能显著降低误判率，这跟 CoT 的原理一致。

五、双 LLM 后端兼容：给迁移留条后路

最后说一下 LLM 客户端的抽象。在把 Dify 彻底拆下来之前，我想要一段时间的"双轨运行"——Agent 节点代码既能跑在新的 DeepSeek 直连客户端上，也能退回到老的 Dify 客户端。这样一旦 DeepSeek 直连出现问题（比如限流、格式不一致），我可以通过环境变量一键切换，不用改代码。

两种客户端的接口完全不一样：

DirectLLMClient.chat_streaming(query, system_prompt)直接返回 token 生成器

DifyClient.chat(query, conversation_id, stream=True) 返回事件字典生成器，token 藏在 event["answer"]` 里

为了让 Agent 节点不关心这两者差异，我在 agents/nodes/_helpers.py里写了一层统一的 call_llm_streaming：

def _is_direct_client(client: Any) -> bool:

    return hasattr(client, "chat_streaming")



def call_llm_streaming(

    client: Any,

    query: str,

    system_prompt: Optional[str] = None,

    conversation_id: Optional[str] = None,

) -> Tuple[str, List[str]]:

    chunks: List[str] = []

    if _is_direct_client(client):

        for token in client.chat_streaming(query, system_prompt=system_prompt):

            chunks.append(token)

    else:

        for event in client.chat(query=query,

                                  conversation_id=conversation_id,

                                  stream=True):

            answer = event.get("answer", "")

            if answer:

                chunks.append(answer)

    return "".join(chunks), chunks

Duck typing 这里比 isinstance(client, DirectLLMClient) 更好用，原因是测试时我经常用 `Mock()` 对象替身，加 chat_streaming 属性比继承真类方便得多。

切换开关放在 orchestrator.py 里，读 USE_DIRECT_LLM 环境变量：

if os.getenv("USE_DIRECT_LLM", "true").lower() == "true":

    self.llm_client = DirectLLMClient(api_key=DEEPSEEK_KEY)

else:

    self.llm_client = DifyClient(base_url=DIFY_URL, api_key=DIFY_KEY)

这个开关的意义，不是为了留恋 Dify，而是为了在迁移期让故障有兜底——上线第一周，我就靠这个开关救过一次场，当时 DeepSeek 某个区域节点抽风，切回 Dify 支撑了二十分钟等对方恢复。

总结

这一周的产出按代码量算不算多：

agents/langgraph_engine.py 366 行，包含 build_graph_v2 和两条路由函数

agents/orchestrator.py 272 行，负责客户端选择、图编译缓存、请求分发

agents/nodes/router.py 54 行，就是上面那个 router 节点

agents/nodes/_helpers.py 79 行，双客户端兼容层

prompts/smart_router.md、agents/state.py 两个小文件

但真正花时间的不是写代码，是调试那些不直观的 LangGraph 细节:total=False 的发现、条件边兜底的加入、lambda 捕获依赖的尝试、Mermaid 可视化的调通。每一个坑踩下去都是一到两个小时。

下周的任务是把另外六个 Agent 节点全部实现出来（QA、批判、翻译、推荐、三阶段深度阅读、延伸提问），以及写完全部 16 个 prompt 模板。深度阅读的三阶段管线是下周的重头戏，因为它是第一个真正意义上用到 LangGraph "多步流水线 + 条件回退" 能力的场景。

这一周最后一天，我们三个人做了一次端到端联调。成员 B 把 MinerU 解析后的论文文本通过 RAGService.ingest() 写入 ChromaDB，拿到 doc_id；成员 C 的前端用这个 doc_id 发起 /api/agents/chat/run请求；我的 LangGraph 引擎从 router 开始路由，一路跑到生成节点，最后通过 SSE 把 token 流推回前端。

第一次跑通时出了一个很典型的对接问题——前端传进来的 document_id 字段在我的 state schema 里叫 paper_metadata.doc_id，name mismatch 让检索节点拿到空字符串，直接从 ChromaDB 返回空列表。后来我在 api/routes/agents.py 的请求预处理里加了一层字段映射，把 HTTP 请求的 document_id 统一翻译成 state 里的 paper_metadata.doc_id。

这个小插曲暴露了一个更普遍的教训：三个人分别在后端、数据、前端埋头写代码时，对同一个字段的命名心智模型完全不同。后端叫 doc_id，前端叫 documentId（驼峰），HTTP 层叫 document_id。下周我打算在 services/models.py 里用 Pydantic 做一层统一的 DTO，前端 → HTTP → state 的字段转换全部集中在一处，而不是散落在各个节点里。

跑通之后的初步效果：

首次冷启动到第一个 token 出现：约 1.2 秒（router 分类 0.6s + 首个 LLM token 0.6s）

简单 chat 问答：平均 4-6 秒全部响应完

深度阅读：一篇 20 页英文论文，总耗时 40-60 秒，能稳定产出结构化的三阶段分析

幻觉检测：触发率约 20%，其中 90% 的重试能在第一次修正后通过

离"生产可用"还差得远，但至少证明了整套 LangGraph + 双 LLM + Self-RAG 的架构是跑得动的。下周会把剩余的 Agent 节点补齐，然后用一批真实论文做批量测试。