learn-langgraph · 04
条件分支：add_conditional_edges
顺序图里每条边都是固定的：A 之后必须走 B。但真实的 Agent 需要根据运行结果决定下一步——用户反馈是正面的走一条路，负面的走另一条路。
LangGraph 用 add_conditional_edges 来处理这种情况。
核心 API
graph.add_conditional_edges(
"source_node", # 从哪个节点出发
routing_function, # 路由函数：接收 state，返回下一个节点名
{
"route_a": "node_a", # 路由函数返回 "route_a" 时，走 node_a
"route_b": "node_b", # 路由函数返回 "route_b" 时，走 node_b
"end": END, # 路由函数返回 "end" 时，结束
}
)
路由函数就是一个普通 Python 函数：
def routing_function(state: MyState) -> str:
if state["some_condition"]:
return "route_a"
else:
return "route_b"
它接收当前 state，返回一个字符串 key，LangGraph 用这个 key 从映射表里找到下一个节点。
示例：用户反馈路由
场景：客服系统收到用户反馈，根据情绪（正面/负面/中性）走不同的处理流程。
用户输入
|
情绪分析
|
+-- 正面 --> 感谢回复
|
+-- 负面 --> 道歉 + 升级处理
|
+-- 中性 --> 标准回复
|
最终格式化输出
实现
from typing import TypedDict
from langgraph.graph import StateGraph, END
class FeedbackState(TypedDict):
user_input: str
sentiment: str # "positive" / "negative" / "neutral"
response: str
final_output: str
# ===== 节点 =====
def analyze_sentiment(state: FeedbackState) -> dict:
"""情绪分析（简化版，实际可以用 LLM）"""
text = state["user_input"].lower()
positive_words = ["好", "棒", "满意", "喜欢", "excellent", "great", "happy"]
negative_words = ["差", "烂", "不满", "投诉", "terrible", "bad", "angry"]
pos_count = sum(1 for w in positive_words if w in text)
neg_count = sum(1 for w in negative_words if w in text)
if pos_count > neg_count:
sentiment = "positive"
elif neg_count > pos_count:
sentiment = "negative"
else:
sentiment = "neutral"
return {"sentiment": sentiment}
def handle_positive(state: FeedbackState) -> dict:
"""处理正面反馈"""
response = f"感谢您的好评！很高兴我们的服务让您满意。您的反馈：'{state['user_input']}'"
return {"response": response}
def handle_negative(state: FeedbackState) -> dict:
"""处理负面反馈"""
response = (
f"非常抱歉给您带来了不好的体验！"
f"您的反馈已记录：'{state['user_input']}'。"
f"我们的高级客服将在 24 小时内联系您。"
)
return {"response": response}
def handle_neutral(state: FeedbackState) -> dict:
"""处理中性反馈"""
response = f"感谢您的反馈：'{state['user_input']}'。我们会继续改进服务。"
return {"response": response}
def format_output(state: FeedbackState) -> dict:
"""格式化最终输出"""
output = f"[情绪：{state['sentiment']}]\n{state['response']}"
return {"final_output": output}
# ===== 路由函数 =====
def route_by_sentiment(state: FeedbackState) -> str:
"""根据情绪返回路由 key"""
return state["sentiment"] # 直接返回 "positive" / "negative" / "neutral"
# ===== 组装图 =====
graph = StateGraph(FeedbackState)
graph.add_node("analyze", analyze_sentiment)
graph.add_node("positive_handler", handle_positive)
graph.add_node("negative_handler", handle_negative)
graph.add_node("neutral_handler", handle_neutral)
graph.add_node("format", format_output)
graph.set_entry_point("analyze")
# 条件分支：analyze 之后根据 sentiment 走不同节点
graph.add_conditional_edges(
"analyze",
route_by_sentiment,
{
"positive": "positive_handler",
"negative": "negative_handler",
"neutral": "neutral_handler",
}
)
# 三个分支最终都汇入 format
graph.add_edge("positive_handler", "format")
graph.add_edge("negative_handler", "format")
graph.add_edge("neutral_handler", "format")
graph.add_edge("format", END)
app = graph.compile()
运行
# 正面反馈
r = app.invoke({
"user_input": "产品很棒，服务也很满意！",
"sentiment": "",
"response": "",
"final_output": ""
})
print(r["final_output"])
# [情绪：positive]
# 感谢您的好评！很高兴我们的服务让您满意。...
# 负面反馈
r = app.invoke({
"user_input": "太差了，完全不满意，要投诉！",
"sentiment": "",
"response": "",
"final_output": ""
})
print(r["final_output"])
# [情绪：negative]
# 非常抱歉给您带来了不好的体验！...
图结构
START
|
analyze
|
+-- positive --> positive_handler --> format --> END
|
+-- negative --> negative_handler --> format --> END
|
+-- neutral --> neutral_handler --> format --> END
三条分支汇入同一个 format 节点——这叫 Fan-out + Fan-in，是多分支合并的标准模式。
配合 LLM 做情绪分析
实际项目里，analyze_sentiment 会调用 LLM：
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def analyze_sentiment(state: FeedbackState) -> dict:
prompt = f"""分析以下文本的情绪，只回复 "positive"、"negative" 或 "neutral" 之一。
文本：{state['user_input']}"""
response = llm.invoke([HumanMessage(content=prompt)])
sentiment = response.content.strip().lower()
# 容错处理
if sentiment not in ["positive", "negative", "neutral"]:
sentiment = "neutral"
return {"sentiment": sentiment}
路由函数本身不需要改，它只看 state 里的 sentiment 字段。节点负责更新状态，路由函数负责读状态做决策 ——这是 LangGraph 里关注点分离的核心思路。
Pydantic 结构化输出
如果你想用 Pydantic 让 LLM 输出更可靠：
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
class SentimentOutput(BaseModel):
sentiment: str # "positive" / "negative" / "neutral"
confidence: float # 0.0 ~ 1.0
reason: str # 判断原因
llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(SentimentOutput)
def analyze_sentiment(state: FeedbackState) -> dict:
result = structured_llm.invoke(
f"分析情绪：{state['user_input']}"
)
return {
"sentiment": result.sentiment,
"confidence": result.confidence,
}
with_structured_output 让 LLM 返回结构化的 Pydantic 对象，而不是纯文本。这样路由函数拿到的字段更可靠。
条件分支 vs if-else
你可能会问：直接在一个节点里写 if-else 不是一样的效果吗？
技术上可以，但有缺点：
- 可见性：
add_conditional_edges让图的结构可以被可视化，if-else 藏在函数里看不出来 - 可测试性：路由函数可以单独测试，不依赖完整的节点执行
- 可维护性：加一条新分支只需要加一个节点 + 在映射表里加一行，不用修改现有逻辑
小结
add_conditional_edges(source, routing_fn, mapping)是 LangGraph 的条件分支 API- 路由函数返回字符串 key，mapping 表把 key 映射到节点名
- 路由函数只读 state，不做操作——决策和操作分离
- 多个分支可以汇入同一个节点（Fan-in）
下一篇：多个节点并行执行：并行执行：Fan-out / Fan-in。