【LangGraph实战】《LangGraph实战》_125.[第7章 架构设计] 智能体系统架构设计全景:从单Agent到多Agent

从“单兵作战”到“智能军团”:LangGraph架构设计全景揭秘——别让Agent变成无头苍蝇!6大设计要点带你从Demo走向企业级多Agent系统。本文将深入剖析单Agent状态机、状态管理、路由决策、多Agent协作、记忆持久化及生产级落地等核心架构设计,帮你避开“一把梭”的思维误区,搭建真正能打仗的智能体系统。
文字目录
- 一、单Agent架构:别让Agent变成“一根筋”
- 二、状态管理:你的图为什么总是“跑飞”
- 三、路由与决策:给Agent装上“脑子”和“手脚”
- 四、多Agent协作:从“自言自语”到“团队作战”
- 五、记忆与持久化:没有记忆的Agent只是个脚本
- 六、生产级架构:从Demo到能打仗的系统
嗨,大家好呀,我是你的老朋友精通代码大仙。接下来我们一起学习 《LangChain核心技术与LLM项目实践》
俗话说“一个和尚挑水喝,两个和尚抬水喝,三个和尚没水喝”。这话放到Agent开发里,简直不要太真实。你是不是也这样:刚开始学LangChain,写个单Agent觉得自己行了;一看业务复杂,直接堆提示词硬刚,结果Agent像个无头苍蝇一样乱转。后来听说多Agent厉害,咔咔往上堆,三个Agent直接“没水喝”——互相甩锅、死循环、状态乱成一锅粥。别慌,今天咱们就把智能体系统架构这事掰开了、揉碎了讲明白。
一、单Agent架构:别让Agent变成“一根筋”
很多新手对Agent的第一印象就是“大模型+提示词+工具调用”,链式执行一轮就完事。这叫Chain,不叫Agent。真正的单Agent,在LangGraph里是一个有状态、可循环、能分支的状态机。它得有输入、有处理、有决策、有反馈,形成一个完整的闭环。你想想,人做事也不是一条道走到黑的,写代码要调试,查资料要反复确认,Agent凭啥就得“一次成型”?
你刚开始写的时候,是不是这么干的?搞一个巨大的system prompt,把所有业务规则塞进去,然后调用llm.invoke(),期待它一次性给出完美答案。碰到需要“写代码→运行→报错→修改”这种场景,直接傻眼。要么写个for循环硬套,要么让LLM自己决定继续不继续,结果不是死循环就是提前结束。还有人把ReAct包了一层就以为是Agent了。ReAct确实能思考-行动,但它的逻辑是线性的,你想加个“人工确认”的节点?想加个“代码审查”的分支?难如登天。
更离谱的是“超级节点”现象。一个节点里既调LLM、又调工具、又做字符串解析、又改状态。代码耦合到天际,出问题了连从哪开始Debug都不知道。这就像让一个人同时当产品经理、程序员、测试、运维,不出乱子才怪。
# 错误示范:把所有逻辑塞进一个函数,期望LLM是全能神
def bad_agent(query):
prompt = f"你是一名助手。请遵循以下100条规则,调用工具,自己循环直到完成...{query}"
response = llm.invoke(prompt)
# 期望它自己调用工具、自己判断、自己循环、自己纠错
return response
这就是典型的“一根筋”思维。你把Agent当神仙,其实它就是个需要架构支撑的打工人。你不给它搭好舞台,它只能原地打转。
在LangGraph里,哪怕只有一个Agent,你也得用StateGraph来搭。把任务拆成节点:理解需求、制定计划、调用工具、观察结果、判断继续还是结束。节点之间用边连接,该循环的地方画环,该分支的地方加条件。每个节点只做一件事,做好一件事。
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
# 1. 定义状态契约
class State(TypedDict):
query: str
code: str
error: str
# 2. 节点只做一件事:生成代码
def generate(state: State):
code = llm.invoke(f"请生成代码解决:{state['query']}")
return {"code": code.content}
# 3. 节点只做一件事:运行检查
def check(state: State):
result = run_code(state["code"])
if result.get("error"):
return {"error": result["error"]}
return {"error": ""}
# 4. 条件边:决定是重试还是结束
def decide(state: State):
if state["error"]:
return "generate" # 回去重写
return END
# 5. 组装图
builder = StateGraph(State)
builder.add_node("generate", generate)
builder.add_node("check", check)
builder.add_edge(START, "generate")
builder.add_edge("generate", "check")
builder.add_conditional_edges("check", decide)
graph = builder.compile()
看见没?三个函数,职责清晰。generate只管写代码,check只管验证,decide只管拍板。Agent不是bigger prompt,而是better graph。你拆得越细,调试越轻松,后面想加节点(比如加个review节点)也就是一行代码的事。
单Agent不是“提示词工程”,而是“状态机工程”。把流程拆开,让节点各司其职,你的Agent才真的有“脑子”。
二、状态管理:你的图为什么总是“跑飞”
如果说节点是Agent的器官,那State就是血液。LangGraph的StateGraph之所以能支撑复杂交互,核心在于它有一个全局状态在各个节点间流转。但就是这个State,让80%的新手栽过跟头。状态设计不好,整个图就像一辆刹车失灵的车,指不定在哪个节点就“跑飞”了。
你是不是遇到过这种诡异bug?节点A明明更新了state里的某个字段,节点B读到的却是老数据。或者两个并行节点同时改一个列表,结果数据乱了套。更有甚者,定义State时图省事直接用裸dict,字段想加就加,最后自己都记不清图里到底有哪些数据在流动。最经典的是对“update机制”的误解。LangGraph默认会把节点的返回值和原状态做合并,不是覆盖。你以为return一个新dict就会清空其他字段?Too naive。还有Annotated[list, add_messages]这种reducer,新手看到就头大,用错了直接内存爆炸。
class BadState(TypedDict):
messages: list # 没有reducer,出事了吧?
result: str
def node_a(state: BadState):
# 本想追加消息,结果如果直接return新list,默认行为可能替换
return {"messages": [new_message]}
def node_b(state: BadState):
# 和node_a并行,都改result,最后谁生效全凭运气
return {"result": "来自B的结果"}
上面的messages没有指定reducer,每次return都容易被替换。而且并行写入同一字段,默认行为可能跟你想的不一样。 state就像公共卫生间,你不立规矩,大家进去就乱涂乱画。
定义State时,务必用TypedDict加Annotations明确每个字段的更新策略。需要追加的用add_messages或自定义reducer,需要覆盖的保持默认。字段要精简,不要把全世界都塞进State。记住一个原则:State是节点间的契约,不是垃圾桶。
from typing import Annotated
from langgraph.graph import add_messages
class GoodState(TypedDict):
# 消息自动追加,不会丢历史,内存不够了再想办法裁剪
messages: Annotated[list, add_messages]
# 代码字段,默认覆盖,永远只保留最新版
code: str
# 错误信息,默认覆盖
error: str
# 执行次数,用自定义reducer累加,记录重试了多少次
step_count: Annotated[int, lambda x, y: x + y]
def generate(state: GoodState):
# 安全地更新,不用担心把其他字段误删
return {
"messages": [AIMessage(content="生成代码...")],
"code": "print('hello')",
"step_count": 1
}
关键点:每个节点只该读写自己关心的字段,不要随意污染全局状态。并发节点尤其要注意,如果必须读写同一字段,考虑用reducer定义合并策略,或者干脆在架构上避免并发写冲突。
状态设计是架构的地基,地基歪了,楼盖再高也得塌。明确Schema、选对reducer、控制字段粒度,你的图才能稳如老狗。
三、路由与决策:给Agent装上“脑子”和“手脚”
有了状态和节点,Agent还得知道什么时候该干什么。这就是路由的作用。条件边让Graph从“死板的流水线”变成“灵活的决策网”。同时,工具调用是Agent感知和改造外部世界的“手脚”。没有路由,Agent是瞎子;没有工具,Agent是瘫子。
我见过太多新手把路由写得稀碎。要么整个图一条道走到黑,遇到异常直接崩溃;要么条件边里塞了一堆if-else,返回的字符串和下游节点名对不上,运行时报NodeNotFound错。还有把API调用逻辑直接写在LLM节点里的,节点又负责思考又负责干活,代码耦合到天际。更惨的是,路由判断太粗糙,比如只看LLM返回里有没有“错误”两个字就决定重试,结果正常回复里刚好带了这俩字,直接触发死循环。
def bad_router(state):
response = llm.invoke(state["messages"])
# 这种字符串匹配也太草率了,跟掷骰子差不多...
if "错误" in response.content:
return "retry_node"
elif "工具" in response.content:
return "tool_node"
else:
return "end" # 万一拼写错了呢?图直接崩溃
# 节点里直接塞工具逻辑,又当爹又当妈
def llm_and_tool_node(state):
response = llm.invoke(...)
if response.tool_calls:
result = requests.post(...) # 异常了怎么重试?超时了怎么办?
return {"result": result}
这种代码,维护起来就是噩梦。三天后你自己回看,都不一定能看懂当时怎么想的。
把决策逻辑和执行逻辑彻底分开。条件边只负责“看状态、做决定”,不干别的。工具调用统一走ToolNode,让LangGraph帮你管理工具的生命周期、异常包装、并行调用。路由返回值用常量或清晰的逻辑判断,别搞字符串魔法。
from langgraph.prebuilt import ToolNode
from langgraph.graph import END
# 工具单独定义,统一管理
tools = [search, calculator]
tool_node = ToolNode(tools)
def call_model(state: State):
# LLM绑定工具,由模型自己决定要不要调用
response = model.bind_tools(tools).invoke(state["messages"])
return {"messages": [response]}
# 纯粹的路由逻辑,只关心一件事:有没有工具要调?
def should_continue(state: State) -> str:
last_msg = state["messages"][-1]
if not last_msg.tool_calls:
return "end"
return "tools"
builder.add_node("agent", call_model)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "end": END}
)
builder.add_edge("tools", "agent")
看见没?should_continue只看有没有tool_calls,简单可靠,不会误判。工具执行丢给ToolNode,异常处理、结果回传都能统一做。你的“脑子”只管想,“手脚”只管干,别混在一起。
路由是Agent的神经系统,工具是Agent的手脚。脑子清晰、手脚利落,Agent才能指哪打哪。
四、多Agent协作:从“自言自语”到“团队作战”
当业务复杂度突破单Agent的天花板,你就得考虑多Agent架构了。LangGraph处理多Agent的核心思路是:把每个Agent当作Graph中的一个或一组节点,通过状态传递实现协作。常见模式有Supervisor监督者、Hierarchical层级化、Network去中心化。听起来很美好对吧?但坑也特别多。
新手一听到多Agent就激动,咔咔往上加。结果Agent A和Agent B各自为战,A生成的数据B读不懂,B的返回A不知道怎么处理。最惨的是通信协议不统一,有的Agent直接改全局state,有的期望接收特定格式,最后图跑起来像一团乱麻。还有拓扑选择困难症,上来就搞去中心化,每个Agent都能调用其他Agent,结果呢?A调B,B调C,C觉得不对又调A,死循环!或者三个Agent并行处理,最后合并结果时才发现格式完全不兼容。
# 错误示范:Agent互相硬调用,没有统一指挥,格式随意
def agent_a(state):
result = agent_b.invoke({"task": "帮我查数据"})
# 期望result是字符串,结果B返回了个dict,直接报错
return {"output": result + "处理完毕"}
def agent_b(state):
# B又偷偷调了C,调用链深不见底
c_result = agent_c.invoke(...)
return {"data": c_result} # 返回格式和A的预期不一致!
这种硬编码调用、格式随意、循环依赖的代码,在生产环境里就是定时炸弹。三个和尚没水喝,不是因为人多了,是因为没组织。
新手入门多Agent,强烈建议从Supervisor模式开始。一个中心调度员负责任务分发和结果汇总,子Agent只负责执行,不互相直接调用。通信格式要统一,比如都通过state里的messages或专门的task_result字段传递。每个子Agent干完活必须回到Supervisor报道,由Supervisor决定下一步,这就天然避免了死循环。
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
next: str # Supervisor决定下一步派谁
# 子Agent封装成节点,只关心自己的活
def researcher(state: AgentState):
# 只负责搜索,不关心谁调用它
return {"messages": [AIMessage(content="搜索结果是...")]}
def coder(state: AgentState):
return {"messages": [AIMessage(content="代码如下...")]}
def supervisor(state: AgentState):
# LLM作为Supervisor,根据当前进度决定下一步
decision = router_llm.invoke(state["messages"])
return {"next": decision.content}
builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_edge(START, "supervisor")
# Supervisor根据next字段路由
builder.add_conditional_edges(
"supervisor",
lambda s: s["next"],
{"researcher": "researcher", "coder": "coder", "end": END}
)
builder.add_edge("researcher", "supervisor")
builder.add_edge("coder", "supervisor")
当然,Supervisor不是唯一的模式。业务稳定后,你可以演进成层级结构,比如一个部门经理管几个小组长,小组长再管子Agent。但万变不离其宗:通信协议统一、调用关系有向无环(或者严格受控)、职责边界清晰。
多Agent不是“人多力量大”,而是“组织出战斗力”。选好拓扑、统一通信、中心调度,你的Agent团队才能协同作战。
五、记忆与持久化:没有记忆的Agent只是个脚本
你写的到底是“智能体”还是“脚本”,关键看有没有记忆。短期记忆让Agent能记住当前会话的上下文,长期记忆让Agent能积累知识、认识用户。LangGraph通过Checkpointer机制提供了强大的状态持久化能力,但很多新手根本不会用。
很多新手的Agent每次重启就失忆,用户前脚刚说“我叫张三”,后脚就问“你是谁”。以为存个messages列表就叫有记忆了?那只是短期记忆,而且图一重新编译就消失。长期记忆更是无从谈起,有人拿全局变量存用户画像,多用户一并发直接串台。还有对Checkpointer的误用,配了个MemorySaver就以为万事大吉,结果部署到多实例服务器上,状态存在内存里,请求一到另一台机器就找不到。或者该用数据库时图省事用内存,重启全丢。
# 错误示范:伪记忆,全局变量灾难
user_memory = {} # 全局变量,线程不安全,多用户串台!
def chat_node(state):
user_id = state["user_id"]
if user_id not in user_memory:
user_memory[user_id] = []
user_memory[user_id].append(state["input"])
# 服务器重启归零,并发高了直接乱套
这种代码,Demo能跑,上线必挂。用户会以为在和个金鱼聊天,记忆只有七秒。
短期记忆交给LangGraph的checkpointer,开发用MemorySaver,生产环境务必换成PostgresSaver、RedisSaver或自研实现,别用内存。调用graph时传入唯一的thread_id,状态就会自动持久化,下次接着聊。长期记忆(用户画像、知识库)在节点里显式读写外部存储,比如向量库或关系型数据库。
from langgraph.checkpoint.memory import MemorySaver
# 开发阶段用MemorySaver,生产环境请换PostgresSaver
checkpointer = MemorySaver()
# 编译图时挂上checkpointer
graph = builder.compile(checkpointer=checkpointer)
# 调用时传入thread_id,状态自动持久化
config = {"configurable": {"thread_id": "user_123"}}
result = graph.invoke(
{"messages": [HumanMessage(content="我叫张三,做后端的")]},
config
)
# 过了一小时,用户回来了,带着同样的thread_id,Agent记得你
result2 = graph.invoke(
{"messages": [HumanMessage(content="你觉得我该学Agent吗")]},
config
)
但记住,messages会越长越多,最后撑爆上下文窗口。所以长期记忆不能全靠messages。你可以在节点里定期总结对话,把关键信息提炼出来存到向量库:
def summarize_and_store(state: State):
# 把对话总结成用户画像
profile = llm.invoke(f"总结用户偏好:{state['messages']}")
# 存到长期记忆库
vector_store.upsert(f"user:{state['user_id']}", profile)
return {"user_profile": profile}
下次对话时,先加载长期记忆塞进system prompt,Agent就会说:“老张啊,上次你说想做后端自动化,我这有个方案特别适合你……” 这体验,瞬间就不一样。
记忆让Agent从“一次性脚本”进化为“持续服务的智能体”。短期记忆靠Checkpointer,长期记忆靠外部存储,双管齐下才能真人味十足。
六、生产级架构:从Demo到能打仗的系统
Demo能跑只是万里长征第一步。真正的生产级Agent系统,必须考虑容错、限流、可观测性、人机协同。LangGraph提供了interrupt、retry_policy等机制,但架构思维得你自己有。不然上线那天,就是翻车那天。
你是不是觉得,本地测试百发百中,上线之后LLM就开始“抽风”:返回格式不对、胡说八道、调用不存在的工具。工具侧更惨,API超时、限流、返回500错误。你的图里如果没有错误处理节点,一个异常直接全栈崩溃。还有更危险的:让Agent自动执行敏感操作,比如自动发邮件、自动下单、自动删数据。Demo里图省事直接执行,真出了事哭都来不及。另外,并发一高,第三方API直接把你拉黑了。
def auto_trade_node(state):
# 危险!直接执行,没有确认,没有重试,没有限流
order = llm.invoke("根据分析结果生成交易指令")
execute_trade(order) # 钱没了!数据丢了!
return {"status": "done"}
def parallel_search(state):
# 一口气开10个并发查询,API直接429限流,全挂
results = [search_tool.invoke(q) for q in state["queries"]]
return {"results": results}
这种代码,老板看了流泪,财务看了心碎,运维看了想跑路。
关键操作必须加人机协同回路。LangGraph的interrupt可以在任意节点暂停执行,等用户确认后再继续。节点内要做防御式编程:try-except、参数校验、结果格式校验。工具调用配重试策略(指数退避)。并发控制用异步Semaphore或限制LangGraph的并行度。
from langgraph.types import interrupt
def trade_node(state: State):
# 生成交易建议,但不直接执行
proposal = generate_trade_proposal(state)
# 暂停,等用户确认,这就是人机协同
user_confirm = interrupt({
"message": "系统建议执行以下交易,是否确认?",
"proposal": proposal
})
if user_confirm == "yes":
safe_execute(proposal)
else:
return {"status": "cancelled"}
# 工具节点加保护壳
def safe_tool_call(state: State):
try:
result = tool.invoke(state["args"])
# 格式校验,防止LLM胡言乱语传错参数
assert validate_result(result)
return {"result": result}
except Exception as e:
return {"error": str(e), "retry_count": state.get("retry_count", 0) + 1}
# 条件边判断重试还是转人工
def error_router(state: State):
if state.get("error") and state.get("retry_count", 0) > 3:
return "human_help"
elif state.get("error"):
return "retry"
return "continue"
同时,一定要上可观测性。接入LangSmith或自研trace,每个节点的输入输出、耗时、Token消耗都得看得清清楚楚。不然用户问你“刚才Agent怎么胡说了”,你连它调了哪几个节点都不知道,那还怎么优化?
还有一点容易被忽略:streaming。生产环境里用户不可能干等30秒看Agent发呆。用LangGraph的stream模式,把中间过程实时推给用户,哪怕最后结果没出来,用户也知道“哦,它在查资料了”、“它在写代码了”,体验好很多。
生产环境不相信眼泪,只相信容错和可控。加上人机协同、错误处理、可观测性,你的Agent系统才能真正上战场。
写在最后
老规矩,写到这我得跟你掏心窝子说几句。Agent开发这事,听起来很酷,做起来全是细节。从单Agent的一根筋,到多Agent的团队战;从状态的精心设计,到路由的清晰决策;从短期记忆的Checkpointer,到生产环境的人机协同。每一步都不是多余的,每一环都关乎你的系统能不能从“玩具”变成“工具”。
我知道你现在可能有点焦虑:这么多东西要记,坑这么多,我能搞定吗?听哥一句,编程之路没有白走的路,你踩过的每一个坑,都是架构师路上的垫脚石。LangGraph只是把状态机这个古老而强大的思想,重新带到了LLM时代。抓住“状态、节点、边”这三个核心,剩下的就是不断迭代、不断打磨。
别怕出错,就怕你不敢把图拆细、把职责分清楚。保持好奇,持续迭代,你写的Agent一定会越来越聪明、越来越可靠。咱们下篇见!
关注私信备注:“资料代找获取”,全网计算机学习资料代找:例如:
《课程:AI 大模型工程师系统课程 (22 章完整版 持续更新)》
《课程:AI 大模型系统实战课第四期 (2026 年开课 持续更新)》
《课程:2026 年 AGI 大模型系统课 23 期》
《课程:2026 年 AGI 大模型系统课 21 期》
《课程:AI 大模型实战课 8 期 (2026 年 2 月最新完结版)》
《课程:AI 大模型系统实战课三期》
《课程:AI 大模型系统课程 (2026 年 2 月开课 持续更新)》
《课程:AI 大模型全阶课程 (2025 年 12 月开课 2026 年 6 月结课)》
《课程:AI 大模型工程师全阶课程 (2025 年 10 月开课 2026 年 4 月结课)》
《课程:2026 年最新大模型 Agent 开发系统课 (持续更新)》
《课程:LLM 多模态视觉大模型系统课》
《课程:大模型 AI 应用开发企业级项目实战课 (2026 年 1 月开课)》
《课程:大模型智能体线上速成班 V2.0》
《课程:Java+AI 大模型智能应用开发全阶课》
《课程:Python+AI 大模型实战视频教程》
《书籍:软件工程 3.0: 大模型驱动的研发新范式.pdf》
《课程:人工智能大模型系统课 (2026 年 1 月底完结版)》
《课程:AI 大模型零基础到商业实战全栈课第五期》
《课程:Vue3.5+Electron + 大模型跨平台 AI 桌面聊天应用实战 (2025)》
《课程:AI 大模型实战训练营 从入门到实战轻松上手》
《课程:2026 年 AI 大模型 RAG 与 Agent 智能体项目实战开发课》
《课程:大模型训练营配套补充资料》
更多推荐


所有评论(0)