从零开始学大模型：单体Agent到多智能体协作(LangChain+LangGraph实战，建议收藏)

本文详细介绍了大模型应用开发的两种核心架构：基于LangChain的单体Agent实现（工具定义与记忆功能）和基于LangGraph的多智能体协作系统（共享状态、节点定义与路由逻辑）。通过完整的作家-批评家协作系统代码示例，展示了从基础工具配置到复杂工作流设计的全流程，帮助开发者掌握大模型应用的核心技术架构与实现方法。

大耳朵爱学习

653人浏览 · 2025-12-24 11:24:24

大耳朵爱学习 · 2025-12-24 11:24:24 发布

简介

单体Agent (LangChain基础**)**

1. 定义工具 (Perception & Tools)

后端映射： API 接口定义。利用 docstring 生成文档，利用 Type Hint 生成参数校验。

from langchain.tools import tool
@tool
def get_current_price(symbol: str):
"""
查询加密货币的当前价格。
Args:
symbol: 代币代码 (如 BTC, ETH)，必须大写。
"""
# Runtime 执行逻辑 (数据清洗)
# 只返回关键字段，避免 Token 浪费
return {"price": 67000, "currency": "USD"}

2. 组装与记忆 (Assembly & Memory)

后端映射： 依赖注入 + Session 管理。

from langchain.agents import initialize_agent, AgentType
from langchain.memory import ConversationBufferMemory
# 1. 记忆模块 (Session)
memory = ConversationBufferMemory(
memory_key="chat_history", # 存放在 Prompt 中的变量名
return_messages=True       # Chat 模型必须设为 True
)
# 2. 初始化 Agent
agent = initialize_agent(
tools=[get_current_price],
llm=llm,                   # 配置好 base_url 的 DeepSeek/OpenAI 对象
agent=AgentType.OPENAI_FUNCTIONS,
memory=memory,             # 注入记忆
verbose=True               # 开启日志，观察思考过程
)

多智能体协作 (LangGraph)

核心逻辑： 基于图 (Graph) 的状态机。通过共享状态 (State) 实现多个 Agent (Nodes) 之间的通信与循环。

1. 定义共享状态 (The State)

后端映射： Redis / 共享内存。所有微服务的通信总线。

import operator
from typing import Annotated, List, TypedDict
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
# 关键点：operator.add 表示新消息是 "append" 而不是覆盖
# 这保证了历史记录的完整性
messages: Annotated[List[BaseMessage], operator.add]
loop_count: int

2. 定义节点 (The Nodes)

后端映射： 微服务 (Microservices)。单一职责，输入 State，输出 State 更新量。

def writer_node(state: AgentState):
# 1. 获取上下文
messages = state['messages']
# 2. 执行业务逻辑 (调用 LLM)
response = llm.invoke(...)
# 3. 返回更新量 (只返回变化的部分)
return {
"messages": [response],
"loop_count": state.get("loop_count", 0) + 1
}

3. 构建图与路由 (Graph & Routing)

后端映射： 服务编排 / 网关路由。

from langgraph.graph import StateGraph, END
workflow = StateGraph(AgentState)
# 添加节点 (注册服务)
workflow.add_node("writer", writer_node)
workflow.add_node("critic", critic_node)
# 确定性边缘 (Writer -> Critic)
workflow.add_edge("writer", "critic")
# 条件边缘 (Critic -> ? 根据逻辑判断)
workflow.add_conditional_edges(
"critic",             # 上游节点
decide_next_step,     # 路由函数 (返回 "rewrite" 或 "end")
{
"rewrite": "writer",
"end": END
}
)
app = workflow.compile() # 编译成可执行对象

可运行代码

