LangGraph 的 常见状态图（StateGraph）
示例 1：基础顺序链（Linear Chain）
像工厂流水线，A -> B -> C 一步步往下走。
📊 Mermaid 流程图
graph TD;
START([开始]) --> clean[清洗数据];
clean --> format[格式化输出];
format --> END([结束]);🖥️ 运行生成 ASCII 图
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
text: str
def clean(state): return {"text": state["text"].strip()}
def format(state): return {"text": f"Result: {state['text']}"}
builder = StateGraph(State)
builder.add_node("clean", clean).add_node("format", format)
builder.add_edge(START, "clean").add_edge("clean", "format").add_edge("format", END)
graph = builder.compile()
# 打印控制台流程图
print(graph.get_graph().draw_ascii())
# 输出：
# +-----------+ +----------+ +------+
# | __start__ | --> | clean | --> | format | --> | __end__ |
# +-----------+ +----------+ +------+
# 生成 Mermaid 代码（可复制到支持 Mermaid 的工具查看）
print(graph.get_graph().draw_mermaid())示例 2：条件路由（Conditional Branching）
类似“智能分拣”，根据业务规则（如关键词、分数）走不同的分支。
📊 Mermaid 流程图
graph TD;
START([开始]) --> classify[分类器];
classify -->|"类别: 数学"| math_node[数学处理器];
classify -->|"类别: 闲聊"| chat_node[闲聊回复];
math_node --> END([结束]);
chat_node --> END([结束]);🖥️ 完整带图代码
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
query: str
category: str
def classify(state):
cat = "math" if "计算" in state["query"] else "chat"
return {"category": cat}
def do_math(state): return {"response": "数学答案: 42"}
def do_chat(state): return {"response": "闲聊: 今天天气不错"}
def route_condition(state) -> Literal["math_node", "chat_node"]:
return "math_node" if state["category"] == "math" else "chat_node"
builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("math_node", do_math)
builder.add_node("chat_node", do_chat)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route_condition) # 条件分支
builder.add_edge("math_node", END)
builder.add_edge("chat_node", END)
graph = builder.compile()
# 打印出带有分支箭头的 ASCII 图
print(graph.get_graph().draw_ascii())
# 输出会清晰显示 classify 分叉到两个节点示例 3：ReAct 代理循环（Agent Loop）
最经典的图：Agent 思考 -> 调用工具 -> 观察结果 -> 再思考... 直到结束。图中包含自循环。
📊 Mermaid 流程图
graph TD;
START([开始]) --> agent[Agent (LLM)];
agent -->|"有工具调用"| tools[执行工具];
agent -->|"无工具调用"| END([结束]);
tools --> agent;
style tools fill:#f9f,stroke:#333;
style agent fill:#bbf,stroke:#333;🖥️ 完整带图代码（使用内置工具条件）
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
# 模拟工具
search = DuckDuckGoSearchRun()
tools = [search]
model = ChatOpenAI(model="gpt-4o").bind_tools(tools)
def call_model(state: MessagesState):
return {"messages": [model.invoke(state["messages"])]}
builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition) # 内置判断：去 tools 或结束
builder.add_edge("tools", "agent") # 关键：从 tools 回到 agent（形成循环）
graph = builder.compile()
# 打印循环流程图
print(graph.get_graph().draw_ascii())
# 输出：
# +-----------+ +-------+ +-------+
# | __start__ | --> | agent | --> | tools |
# +-----------+ +-------+ +-------+
# ^ | |
# | +-----------+
# | (agent 无工具调用时直接结束)示例 4：并行分发（Map-Reduce / Send API）
一个节点分发多个并行任务，所有任务完成后自动汇总。
📊 Mermaid 流程图
graph TD;
START([开始]) --> splitter[任务分发器];
splitter -->|"Send(worker, url1)"| worker1[Worker 1];
splitter -->|"Send(worker, url2)"| worker2[Worker 2];
splitter -->|"Send(worker, url3)"| worker3[Worker 3];
worker1 --> aggregator[汇总器];
worker2 --> aggregator;
worker3 --> aggregator;
aggregator --> END([结束]);🖥️ 完整带图代码
import operator
from typing import Annotated, List, TypedDict
from langgraph.graph import StateGraph, START, END, Send
class OverallState(TypedDict):
urls: List[str]
collected: Annotated[list, operator.add]
def splitter(state):
# 动态分发并行任务
return [Send("worker", {"url": url}) for url in state["urls"]]
def worker(state):
return {"collected": [f"抓取结果: {state['url']}"]}
def aggregator(state):
return {"final": f"总数据量: {len(state['collected'])}"}
builder = StateGraph(OverallState)
builder.add_node("splitter", splitter)
builder.add_node("worker", worker)
builder.add_node("aggregator", aggregator)
builder.add_edge(START, "splitter")
builder.add_conditional_edges("splitter", lambda s: s, path_map=["worker"]) # 关键：分发
builder.add_edge("worker", "aggregator")
builder.add_edge("aggregator", END)
graph = builder.compile()
print(graph.get_graph().draw_ascii())
# 图中会显示 splitter 有多条线指向 worker（并行感）🚀 进阶：生成高清 PNG 图片（可视化终极方案）
如果你想直接生成一张 漂亮的 PNG 流程图 用于文档或 PPT，运行下面代码（需要安装 pygraphviz）：
pip install pygraphviz# 生成并保存为高清图片
png_data = graph.get_graph().draw_mermaid_png()
with open("my_langgraph_flow.png", "wb") as f:
f.write(png_data)
print("流程图已保存为 my_langgraph_flow.png")💡 针对你的需求总结
以上所有示例，你只需要复制代码运行，就能在终端看到 ASCII 字符画流程图。如果是用 Jupyter Notebook，graph.get_graph().draw_mermaid() 会直接渲染出漂亮的矢量图。