系列衔接:上篇我们深入解析了 DeepAgents 多Agent协调工作的原理——多个智能体分工协作、任务拆解、子节点并行执行,构成了现代生产级Agent系统的骨架。但再精妙的架构,也需要从最基础的一块砖砌起。今天这篇零基础实战篇,我们从12个Python文件入手,手把手带你从单Agent对话、工具调用、记忆管理,一步步搭建出完整的Agent系统,为理解多Agent协作打下坚实的编码基础。
上篇理论,本篇落地。但实战之前,我决定先带你“徒手”拆解并重构一个完整的Agent系统——通过12个文件的阶梯式编码,从空白目录到可运行的智能体,每一步都清晰可见。这既是入门,也是体系化的起点。
🔍 项目概览
根目录下共有 12个Python文件,分为两大类别:
学习路径总览
入门阶段 ──→ 进阶阶段 ──→ 高级阶段 ──→ 调试阶段 │ │ │ │ ├─ LLM调用 ├─ 工具调用 ├─ 工具+记忆 ├─ 环境检查 └─ 基础对话 └─ 状态管理 └─ 完整Agent └─ API测试
🎯 第一部分:核心Agent学习路径
📚 1. lc_agent_demo.py —— LLM调用入门
定位:这是整个项目的起点,教你如何调用大模型。这是一切Agent能力的根基——多Agent系统中的每个子Agent,本质上都是这样一个LLM调用的封装。
核心代码:
from langchain_openai import ChatOpenAImodel = ChatOpenAI( model="qwen-plus", api_key=os.getenv("DASHSCOPE_API_KEY"), openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1")response = model.invoke("Hello, 测试一下")print(response.content)
关键点解读:
| | |
|---|
| 阿里云百炼API | | |
| dotenv安全管理 | | |
| invoke()方法 | | |
适用场景:快速验证模型连接是否正常,构建多Agent前的"Hello World"。
💬 2. lc_chat_agent.py —— 基础对话Agent
定位:在LLM基础上封装成可交互的Agent。这是单Agent的雏形——多Agent协调中每一个子节点,本质上都是这样一个封装后的对话单元。
核心改进:
from langchain.agents import create_agentagent = create_agent( model=model, tools=[], # 空工具列表 system_prompt="你是一个友好的AI助手...")response = agent.invoke({"messages": [{"role": "user", "content": "介绍一下什么是AI Agent"}]})
核心概念:
执行流程:
用户提问 → Agent思考 → 直接回答
🛠️ 3. lc_tool_agent.py —— 工具调用能力
定位:让Agent具备使用外部工具的能力,这是Agent的核心价值!在多Agent系统中,工具调用是子Agent获取外部信息的关键途径。
工具定义:
from langchain_core.tools import toolfrom datetime import datetime@tooldefget_current_time():"""返回当前的日期和时间。当用户问'现在几点'时调用。"""return datetime.now().strftime("%Y-%m-%d %H:%M:%S")@tooldefcalculator(expression: str):"""计算数学表达式。输入应该是一个字符串形式的数学表达式。""" result = eval(expression, {"__builtins__": {}}, {})returnf"计算结果:{expression} = {result}"
Agent创建:
agent = create_agent( model=model, tools=[get_current_time, calculator], # 注入工具 system_prompt="需要实时时间调用get_current_time,计算调用calculator")
工作原理对比:
🔄 工具调用完整流程解析
📋 示例场景
用户提问:"现在几点了?"
工具:get_current_time() —— 返回当前时间
完整流程图
🚀 详细步骤拆解
① 用户提问
user_input = "现在几点了?"
用户输入自然语言问题,发送给Agent系统。
② Agent接收问题
Agent将问题包装成标准格式:
{"messages": [{"role": "user", "content": "现在几点了?"}]}
③ 模型分析:是否需要工具?
核心逻辑:模型根据system_prompt和工具描述(docstring)判断是否需要调用工具。
模型思考过程:
"用户问'现在几点了?',根据系统提示,我需要调用get_current_time工具来获取实时时间。"
④ 选择工具
模型决定调用get_current_time工具,生成工具调用指令:
{"thought": "用户询问当前时间,需要调用get_current_time工具获取实时时间。","tool_name": "get_current_time","tool_args": {} # 此工具不需要参数}
⑤ 调用工具
Agent执行工具函数:
# 执行结果result = get_current_time() # 返回 "2024-01-15 14:30:45"
⑥ 获取工具返回结果
工具执行完成后,返回结果被封装回对话历史:
{"messages": [ {"role": "user", "content": "现在几点了?"}, {"role": "assistant", "content": None, "tool_calls": [ {"name": "get_current_time", "args": {}} ]}, {"role": "tool", "name": "get_current_time", "content": "2024-01-15 14:30:45"} ]}
⑦ 模型整合信息
模型再次被调用,此时输入包含:
⑧ 生成最终答案
# 模型生成最终回答final_response = model.invoke({"messages": [ {"role": "user", "content": "现在几点了?"}, {"role": "assistant", "content": None, "tool_calls": [...]}, {"role": "tool", "name": "get_current_time", "content": "2024-01-15 14:30:45"} ]})
⑨ 返回给用户
最终答案被格式化并返回给用户:
用户:现在几点了?Agent:现在是2024年1月15日下午2点30分45秒。
📊 另一个例子:计算器工具
用户提问:"23乘以45加100等于多少?"
流程:
🧠 模型如何决定是否调用工具?
⚡ 工具调用的优势
- 突破模型限制 —— 获取实时信息(时间、天气)、执行精确计算
- 扩展能力边界 —— 连接外部系统(数据库、API),这是多Agent协作的基础
🧠 4. lc_mem_agent.py —— 记忆功能
定位:让Agent记住对话历史,实现多轮对话。在多Agent协调中,记忆是子Agent之间传递上下文的关键机制——协调器需要知道每个子Agent完成了什么,子Agent需要知道总任务的目标。
🧠 LangGraph核心概念
LangGraph是一个用于构建状态驱动的Agent工作流的框架,它就像一个"智能导演",负责:
🔧 代码详细解析
状态定义
from typing import TypedDict, Annotatedfrom langgraph.graph.message import add_messagesclassAgentState(TypedDict): messages: Annotated[list, add_messages] # 自动累积消息
逐行解释:
| | |
|---|
| | |
| Annotated[list, add_messages] | | |
| | |
为什么需要这个?
# ❌ 如果没有add_messages,每次更新会覆盖:state = {"messages": ["新消息"]} # 旧消息丢失!# ✅ 有了add_messages,自动累积:state = {"messages": ["旧消息1", "旧消息2", "新消息"]} # 保留历史
在多Agent协调中,每个子Agent的执行结果都会追加到同一个状态中,协调器基于完整状态做出下一步决策。
Graph构建
from langgraph.graph import StateGraph, END# 1. 创建状态图graph = StateGraph(AgentState)# 2. 定义节点函数defcall_model(state: AgentState):# state包含所有历史消息 response = model.invoke(state["messages"])# 返回新消息,会自动累加到状态中return {"messages": [response]}# 3. 添加节点graph.add_node("agent", call_model) # 节点名: "agent",执行函数: call_model# 4. 设置入口点graph.set_entry_point("agent") # 从"agent"节点开始执行# 5. 设置结束点graph.add_edge("agent", END) # "agent"执行完后结束
核心组件说明:
💡 多Agent衔接:如果在这个Graph中添加多个node(如"data_reader"、"analyzer"、"reporter"),再用add_edge连接它们,就实现了最简单的多Agent协调!
记忆持久化
from langgraph.checkpoint.memory import InMemorySaver# 创建记忆存储checkpointer = InMemorySaver() # 内存级别存储,重启后丢失# 编译Graph,注入记忆功能graph = graph.compile(checkpointer=checkpointer)
关键机制:
| | |
|---|
| | |
| compile(checkpointer=...) | | |
| | |
实际调用
# 创建配置(关键:thread_id用于区分不同用户)config = {"configurable": {"thread_id": "user_123"}}# 第一轮对话result1 = graph.invoke( {"messages": [{"role": "user", "content": "我叫张三"}]}, config=config)# 第二轮对话(记忆保留)result2 = graph.invoke( {"messages": [{"role": "user", "content": "我叫什么名字?"}]}, config=config)# Agent能正确回答:"你叫张三"
🎯 LangGraph在记忆功能中的核心作用
┌─────────────────────────────────────────────────────┐│ LangGraph 架构 │├─────────────────────────────────────────────────────┤│ ││ 用户输入 ──→ StateGraph ──→ 节点执行 ──→ 输出 ││ │ │ ││ ↓ ↓ ││ Checkpointer 状态更新 ││ │ ││ ↓ ││ 持久化存储 (InMemorySaver) ││ │└─────────────────────────────────────────────────────┘
LangGraph带来的价值:
| | |
|---|
| 状态管理自动化 | | |
| 流程可视化 | | |
| 记忆持久化 | | |
| 会话隔离 | | |
| 可扩展性 | | |
🔄 执行流程示例
第一轮对话:用户: "我叫张三" ↓StateGraph收到消息 ↓调用"agent"节点 → model.invoke(["我叫张三"]) ↓返回: "你好张三!很高兴认识你" ↓Checkpointer保存状态: {"messages": ["我叫张三", "你好张三!"]} ↓返回结果给用户第二轮对话:用户: "我叫什么名字?" ↓StateGraph加载历史状态 ↓调用"agent"节点 → model.invoke(["我叫张三", "你好张三!", "我叫什么名字?"]) ↓返回: "你叫张三呀!" ↓Checkpointer更新状态 ↓返回结果给用户
代码对比 —— 有 vs 无LangGraph:
# ❌ 没有LangGraph:手动管理历史,容易出错history = []history.append({"role": "user", "content": "我叫张三"})response = model.invoke(history)history.append({"role": "assistant", "content": response.content})# ... 需要重复编写大量样板代码# ✅ 有LangGraph:自动管理历史,专注业务逻辑graph.invoke({"messages": [{"role": "user", "content": "我叫张三"}]}, config=config)
🌟 5. lc_tool_mem_agent.py —— 完整功能Agent
定位:集大成者!工具调用 + 记忆功能的完美结合。这也是多Agent系统中每个子Agent的标准形态——既有能力调用外部工具获取信息,又能记住上下文保持对话连贯性。
架构设计:
# 1. 定义工具@tooldefget_current_time(): ...@tooldefcalculator(expression: str): ...# 2. 创建带工具的Agentagent = create_agent( model=model, tools=[get_current_time, calculator], system_prompt="你是一个友好的AI助手...")# 3. 包装成带记忆的GraphclassAgentState(TypedDict): messages: Annotated[list, add_messages]defcall_model(state: AgentState): result = agent.invoke({"messages": state["messages"]})return {"messages": [result["messages"][-1]]}graph = StateGraph(AgentState)graph.add_node("agent", call_model)graph.compile(checkpointer=InMemorySaver())
完整的单Agent执行流程:
用户输入 → 状态更新 → Agent分析 → 工具调用(可选)→ 状态持久化 → 返回结果
应用场景:
🔧 第二部分:调试工具详解
环境检查系列
6. check_env.py
功能:检查API密钥是否正确加载。
api_key = os.getenv("DASHSCOPE_API_KEY")print(f"API Key: {api_key}")print(f"API Key repr: {repr(api_key)}")
7. check_all_env.py
功能:列出所有与AI开发相关的环境变量。
for key in os.environ:ifany(keyword in key.upper() for keyword in ['API', 'KEY', 'LANG']):print(f"{key}={value[:50]}...")
HTTP调试系列
8. test_httpx.py
功能:测试自定义HTTP客户端。
http_client = httpx.Client()model = ChatOpenAI(..., http_client=http_client)
9. test_transport.py
功能:记录HTTP请求头详情,排查认证问题。
classLoggingTransport(httpx.HTTPTransport):defhandle_request(self, request):print("=== Request Headers ===")for key, value in request.headers.items():print(f"{key}: {repr(value)}")returnsuper().handle_request(request)
10. test_headers.py
功能:检查请求头中的非ASCII字符问题。
11. test_encoding.py
功能:修复Windows控制台中文输出乱码问题。
import sysif sys.platform == 'win32':import codecs sys.stdout = codecs.getwriter('utf-8')(sys.stdout.buffer, 'strict')
API测试系列
12. test_api.py
功能:验证API连接是否正常。
try: model = ChatOpenAI(...) response = model.invoke("Hello")print(f"Success: {response.content}")except Exception as e:print(f"Error: {e}")
📊 学习路径建议
┌─────────────────────────────────────────────────────┐│ 学习路径总览 │├─────────────────────────────────────────────────────┤│ ││ 🟢 入门阶段 ││ ├── lc_agent_demo.py → 理解LLM基础调用 ││ └── lc_chat_agent.py → 掌握Agent封装 ││ ││ 🟡 进阶阶段 ││ ├── lc_tool_agent.py → 学会工具调用 ││ └── lc_mem_agent.py → 理解状态管理 ││ ││ 🔴 高级阶段 ││ └── lc_tool_mem_agent.py → 构建完整Agent ││ ││ ⚪ 调试阶段 ││ ├── check_env.py → 环境检查 ││ ├── test_api.py → API测试 ││ └── test_encoding.py → 编码问题 ││ │└─────────────────────────────────────────────────────┘
阶段目标与多Agent能力映射
💡 核心概念总结
Agent三要素
┌─────────────────────┐ │ 🧠 模型 │ │ 大语言模型作为核心 │ │ 推理引擎 │ └──────────┬──────────┘ │ ┌──────────────┼──────────────┐ │ │ │ ▼ ▼ ▼┌────────┐ ┌────────┐ ┌────────┐│ 🛠️ 工具 │ │ 💾 记忆 │ │ 🔗 协调 ││ 扩展能力 │ │ 维持上下文│ │ 连接多Agent│└────────┘ └────────┘ └────────┘
LangGraph关键组件速查
| | | |
|---|
| StateGraph | | | |
| State | | | |
| Node | | graph.add_node("agent", fn) | |
| Edge | | | |
| Checkpointer | | compile(checkpointer=...) | |
🎯 实践建议
- 从简单开始 —— 先运行lc_agent_demo.py验证环境是否通
- 逐步进阶 —— 每掌握一个文件再进入下一个,不要跳级
- 动手修改 —— 尝试添加自己的工具(如天气查询、数据库查询)
- 调试技巧 —— 遇到问题先用check_env.py排查环境
- 为多Agent打基础 —— 理解单Agent的结构后,将多个agent节点接入同一个StateGraph,就实现了最简单的多Agent协调
✨ 结语
通过这12个文件的学习,你已经掌握了构建AI Agent的核心技能:
- ✅ 状态管理 —— 用LangGraph编排Agent流程
🚀 下一步预告:理解了单Agent的构建,下一篇我们将把这些独立Agent串起来,用LangGraph构建真正的多Agent协调系统——让多个Agent分工协作、并行执行、共同完成复杂任务!
现在你可以开始构建自己的AI Agent应用了!如果有任何问题,欢迎留言讨论。
📌 关注我,获取更多AI Agent开发干货!