LangGraph 思维模式
学会用 LangGraph 的思维方式来构建 Agent。本文通过一个实际的客服邮件 Agent 案例，带你走完整的设计流程。
从流程开始
假设你要构建一个处理客服邮件的 AI Agent。需求如下：
Agent 需要：
- 阅读客户邮件
- 按紧急程度和主题分类
- 搜索文档回答问题
- 起草回复
- 将复杂问题升级给人工
- 安排后续跟进第 1 步：拆解为离散步骤
每个步骤对应一个节点：
mermaid
flowchart TD
A[START] --> B[Read Email]
B --> C[Classify]
C --> D[Doc Search]
C --> E[Human Review]
D --> F[Draft Reply]
E --> F
F --> G[Send]
G --> H[END]Read Email：提取邮件内容Classify：用 LLM 分类紧急程度和主题Doc Search：搜索知识库Draft Reply：生成回复Human Review：升级给人工Send：发送邮件
第 2 步：识别每个步骤的类型
LLM 步骤（需要理解/生成）
python
def classify_intent(state: State):
"""用 LLM 对邮件分类"""
prompt = f"""
分类以下邮件：
内容: {state['email']}
返回: urgent / normal / spam
"""
result = llm.invoke(prompt)
return {"classification": result.content.strip()}数据处理步骤（检索/查询）
python
def search_docs(state: State):
"""搜索知识库"""
results = vectorstore.similarity_search(state["email"], k=3)
return {"search_results": results}操作步骤（执行外部操作）
python
def send_email(state: State):
"""发送邮件"""
mail_client.send(to=state["customer_email"], content=state["draft"])
return {"sent": True, "sent_at": time.now()}第 3 步：设计 State
State 是所有节点共享的"记事本"：
python
from typing import TypedDict, Annotated, List
import operator
class EmailState(TypedDict):
# 输入
email: str
customer_email: str
# 处理结果
classification: str # urgent / normal / spam
search_results: Annotated[List[str], operator.add]
draft: str
# 状态追踪
needs_human: bool
sent: bool第 4 步：连接节点
python
from langgraph.graph import StateGraph, START, END
from typing import Literal
def route_from_classify(state: EmailState) -> Literal["search_docs", "human_review"]:
if state["classification"] == "urgent":
return "human_review" # 紧急邮件直接送人工
return "search_docs"
def route_from_review(state: EmailState) -> Literal["search_docs", "draft"]:
if state["needs_human"]:
return "search_docs" # 人工处理后继续
return "draft"
# 构建
builder = StateGraph(EmailState)
builder.add_node("read", read_email)
builder.add_node("classify", classify_intent)
builder.add_node("search_docs", search_docs)
builder.add_node("draft", draft_reply)
builder.add_node("human_review", human_review)
builder.add_node("send", send_email)
builder.add_edge(START, "read")
builder.add_edge("read", "classify")
builder.add_conditional_edges("classify", route_from_classify, ...)
builder.add_edge("search_docs", "draft")
builder.add_conditional_edges("human_review", route_from_review, ...)
builder.add_edge("draft", "send")
builder.add_edge("send", END)第 5 步：考虑边界情况
python
def robust_search(state: EmailState):
"""带容错的搜索"""
backends = [vectorstore, web_search, cache]
for backend in backends:
try:
results = backend.search(state["email"])
if results:
return {"search_results": results}
except Exception:
continue
return {"search_results": ["未找到相关文档"]}常见设计模式
问自己这些问题
- 这个步骤需要 LLM 吗？ → 需要理解/生成就用 LLM
- 数据来自哪里？ → 外部 API 需要容错
- 需要人工介入吗？ → 敏感操作加 interrupt
- 这个信息需要持久化吗？ → 放 State 里
- 失败后怎么办？ → 重试还是降级？
从哪里开始
| 场景 | 推荐路径 |
|---|---|
| 简单工具调用 | ReAct 模式 |
| 知识库问答 | RAG 模式 |
| 需要人工介入 | HITL 模式 |
| 复杂多步骤 | Orchestrator 模式 |
| 不确定 | 从 ReAct 开始，逐步加节点 |