import operator
from typing
import Annotated, List, TypedDict, Union
from langchain_openai
import ChatOpenAI
from langchain_core.messages
import SystemMessage, HumanMessage, BaseMessage
from langgraph.graph
import StateGraph, END
# === === === === === === === === === === === === === === #1. 定义状态(The State)# === === === === === === === === ===
=== === === === === #这就像是后端服务中的 Context 或 Session 对象# 所有节点(Nodes) 都能读取并修改这个 State
class AgentState(TypedDict): #Annotated[List, operator.add] 意味着：# 当有节点返回新的 messages 时， 不要覆盖旧的， 而是 append(追加) 到列表后面
messages: Annotated[List[BaseMessage], operator.add]# 用于记录循环次数， 防止死循环
loop_count: int
# === === === === === === === === === === === === === === #2. 初始化大脑(The Brain)# === === === === === === === === ===
=== === === === === #这里配置 DeepSeek(兼容 OpenAI 协议)
deepseek_api_key_v2 = "test"
deepseek_api_base = "https://api.deepseek.com"
llm = ChatOpenAI(
model = "deepseek-chat", #或者 gpt - 4 openai_api_key = deepseek_api_key_v2, #替换你的 Key openai_api_base =
"https://api.deepseek.com/v1",
temperature = 0.7,
verbose = True
)
# === === === === === === === === === === === === === === #3. 定义节点(Nodes / Agents)# === === === === === === === ===
=== === === === === ===
def writer_node(state: AgentState):
""
"作家节点：负责生成内容"
""
print(f "\n--- [Writer] 正在思考 (轮次: {state.get('loop_count', 0)}) ---")
# 获取历史消息
messages = state['messages']
# 给 Writer 的指令# === 修改点： 根据是否有历史记录， 动态调整 System Prompt ===
loop_count = state.get("loop_count", 0)
if loop_count > 0: #如果不是第一轮， 说明被怼了。 加重语气！
instruction = ""
"
你之前的笑话被审核员驳回了。
请仔细阅读审核员（ Critic） 的反馈意见。
1. 绝对不要重复之前的笑话。
2. 根据反馈进行修改， 或者写一个全新的。
3. 目标是让审核员回复 "PASS"。
""
"
else :#第一轮， 正常发挥
instruction = "你是一个脱口秀编剧。请写一个关于 Python 程序员的短笑话。"
system_prompt = SystemMessage(content = instruction)
# 调用 LLM
response = llm.invoke([system_prompt] + messages)
print(f "📝 [Writer 生成内容]:\n{response.content}")# 更新状态： 追加消息， 并增加计数器
return {
"messages": [response],
"loop_count": state.get("loop_count", 0) + 1
}
def critic_node(state: AgentState):
""
"批评家节点：负责审核"
""
print("\n--- [Critic] 正在审核 ---")
messages = state['messages']
last_joke = messages[-1].content
# 给 Critic 的指令
prompt = f ""
"
你是一个严厉的批评家。 请点评下面这个笑话：
"{last_joke}"
如果笑话非常幽默， 请只回复 "PASS"。
如果笑话一般或不好笑， 请给出修改建议， 并要求重写。
""
"
response = llm.invoke([HumanMessage(content = prompt)])
print(f "🧐 [Critic 审核意见]:\n{response.content}")
return {
"messages": [response]
}
# === === === === === === === === === === === === === === #4. 定义边缘逻辑(Edges / Routing)# === === === === === === === ===
=== === === === === ===
def decide_next_step(state: AgentState) - > str:
""
"路由函数：决定下一步去哪里"
""
messages = state['messages']
last_message = messages[-1].content
loop_count = state.get("loop_count", 0)
# 兜底机制： 如果尝试超过3次， 强制结束(防止无限死循环)
if loop_count >= 5:
print("\n--- [System] 尝试次数过多，强制结束 ---")
return "end"
# 核心判断逻辑
clean_msg = last_message.strip().upper()
# 如果 LLM 真的很听话， 只回了 "PASS"
if clean_msg == "PASS":
print("\n--- [System] 审核通过！ ---")
return "end"
# 或者稍微宽容一点： 如果它说了 "PASS"
且总字数少于 10 个字符（ 说明没有长篇大论的批评）
if "PASS" in clean_msg and len(clean_msg) < 10:
print("\n--- [System] 审核通过！ ---")
return "end"
print("\n--- [System] 审核不通过，打回重写 ---")
return "rewrite"
# === === === === === === === === === === === === === === #5. 组装图(Graph Assembly)# === === === === === === === ===
=== === === === === ===
#初始化图， 传入状态类型
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("writer", writer_node)
workflow.add_node("critic", critic_node)
# 定义流程# 入口 - > Writer
workflow.set_entry_point("writer")
# Writer - > Critic(无条件流转)
workflow.add_edge("writer", "critic")
# Critic - > ? (条件流转)
workflow.add_conditional_edges(
"critic", #从哪个节点开始判断 decide_next_step, #运行哪个判断函数 {
#
映射关系： 函数返回值 - > 下一个节点名 "rewrite": "writer", #如果返回 rewrite， 回到 writer "end": END# 如果返回 end， 结束流程
}
)
# 编译图(这也像编译代码一样， 把它变成可执行对象)
app = workflow.compile()
# === === === === === === === === === === === === === === #6. 运行(Execution)# === === === === === === === === ===
=== === === === ===
print("🚀 启动 LangGraph工作流...")
inputs = {
"messages": [HumanMessage(content = "开始工作")],
"loop_count": 0
}
# stream 模式可以看到每一步的输出
for output in app.stream(inputs): #这里只是为了让主进程挂起等待， 具体的打印已经在节点函数里做了
pass

AI大模型从0到精通全套学习大礼包

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

只要你是真心想学AI大模型，我这份资料就可以无偿共享给你学习。大模型行业确实也需要更多的有志之士加入进来，我也真心希望帮助大家学好这门技术，如果日后有什么学习上的问题，欢迎找我交流，有技术上面的问题，我是很愿意去帮助大家的！

如果你也想通过学大模型技术去帮助就业和转行，可以扫描下方链接👇👇
大模型重磅福利：入门进阶全套104G学习资源包免费分享！

在这里插入图片描述

01.从入门到精通的全套视频教程

包含提示词工程、RAG、Agent等技术点
在这里插入图片描述

02.AI大模型学习路线图（还有视频解说）

全过程AI大模型学习路线

在这里插入图片描述

03.学习电子书籍和技术文档

市面上的大模型书籍确实太多了，这些是我精选出来的

在这里插入图片描述

04.大模型面试题目详解

在这里插入图片描述

05.这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

所有的视频由智泊AI老师录制，且资料与智泊AI共享，相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

在这里插入图片描述