描述一个你认为最适合用 LangGraph 构建的复杂业务场景,你会如何设计它的 State 结构,规划哪些核心的 Node 和 Edge,并处理潜在的异常流转?
我认为最适合用 LangGraph 构建的复杂业务场景是:企业级智能招投标书(RFP)多智能体协同撰写与审核系统。
为什么这个场景最适合 LangGraph?
撰写大型投标书通常需要拆解任务、检索公司知识库、多轮起草与修改(循环)、合规性审查,以及关键节点的“人类介入(Human-in-the-loop)”。传统的线性 Chain 无法处理“起草-审查-打回修改”的循环逻辑,而基于提示词的 AutoGen 缺乏对流程和状态的精确控制。LangGraph 的状态机(State Graph)和循环控制能力完美契合这一需求。
一、 State(状态)结构设计
在 LangGraph 中,State 贯穿所有节点。针对这个场景,我们需要记录原始需求、大纲、草稿、审查反馈、执行次数和异常信息。
python
from typing import Annotated, List, Dict, TypedDict
import operator
class RFPGenerationState(TypedDict):
# 1. 基础输入
rfp_document: str # 客户原始招标文件内容
company_kb_context: str # 检索到的公司内部知识库上下文
# 2. 规划与拆解
extracted_requirements: List[str] # 提取出的硬性指标和得分点
outline: List[Dict[str, str]] # 投标书大纲(章节名及核心要点)
# 3. 核心流转数据
# 使用 Annotated 和 operator.set/add 来定义状态的更新逻辑
current_draft: Annotated[Dict[str, str], operator.ior] # 当前各章节草稿合并
review_feedback: str # 审查者的修改意见
# 4. 循环控制与异常处理
iteration_count: int # 记录修改轮次,防止无限死循环
status: str # 状态流转标识 (e.g., "drafting", "reviewing", "approved", "error")
error_logs: Annotated[List[str], operator.add] # 收集各节点的异常信息
二、 核心 Node(节点)规划
我们将系统拆分为扮演不同角色的 Agent Node:
Information_Extractor(需求拆解者):- 分析客户 RFP 文件,提取交期、技术资质、报价要求等硬性指标。
Planner(架构规划者):- 根据提取的需求,生成投标书大纲(包含章节标题和每章需涵盖的要点)。
Researcher(知识检索者 - Tool Agent):- 根据大纲,调用企业内部知识库(RAG/VectorDB)、历史成功标书、ERP 系统等,检索支撑材料。
Drafter(主笔起草者):- 根据大纲和检索到的知识,按章节撰写投标书正文。
Reviewer_Critic(合规与质量审查者):- 核心逻辑:将生成的草稿与
extracted_requirements进行对比打分。 - 如果发现遗漏或语气不专业,生成具体的
review_feedback,并将status设为needs_revision。 - 如果完全合规,将
status设为approved。
- 核心逻辑:将生成的草稿与
Human_Intervention(人类审核节点 - HITL):- LangGraph 可以在此节点暂停(interrupt),让人类产品经理或销售总监查看最终草稿或解决系统无法处理的死循环。
三、 核心 Edge(边与路由)规划
工作流中包含直线流转和条件流转(Conditional Edges):
plaintext
graph TD
START((START)) --> Extractor(Information_Extractor)
Extractor --> Planner(Planner)
Planner --> Researcher(Researcher)
Researcher --> Drafter(Drafter)
Drafter --> Reviewer(Reviewer_Critic)
Reviewer --> Condition{审核结果与轮次判断}
Condition -- "Approved (通过)" --> Human(Human_Intervention)
Condition -- "Needs Revision & 轮次 < 3" --> Drafter
Condition -- "Needs Revision & 轮次 >= 3" --> Human
Condition -- "Error (系统异常)" --> Error_Handler(Error_Recovery)
Error_Handler --> Human
Human --> END((END))
关键条件路由逻辑 (routing_logic 函数):
python
def review_routing(state: RFPGenerationState) -> str:
if state["status"] == "approved":
return "Human_Intervention"
elif state["status"] == "needs_revision":
if state["iteration_count"] >= 3:
return "Human_Intervention" # 达到最大修改次数,强制转人工
else:
return "Drafter" # 打回重新起草
elif state["status"] == "error":
return "Error_Recovery"
四、 潜在异常流转与处理机制(Robustness)
在复杂的业务场景中,必须考虑到大模型幻觉、API 失败或逻辑死循环。通过 LangGraph 的状态管理,可以优雅地处理这些异常:
1. 异常场景一:无限修改死循环 (Infinite Loop)
- 现象:
Reviewer总是觉得Drafter写的不好,两者陷入无限相互反驳。 - 处理策略:
- 在
Reviewer节点中,每次执行强制state["iteration_count"] += 1。 - 在条件边中设置硬性阈值(如
iteration_count >= 3)。一旦触发,系统暂停,保留当前最好的草稿及审查意见,流转至Human_Intervention节点,交由人类决定是手动修改还是调整大纲。
- 在
2. 异常场景二:知识库未检索到关键信息 (Missing Context)
- 现象:客户要求提供某项 ISO 资质,但
Researcher节点在知识库没查到,导致Drafter开始胡编乱造(幻觉)。 - 处理策略:
- 在
Researcher节点配置 Tool 调用检查。如果检索结果为空,主动将状态置为:error_logs.append("Missing ISO Certification")。 - 此时工作流不进入
Drafter,而是通过条件边直接触发Human_Intervention,向前端抛出提示:“请人工上传相关资质文件”,待人工补充数据更新 State 后,再继续执行Drafter。
- 在
3. 异常场景三:大模型 API 超时或上下文超载 (Context Window Exceeded)
- 现象:投标书过长,合并后的草稿超出了 LLM 的 Token 限制,或者 API 突然 502。
- 处理策略:
- 利用 LangChain 底层的
.with_fallbacks()机制。如果主要模型(如 GPT-4o)超时,自动降级切换到备用模型(如 Claude-3-Haiku)尝试。 - 如果底层报错,
try-except捕获异常,将其写入state["error_logs"],并将状态流转至专门的Error_Recovery节点。该节点可以尝试对文本进行截断、摘要压缩,然后重新提交。
- 利用 LangChain 底层的
总结
这种设计充分利用了 LangGraph 的三个核心优势:
- Stateful(状态持久化):长文本生成过程中,随时可以暂停、恢复,不怕断点。
- Cyclical(循环控制):实现了 Draft-Review 的多轮自我博弈优化。
- Human-in-the-loop(人类介入):在关键决策点(合规失败、超轮次、最终定稿)完美融合机器的效率与人类的判断力。