LangGraph框架概述与核心概念
LangGraph是LangChain推出的智能体编排框架，基于有向状态图（StateGraph）构建多步骤、多Agent协作的工作流。传统LangChain Chain采用线性执行链路，面对需要条件分支、循环回退、状态管理的复杂场景时表达能力不足。LangGraph将工作流抽象为节点（Node）和边（Edge），每个节点是一个可执行函数，边定义节点间的跳转逻辑，支持条件边（Conditional Edge）实现动态路由。AI Agent的编排能力直接决定了智能对话系统在实际业务中的落地效果。
核心组件包括：
– StateGraph：有向图容器，管理全局状态对象的读写
– Node：执行单元，接收当前状态返回更新后的状态
– Edge：静态跳转，固定从一个节点指向另一个节点
– Conditional Edge：动态跳转，根据状态内容决定下一个执行节点
– Checkpointer：状态持久化，支持断点续跑与人机交互
状态图构建与条件路由实现
LangGraph的状态管理依赖TypedDict定义的数据结构，每个节点函数接收并返回该结构的部分字段，框架自动合并更新。以下是一个技术支持场景的多Agent协作示例，包含意图分类、知识检索、代码生成三个节点：
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
intent: str
context: str
response: str
def classify_intent(state: AgentState) -> dict:
last_msg = state["messages"][-1].content
if "报错" in last_msg or "异常" in last_msg:
return {"intent": "debug"}
elif "代码" in last_msg or "实现" in last_msg:
return {"intent": "code"}
else:
return {"intent": "qa"}
def retrieve_knowledge(state: AgentState) -> dict:
intent = state["intent"]
if intent == "debug":
context = search_error_db(state["messages"][-1].content)
else:
context = search_docs(state["messages"][-1].content)
return {"context": context}
def generate_response(state: AgentState) -> dict:
reply = llm.generate(
prompt=state["messages"][-1].content,
context=state["context"]
)
return {"response": reply}
def route_intent(state: AgentState) -> str:
if state["intent"] == "qa":
return "retrieve"
elif state["intent"] == "code":
return "generate"
else:
return "retrieve"
workflow = StateGraph(AgentState)
workflow.add_node("classify", classify_intent)
workflow.add_node("retrieve", retrieve_knowledge)
workflow.add_node("generate", generate_response)
workflow.set_entry_point("classify")
workflow.add_conditional_edges(
"classify",
route_intent,
{"retrieve": "retrieve", "generate": "generate"}
)
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", END)
app = workflow.compile()上述代码定义了三节点状态图，classify节点执行意图分类后，通过条件边route_intent决定走知识检索还是直接生成。retrieve节点完成后固定跳转到generate节点，最终结束。StateGraph通过compile()编译为可执行应用，调用时传入初始状态即可启动。
多Agent协作与消息传递模式
多Agent场景下，每个Agent封装为独立节点，通过共享状态实现信息传递。常见的协作模式包括：
– Hub-Spoke模式：中央协调Agent分发任务给专业Agent，汇总结果后统一输出
– Sequential模式：Agent按固定顺序依次执行，前一个的输出作为后一个的输入
– Debate模式：多个Agent对同一问题给出不同视角的回答，由裁判Agent综合裁决
以下是一个Researcher-Writer-Critic三角色协作的代码实现：
from langgraph.graph import StateGraph, END
from typing import TypedDict
class WritingState(TypedDict):
topic: str
research_notes: str
draft: str
critique: str
iteration: int
def researcher(state: WritingState) -> dict:
notes = search_and_summarize(state["topic"])
return {"research_notes": notes}
def writer(state: WritingState) -> dict:
draft = write_article(
topic=state["topic"],
notes=state["research_notes"]
)
return {"draft": draft}
def critic(state: WritingState) -> dict:
feedback = critique_writing(
draft=state["draft"],
criteria=["准确性", "完整性", "可读性"]
)
return {"critique": feedback, "iteration": state.get("iteration", 0) + 1}
def should_revise(state: WritingState) -> str:
if state["iteration"] < 3 and "需要修改" in state["critique"]:
return "writer"
return END
workflow = StateGraph(WritingState)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("critic", critic)
workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "critic")
workflow.add_conditional_edges(
"critic",
should_revise,
{"writer": "writer", END: END}
)
app = workflow.compile()
result = app.invoke({
"topic": "向量数据库选型对比",
"iteration": 0
})writer和critic之间形成循环，critic根据评审结果决定是否让writer重新修改。通过iteration字段控制最大循环次数，避免死循环。这种自迭代机制在自动化内容生成、代码审查等场景中实用价值较高。
断点续跑与人机交互机制
LangGraph通过Checkpointer实现状态持久化，支持在任意节点暂停后恢复执行。MemorySaver是内存实现，SqliteSaver和PostgresSaver提供持久化存储。人机交互（Human-in-the-Loop）场景下，可以在关键节点设置中断，等待人工确认后继续：
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["critic"]
)
config = {"configurable": {"thread_id": "session-001"}}
result = app.invoke(
{"topic": "向量数据库选型", "iteration": 0},
config=config
)
# 执行暂停在critic前，可检查state
current_state = app.get_state(config)
print(current_state.values.get("draft"))
# 人工确认后继续执行
result = app.invoke(None, config=config)thread_id标识一个执行会话，中断后可通过get_state查看当前状态，人工审核draft内容后调用invoke(None, config)恢复执行。该机制适用于需要人工审核中间结果的自动化流水线，如代码生成后的安全审查、文档生成后的事实核对等场景。
部署与性能考量
LangGraph应用部署时需注意几个关键点。状态对象应保持轻量，避免在状态中存储大体积数据，建议将检索结果、生成内容等存储在外部系统中，状态仅保留引用ID。节点函数应设计为幂等操作，便于重试和恢复。对于长时间运行的工作流，使用持久化Checkpointer而非MemorySaver，防止进程重启丢失状态。
LangGraph支持流式输出，通过stream方法逐步获取节点执行结果，适合前端实时展示多Agent协作过程。生产环境中建议结合Redis或PostgreSQL作为Checkpointer后端，配合监控指标跟踪每个节点的执行耗时和成功率。AI模型部署时，节点内的LLM调用应设置超时和重试策略，避免单个节点阻塞整个工作流。Prompt工程的优化也应贯穿每个节点的设计，确保Agent间的信息传递清晰准确。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/ai-zhi-neng-ti-kuang-jia-shi-zhan-langgraph-zhuang-tai-tu/
