AI Agent 智能体开发实战：从原理到项目落地（AutoGen/LangChain Agent）
·
📖 目录
- AI Agent 是什么？为什么2025年所有大厂都在抢？
- 三大核心架构：ReAct / Handoff / CodeAgent
- 环境准备与依赖安装
- AutoGen 多智能体协作实战
- LangChain Agent 工具调用实战
- RAG + Agent 深度集成
- 完整项目：智能客服 Agent 实战
- 🚨 避坑指南（2025年最新踩坑总结）
- 总结与下一步建议
1. AI Agent 是什么？为什么2025年所有大厂都在抢？
1.1 一句话定义
AI Agent（智能体）≠ 大语言模型（LLM）。
LLM 是"大脑"，能理解能生成，但不会主动做事。
AI Agent 是"大脑 + 手 + 眼 + 工具"——它能自主感知环境、制定计划、调用工具、执行动作、从反馈中学习，最终完成一个复杂目标。
用户提问 → Agent(LLM大脑) → 思考(推理) → 规划(Plan) → 执行(Action)
↓
← 观察(Observation) ← 工具调用
1.2 为什么2025年这么火？
| 年份 | 里程碑 |
|---|---|
| 2023 | LLM 爆发，ChatGPT 引领对话式 AI |
| 2024 | Agent 框架百花齐放（AutoGen、CrewAI、LangGraph） |
| 2025 | Agent 进入生产化阶段：MCP 协议标准化、多智能体协作成熟、企业级 RAG+Agent 落地 |
- OpenAI 推出 Agents SDK / Responses API
- Microsoft AutoGen 成为多智能体事实标准
- LangChain LangGraph 重构 Agent 运行时
- MCP 协议 让 Agent 工具调用标准化，不再每家写一套
1.3 适用场景
| 场景 | 传统方案 | Agent 方案 |
|---|---|---|
| 智能客服 | 关键词匹配 + 固定流程 | Agent 动态规划 + 工具查询 |
| 代码开发 | Copilot 补全 | 多 Agent 协作完成完整需求 |
| 数据分析 | 手动写 SQL、画图 | Agent 自动查库、分析、可视化 |
| 自动化运维 | Shell 脚本 + Cron | Agent 监控 → 诊断 → 自愈 |
2. 三大核心架构：ReAct / Handoff / CodeAgent
2.1 ReAct（Reasoning + Acting）— 最经典的 Agent 模式
论文出处：“ReAct: Synergizing Reasoning and Acting in Language Models” (ICLR 2023)
核心思想：思考 → 行动 → 观察 → 再思考 的循环。
Thought: 用户想查北京的天气，我需要调用天气查询工具
Action: get_weather(city="北京")
Observation: {"temp": 32, "weather": "晴"}
Thought: 获取到天气数据了，现在用中文回复用户
Final Answer: 北京今天32°C，天气晴朗 ☀️
适用场景：单轮工具调用、知识问答、信息检索类任务。
2.2 Handoff（移交模式）— 多 Agent 协作的基础
将一个复杂任务拆解，专业 Agent 接力完成。
用户: "帮我分析Q2财报并生成PPT"
↓
主管Agent: 拆解任务
├── 数据分析Agent → 拉取数据、计算结果
├── 图表生成Agent → 生成趋势图
└── 报告编写Agent → 组装PPT
↓
主管Agent: 汇总输出
适用场景：复杂业务流程、多角色协作、企业级工作流。
2.3 CodeAgent — 写代码解决问题
Agent 不是对话，是写代码执行。CodeAgent 让 LLM 生成 Python 代码并执行，天然适合数学计算、数据操作。
用户: "帮我计算过去30天用户留存率"
CodeAgent 生成:
```python
def retention_rate():
df = pd.read_sql("SELECT * FROM user_login", conn)
df['date'] = pd.to_datetime(df['date'])
# ... 计算每日留存
return result
执行代码 → 返回结果
**适用场景**：数据清洗、数学计算、文件处理等需要精确执行的场景。
### 2.4 三种架构怎么选？
| 架构 | 复杂度 | 适合场景 | 代表框架 |
|------|--------|---------|---------|
| ReAct | ⭐低 | 单工具调用、问答 | LangChain Agent、OpenAI Function Calling |
| Handoff | ⭐⭐⭐高 | 多角色协作、复杂流程 | AutoGen、CrewAI、LangGraph |
| CodeAgent | ⭐⭐中 | 数据处理、代码执行 | Code Interpreter、AutoGen CodeAgent |
---
<h2 id="3">3. 环境准备与依赖安装</h2>
### 3.1 Python 环境
```bash
# 推荐 Python 3.11+
python --version
# 创建虚拟环境
python -m venv agent_env
source agent_env/bin/activate # Linux/Mac
# agent_env\Scripts\activate # Windows
3.2 安装依赖
# === 核心依赖 ===
pip install pyautogen==0.35.0 # AutoGen 多智能体框架
pip install langchain==0.3.29 # LangChain 核心
pip install langchain-openai==0.3.10 # OpenAI 支持
pip install langchain-community==0.3.23
pip install langgraph==0.3.28 # LangGraph Agent 运行时
# === 工具依赖 ===
pip install duckduckgo-search # 网页搜索工具
pip install wikipedia-api # Wikipedia 工具
pip install yfinance # 股票查询工具
pip install requests beautifulsoup4 # 网页抓取
pip install pandas numpy # 数据处理
pip install matplotlib # 图表生成
# === RAG 依赖 ===
pip install chromadb # 向量数据库
pip install sentence-transformers # 向量嵌入
pip install pypdf # PDF 解析
pip install tiktoken # Token 计数
# === 开发辅助 ===
pip install python-dotenv # 环境变量管理
3.3 环境变量配置
创建 .env 文件：
# .env
OPENAI_API_KEY=sk-your-openai-api-key
# 国内推荐使用兼容 OpenAI 的 API（如 DeepSeek、通义千问、硅基流动等）
# OPENAI_BASE_URL=https://api.siliconflow.cn/v1
# OPENAI_API_KEY=sk-your-siliconflow-key
# AutoGen 配置
AUTOGEN_LLM_CONFIG='{"config_list":[{"model":"gpt-4o","api_key":"sk-xxx"}]}'
💡 国内用户替代方案：DeepSeek V3、Qwen3、GLM-4 等国产模型均已兼容 OpenAI API 格式，可直接替换。
3.4 验证安装
# test_env.py
import autogen
from langchain.agents import AgentExecutor
from langchain_openai import ChatOpenAI
print("All imports successful!")
print(f"AutoGen version: {autogen.__version__}")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
resp = llm.invoke("Hello!")
print(f"LLM test: {resp.content[:50]}...")
4. AutoGen 多智能体协作实战
4.1 AutoGen 核心概念
AutoGen 是 Microsoft 推出的多智能体对话框架，核心组件：
| 组件 | 说明 |
|---|---|
| AssistantAgent | LLM 驱动的智能体，负责推理和生成 |
| UserProxyAgent | 模拟人类，负责执行代码和提供反馈 |
| GroupChat | 多智能体群聊管理器 |
| GroupChatManager | 群聊调度，控制发言顺序 |
4.2 实战 1：单 Agent 工具调用（已废弃 v0.2→v0.35 新写法）
⚠️ 重要提示：AutoGen v0.20+ 已经重构了 Agent API，本文使用的是 v0.35+ 的写法，和网上旧教程完全不同。
# autogen_simple_agent.py
import json
import autogen
from autogen import AssistantAgent, UserProxyAgent, register_function
# ========== 1. 定义工具函数 ==========
def get_weather(city: str) -> str:
"""获取指定城市的当前天气信息"""
# 这里模拟天气查询，实际可接入真实 API
weather_db = {
"北京": "晴转多云，28°C，东风3级",
"上海": "小雨，25°C，湿度85%",
"深圳": "多云，32°C，南风4级",
"广州": "雷阵雨，30°C",
"杭州": "阴天，26°C，体感舒适",
}
return weather_db.get(city, f"暂未收录 {city} 的天气数据")
def calculate(expression: str) -> str:
"""执行数学计算，传入数学表达式如 '1+2*3' """
try:
# 安全计算，仅允许基本运算
allowed = set("0123456789+-*/(). ")
if not all(c in allowed for c in expression):
return "Error: 表达式包含非法字符"
result = eval(expression)
return f"{expression} = {result}"
except Exception as e:
return f"计算错误: {str(e)}"
# ========== 2. 配置 LLM ==========
llm_config = {
"config_list": [
{
"model": "gpt-4o-mini", # 或 deepseek-chat / qwen-plus 等
"api_key": "sk-your-api-key", # 请替换为实际 API Key
}
],
"temperature": 0.1,
"timeout": 60,
}
# ========== 3. 创建 Agent ==========
# 助理 Agent — 负责思考与回复
assistant = AssistantAgent(
name="Assistant",
system_message=(
"你是一个智能助手，可以调用工具来回答用户问题。\n"
"你可以使用以下工具：\n"
"1. get_weather(city) — 查询天气\n"
"2. calculate(expression) — 数学计算\n"
"在回答中，如果需要调用工具，明确说明你要调用哪个工具。\n"
"如果工具返回结果，基于结果组织你的回复。"
),
llm_config=llm_config,
)
# 用户代理 — 负责接收工具执行结果
user_proxy = UserProxyAgent(
name="UserProxy",
human_input_mode="NEVER", # 不询问人类，完全自动
is_termination_msg=lambda x: x.get("content", "").find("TERMINATE") >= 0,
code_execution_config=False, # 不执行代码，只做对话路由
)
# ========== 4. 注册工具 ==========
register_function(
get_weather,
caller=assistant,
executor=user_proxy,
name="get_weather",
description="查询指定城市的天气",
)
register_function(
calculate,
caller=assistant,
executor=user_proxy,
name="calculate",
description="执行数学计算",
)
# ========== 5. 启动对话 ==========
def run_agent():
user_proxy.initiate_chat(
assistant,
message="你好！请问北京今天天气怎么样？然后帮我算一下 (85+37)*12 等于多少？",
max_turns=10,
)
if __name__ == "__main__":
run_agent()
运行结果示例：
UserProxy (to Assistant):
你好！请问北京今天天气怎么样？然后帮我算一下 (85+37)*12 等于多少？
Assistant (to UserProxy):
我来查一下北京的天气并计算这个表达式。
**天气查询**：调用 get_weather("北京")
**计算**：调用 calculate("(85+37)*12")
UserProxy (to Assistant):
>>>> EXECUTING FUNCTION get_weather...
>>>> EXECUTING FUNCTION calculate...
UserProxy (to Assistant):
get_weather -> 晴转多云，28°C，东风3级
calculate -> (85+37)*12 = 1464
Assistant (to UserProxy):
为您查询到以下结果：
1️⃣ **北京天气**：晴转多云，28°C，东风3级
2️⃣ **数学计算**：(85+37)×12 = **1464**
天气不错，适合外出活动！☀️
4.3 实战 2：多 Agent 群聊协作（GroupChat）
# autogen_groupchat.py
import autogen
from autogen import (
AssistantAgent,
UserProxyAgent,
GroupChat,
GroupChatManager,
)
# ========== LLM 配置 ==========
llm_config = {
"config_list": [
{
"model": "gpt-4o-mini",
"api_key": "sk-your-api-key",
}
],
"temperature": 0.1,
}
# ========== 定义多个 Specialist Agent ==========
# 管理员 — 负责拆解和汇总
manager_agent = AssistantAgent(
name="Manager",
system_message=(
"你是项目经理 Agent，负责拆解用户需求并分配给不同的专家 Agent。\n"
"当用户提出需求时，分析需求需要哪些专家参与，\n"
"然后引导对话让对应 Agent 响应。最终汇总结果。"
),
llm_config=llm_config,
)
# 数据分析师 Agent
analyst_agent = AssistantAgent(
name="DataAnalyst",
system_message=(
"你是数据分析师 Agent。你擅长 Python 数据处理（pandas、numpy）、\n"
"数据可视化和统计分析。当需要处理数据时，生成 Python 代码并执行。\n"
"只关注数据分析相关任务。"
),
llm_config=llm_config,
)
# 代码审查 Agent
code_review_agent = AssistantAgent(
name="CodeReviewer",
system_message=(
"你是代码审查员 Agent。你擅长审查 Python 代码的质量、\n"
"安全性、性能和最佳实践。给出具体的改进建议。\n"
"只关注代码审查相关任务。"
),
llm_config=llm_config,
)
# 技术写作 Agent
writer_agent = AssistantAgent(
name="TechnicalWriter",
system_message=(
"你是技术写作者 Agent。你擅长将技术内容转化为清晰、\n"
"易懂的中文文档和报告。注重逻辑结构、用词准确、\n"
"示例清晰。只关注技术文档编写任务。"
),
llm_config=llm_config,
)
# 用户代理 — 模拟用户输入（不参与群聊思考）
user_proxy = UserProxyAgent(
name="User",
human_input_mode="NEVER",
code_execution_config=False,
)
# ========== 创建 GroupChat ==========
agents = [manager_agent, analyst_agent, code_review_agent, writer_agent, user_proxy]
group_chat = GroupChat(
agents=agents,
messages=[],
max_round=20,
speaker_selection_method="auto", # 自动选择下一个发言者
allow_repeat_speaker=False, # 不允许同一人连续发言
)
manager = GroupChatManager(
groupchat=group_chat,
llm_config=llm_config,
)
# ========== 启动 ==========
def run_group_chat():
user_proxy.initiate_chat(
manager,
message=(
"帮我做一个数据分析项目：\n"
"1. 生成一份模拟的销售数据（包含日期、产品、销售额、数量）\n"
"2. 分析：哪个产品销售额最高？哪个月的销量最佳？\n"
"3. 写一段总结报告"
),
)
if __name__ == "__main__":
run_group_chat()
4.4 AutoGen 核心参数调优
# 关键参数配置模板
llm_config_best_practice = {
"config_list": [
{
"model": "gpt-4o",
"api_key": "sk-xxx",
# === 以下参数避免 Agent 循环或跑偏 ===
"max_tokens": 4096, # 防止 Agent 无限生成
}
],
"temperature": 0.1, # 越低越稳定，高则更有创意
"timeout": 120, # API 超时(秒)
# AutoGen v0.30+ 新参数
"allow_format_str_template": False, # 防止注入
"cache_seed": 42, # 缓存相同请求，调试时加速
}
# 终止条件最佳实践
def should_terminate(msg):
"""判断对话是否该结束"""
if msg.get("content") is None:
return False
content = msg["content"]
# 终止信号
if "TERMINATE" in content:
return True
# 防止空循环：连续 3 轮相同内容
return False
5. LangChain Agent 工具调用实战
5.1 LangChain Agent 架构演进（2025 最新）
2025 年的 LangChain Agent 已全面迁移到 LangGraph，旧的 AgentExecutor + ZeroShotAgent 已不推荐。
LangChain 0.1.x: AgentExecutor (旧)
LangChain 0.2.x: 推荐 LangGraph
LangChain 0.3.x: AgentExecutor 标记为 LEGACY
LangGraph 0.3+: 官方推荐 Agent 方案
5.2 LangGraph Agent — 2025 最新写法
# langgraph_agent.py
import json
from typing import Literal
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
# ========== 1. 定义工具 ==========
@tool
def search_web(query: str) -> str:
"""搜索网络获取最新信息"""
# 实际项目中可接入 DuckDuckGo、SerpAPI、Bing Search 等
import requests
try:
# 示例：用 DuckDuckGo 搜索（无 API Key 限制）
from duckduckgo_search import DDGS
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=3))
return "\n".join([
f"[{r['title']}]({r['href']})\n{r['body'][:200]}"
for r in results
])
except Exception as e:
return f"搜索失败: {e}"
@tool
def get_stock_price(symbol: str) -> str:
"""查询股票实时价格，symbol 为股票代码如 AAPL、TSLA、0700.HK"""
import yfinance as yf
try:
stock = yf.Ticker(symbol)
info = stock.info
price = info.get("currentPrice") or info.get("regularMarketPrice")
name = info.get("longName", symbol)
change = info.get("regularMarketChangePercent", 0)
return (
f"{name} ({symbol})\n"
f"当前股价: ${price:.2f}\n"
f"涨跌幅: {change:.2f}%"
)
except Exception as e:
return f"查询失败: {e}"
@tool
def calculate(expression: str) -> str:
"""执行数学计算，传入合法数学表达式"""
try:
# 使用 ast 做安全计算
import ast
import operator
operators = {
ast.Add: operator.add, ast.Sub: operator.sub,
ast.Mult: operator.mul, ast.Div: operator.truediv,
ast.Pow: operator.pow, ast.USub: operator.neg,
}
def safe_eval(node):
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
return node.value
elif isinstance(node, ast.BinOp):
op = operators[type(node.op)]
return op(safe_eval(node.left), safe_eval(node.right))
elif isinstance(node, ast.UnaryOp):
op = operators[type(node.op)]
return op(safe_eval(node.operand))
elif isinstance(node, ast.Expression):
return safe_eval(node.body)
raise ValueError(f"不支持的操作: {type(node)}")
tree = ast.parse(expression, mode="eval")
result = safe_eval(tree.body)
return f"{expression} = {result}"
except Exception as e:
return f"计算错误: {e}"
# ========== 2. 创建 LLM ==========
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
api_key="sk-your-api-key",
)
# ========== 3. 注册工具并创建 Agent ==========
tools = [search_web, get_stock_price, calculate]
# LangGraph 的 create_react_agent 是 2025 年推荐方式
agent = create_react_agent(
model=llm,
tools=tools,
# 以下添加系统提示词
prompt=(
"你是一个智能助手，可以使用多种工具来回答问题。\n"
"步骤：\n"
"1. 理解用户问题，判断需要哪些工具\n"
"2. 按需调用工具，获取结果\n"
"3. 基于结果组织回答\n"
"注意：如果工具返回错误，请尝试其他方法或如实告知用户。\n"
"回答使用中文。"
),
# 使用 MemorySaver 可以让 Agent 记住对话历史
checkpointer=MemorySaver(),
)
# ========== 4. 运行 Agent ==========
def run_agent():
config = {"configurable": {"thread_id": "thread-001"}}
# 第一轮
messages = agent.invoke(
{
"messages": [
("human", "帮我查一下苹果公司（AAPL）今天的股价，然后算一下如果买100股需要多少钱？")
]
},
config=config,
)
print("=== 第一轮 ===")
print(messages["messages"][-1].content)
print()
# 第二轮（利用对话历史）
messages = agent.invoke(
{
"messages": [
("human", "顺便搜一下最近的苹果新闻")
]
},
config=config,
)
print("=== 第二轮 ===")
print(messages["messages"][-1].content)
if __name__ == "__main__":
run_agent()
运行结果示例：
=== 第一轮 ===
🔍 我来查询苹果公司的最新股价并帮你计算！
1️⃣ **AAPL 当前股价**
苹果公司（AAPL）当前股价为 **$218.36**，今日上涨 0.52%。
2️⃣ **计算买入成本**
100 股 × $218.36 = **$21,836.00**
💡 提示：实际交易还需要考虑佣金和相关税费，请以券商平台为准。
=== 第二轮 ===
📰 **苹果最新动态**
1. **Apple Intelligence 新功能上线**
苹果在最新的 iOS 19 系统中推出了更强大的 AI 功能，包括...
2. **新一代 MacBook Pro 即将发布**
据彭博社报道，苹果计划在 9 月发布搭载 M5 芯片的...
3. **服务业务收入创新高**
...
5.3 自定义 Agent 节点（LangGraph 进阶）
# langgraph_custom_agent.py
from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# ========== 自定义状态 ==========
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
next_step: str
tool_results: dict
# ========== 节点函数 ==========
def call_llm(state: AgentState) -> dict:
"""LLM 推理节点"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
response = llm.invoke(state["messages"])
return {
"messages": [response],
"next_step": "route",
}
def should_continue(state: AgentState) -> Literal["tools", "end"]:
"""判断是否需要继续调用工具"""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "end"
def call_tool(state: AgentState) -> dict:
"""工具执行节点"""
last_message = state["messages"][-1]
tool_results = []
for tc in last_message.tool_calls:
# 模拟工具执行
result = f"执行了工具: {tc['name']}，参数: {tc['args']}"
tool_results.append(ToolMessage(content=result, tool_call_id=tc["id"]))
return {"messages": tool_results, "next_step": "continue"}
# ========== 构建图 ==========
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_llm)
workflow.add_node("tools", call_tool)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {
"tools": "tools",
"end": END,
})
workflow.add_edge("tools", "agent")
app = workflow.compile()
# 执行
result = app.invoke({
"messages": [HumanMessage(content="你好，今天是2025年7月10日，帮我查一下今天是什么日子？")],
"next_step": "start",
"tool_results": {},
})
6. RAG + Agent 深度集成
6.1 为什么需要 RAG + Agent？
单纯 RAG（检索增强生成）是被动检索：用户问什么，就检索什么。
RAG + Agent 是主动检索：Agent 先理解用户意图，再决定查什么、查几次、查完怎么加工。
❌ 纯 RAG: 用户 → 向量检索 → LLM 生成回答
✅ RAG+Agent: 用户 → Agent思考 → 多轮检索(RAG) → 分析结果 → 综合回答
↓ ↑
反思是否需要补充检索 ──────────────┘
6.2 构建 RAG 检索工具
# rag_agent_tool.py
import os
from typing import List
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.tools import tool
# ========== RAG 引擎 ==========
class RAGEngine:
"""轻量级 RAG 引擎，支持文档导入和检索"""
def __init__(self, persist_dir: str = "./chroma_db"):
self.persist_dir = persist_dir
# 使用国产 embedding 模型（bge-small 性能好、速度快）
self.embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5",
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True},
)
self.vector_store = None
self._load_or_create()
def _load_or_create(self):
"""加载已有向量库，或创建新的"""
if os.path.exists(self.persist_dir):
self.vector_store = Chroma(
persist_directory=self.persist_dir,
embedding_function=self.embeddings,
)
print(f"✅ 加载已有向量库: {self.persist_dir}")
else:
self.vector_store = Chroma(
persist_directory=self.persist_dir,
embedding_function=self.embeddings,
)
print(f"🆕 创建新向量库: {self.persist_dir}")
def add_documents(self, file_path: str):
"""导入文档到向量库"""
# 根据文件类型选择加载器
if file_path.endswith(".pdf"):
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path, encoding="utf-8")
documents = loader.load()
print(f"📄 加载了 {len(documents)} 页/段")
# 分块
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=100,
separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""],
)
chunks = splitter.split_documents(documents)
print(f"✂️ 分割为 {len(chunks)} 个文本块")
# 添加到向量库
self.vector_store.add_documents(chunks)
self.vector_store.persist()
print(f"✅ 成功导入 {file_path}")
def search(self, query: str, k: int = 5) -> List[str]:
"""检索相关文档"""
if self.vector_store is None:
return ["向量库为空，请先导入文档"]
docs = self.vector_store.similarity_search(query, k=k)
return [doc.page_content for doc in docs]
# ========== 全局 RAG 引擎实例 ==========
rag_engine = RAGEngine()
# ========== 注册为 Agent 工具 ==========
@tool
def add_knowledge_document(file_path: str) -> str:
"""向知识库中添加一个文档文件（支持 PDF/TXT），之后 Agent 可以检索"""
try:
rag_engine.add_documents(file_path)
return f"✅ 文档 {file_path} 已成功导入知识库"
except Exception as e:
return f"❌ 导入失败: {str(e)}"
@tool
def query_knowledge_base(query: str) -> str:
"""从知识库中检索与 query 相关的信息"""
results = rag_engine.search(query, k=3)
if not results:
return "知识库中未找到相关信息"
return "\n\n---\n\n".join([f"📌 相关片段 {i+1}:\n{r}" for i, r in enumerate(results)])
# ========== 独立测试 ==========
if __name__ == "__main__":
# 测试导入文档
if os.path.exists("./sample.pdf"):
add_knowledge_document("./sample.pdf")
# 测试检索
result = query_knowledge_base("什么是AI Agent")
print(result)
6.3 将 RAG 集成到 AutoGen Agent
# autogen_rag_agent.py
import autogen
from autogen import AssistantAgent, UserProxyAgent, register_function
from rag_agent_tool import query_knowledge_base, add_knowledge_document
llm_config = {
"config_list": [{"model": "gpt-4o-mini", "api_key": "sk-your-api-key"}],
"temperature": 0.1,
}
# 创建 Agent
rag_assistant = AssistantAgent(
name="RAGAssistant",
system_message=(
"你是一个基于知识库的智能问答助手。\n"
"规则：\n"
"1. 先使用 query_knowledge_base 工具检索知识库\n"
"2. 如果知识库有相关内容，基于检索结果回答\n"
"3. 如果知识库没有相关内容，使用自己的知识回答\n"
"4. 始终在回答中标注信息来源\n"
"5. 用户也可以使用 add_knowledge_document 上传新文档"
),
llm_config=llm_config,
)
rag_user = UserProxyAgent(
name="UserProxy",
human_input_mode="NEVER",
code_execution_config=False,
)
# 注册 RAG 工具
register_function(
query_knowledge_base,
caller=rag_assistant,
executor=rag_user,
name="query_knowledge_base",
description="从知识库中检索信息",
)
register_function(
add_knowledge_document,
caller=rag_assistant,
executor=rag_user,
name="add_knowledge_document",
description="向知识库中添加文档",
)
def run():
rag_user.initiate_chat(
rag_assistant,
message="请帮我查一下知识库中关于「AI Agent 架构」的信息",
max_turns=5,
)
if __name__ == "__main__":
run()
6.4 将 RAG 集成到 LangChain Agent
# langchain_rag_agent.py
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.tools import tool
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
# ========== RAG 工具 ==========
@tool
def search_knowledge_base(query: str) -> str:
"""从企业内部知识库中搜索信息"""
embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5"
)
vector_store = Chroma(
persist_directory="./chroma_db",
embedding_function=embeddings,
)
docs = vector_store.similarity_search(query, k=3)
if not docs:
return "未找到相关信息"
result = []
for i, doc in enumerate(docs, 1):
result.append(f"[片段{i}]\n{doc.page_content}")
return "\n\n".join(result)
@tool
def search_web(query: str) -> str:
"""搜索网络获取最新信息"""
from duckduckgo_search import DDGS
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=3))
return "\n".join([
f"[{r['title']}]({r['href']})\n{r['body'][:200]}"
for r in results
])
# ========== Agent 策略 ==========
# 策略：先查知识库（RAG），不够再搜索网络
RAG_SYSTEM_PROMPT = """你是一个企业智能问答助手。
工作流程：
1. 如果问题是关于企业内部的，**先调用 search_knowledge_base** 检索知识库
2. 如果知识库结果不完整或用户要求最新信息，**再调用 search_web** 搜索网络
3. 综合两个来源的信息，给出准确、完整的回答
4. 标注信息来源（知识库 vs 网络）
5. 如果知识库和网络都没有相关信息，如实告知用户
"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = create_react_agent(
model=llm,
tools=[search_knowledge_base, search_web],
prompt=RAG_SYSTEM_PROMPT,
checkpointer=MemorySaver(),
)
def ask(question: str, thread_id: str = "rag-001"):
config = {"configurable": {"thread_id": thread_id}}
result = agent.invoke(
{"messages": [("human", question)]},
config=config,
)
return result["messages"][-1].content
if __name__ == "__main__":
answer = ask("我们公司的产品有哪些核心功能？")
print(answer)
7. 完整项目：智能客服 Agent 实战
这个项目综合了 AutoGen 多智能体 + RAG 知识库 + 外部工具，实现了一个完整的智能客服系统。
7.1 项目结构
smart_customer_service/
├── main.py # 主程序入口
├── config.py # 配置文件
├── tools/
│ ├── __init__.py
│ ├── rag_tool.py # RAG 知识库检索
│ ├── order_tool.py # 订单查询工具
│ └── ticket_tool.py # 工单创建工具
├── agents/
│ ├── __init__.py
│ ├── router_agent.py # 路由/分类 Agent
│ ├── knowledge_agent.py # 知识问答 Agent
│ └── service_agent.py # 业务处理 Agent
├── knowledge/
│ └── faq.pdf # 常见问题文档
└── requirements.txt # 依赖清单
7.2 完整代码
config.py
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
# LLM 配置
LLM_CONFIG = {
"config_list": [
{
"model": os.getenv("LLM_MODEL", "gpt-4o-mini"),
"api_key": os.getenv("OPENAI_API_KEY", ""),
"base_url": os.getenv("OPENAI_BASE_URL", ""),
}
],
"temperature": 0.05, # 客服场景尽量稳定
"timeout": 60,
}
# 向量库配置
CHROMA_PERSIST_DIR = "./chroma_db"
# 客服路由规则
CATEGORY_KEYWORDS = {
"售后": ["退货", "退款", "维修", "换货", "售后"],
"订单": ["订单", "物流", "快递", "配送", "发货"],
"产品": ["功能", "使用", "安装", "配置", "参数"],
"其他": [],
}
tools/rag_tool.py
# tools/rag_tool.py
"""RAG 知识库检索工具"""
from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from config import CHROMA_PERSIST_DIR
class KnowledgeBase:
"""客服知识库"""
def __init__(self):
self.embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5",
model_kwargs={"device": "cpu"},
)
self.persist_dir = CHROMA_PERSIST_DIR
def load_or_create(self):
db_path = Path(self.persist_dir)
if db_path.exists():
return Chroma(
persist_directory=self.persist_dir,
embedding_function=self.embeddings,
)
return Chroma(
persist_directory=self.persist_dir,
embedding_function=self.embeddings,
)
def add_faq_documents(self, docs_dir: str = "./knowledge"):
"""导入 FAQ 文档"""
kb = self.load_or_create()
loader = DirectoryLoader(
docs_dir,
glob="**/*.pdf",
loader_cls=PyPDFLoader,
)
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(
chunk_size=300,
chunk_overlap=50,
)
chunks = splitter.split_documents(docs)
kb.add_documents(chunks)
kb.persist()
return len(chunks)
def search(self, query: str, k: int = 3) -> list[str]:
kb = self.load_or_create()
docs = kb.similarity_search(query, k=k)
return [d.page_content for d in docs]
# 全局单例
kb = KnowledgeBase()
def search_faq(query: str) -> str:
"""搜索 FAQ 知识库（Agent 工具函数）"""
results = kb.search(query)
if not results:
return "未找到相关 FAQ"
return "\n\n".join(
f"📖 FAQ #{i + 1}:\n{r}" for i, r in enumerate(results)
)
tools/order_tool.py
# tools/order_tool.py
"""订单查询工具"""
import json
from datetime import datetime, timedelta
import random
# 模拟订单数据库
MOCK_ORDERS = {
"ORD20250701001": {
"status": "已发货",
"product": "AI智能音箱 Pro",
"price": 599.00,
"logistics": "顺丰快递 SF1234567890",
"estimated_delivery": "2025-07-12",
"customer": "张三",
},
"ORD20250701002": {
"status": "已签收",
"product": "智能手表 Max",
"price": 1299.00,
"logistics": "中通快递 ZT9876543210",
"signed_time": "2025-07-08 14:30",
"customer": "李四",
},
"ORD20250701003": {
"status": "待发货",
"product": "无线降噪耳机",
"price": 399.00,
"estimated_delivery": "预计2个工作日内发出",
"customer": "王五",
},
}
def query_order(order_id: str) -> str:
"""查询订单状态（Agent 工具函数）"""
order = MOCK_ORDERS.get(order_id)
if not order:
return f"未找到订单号 {order_id}，请核对后重试"
result = [
f"📦 订单号: {order_id}",
f"商品: {order['product']}",
f"金额: ¥{order['price']:.2f}",
f"状态: {order['status']}",
]
if "logistics" in order:
result.append(f"物流: {order['logistics']}")
if "estimated_delivery" in order:
result.append(f"预计送达: {order['estimated_delivery']}")
if "signed_time" in order:
result.append(f"签收时间: {order['signed_time']}")
return "\n".join(result)
def cancel_order(order_id: str, reason: str = "") -> str:
"""取消订单（Agent 工具函数）"""
if order_id not in MOCK_ORDERS:
return f"未找到订单 {order_id}"
order = MOCK_ORDERS[order_id]
if order["status"] == "已发货":
return f"订单 {order_id} 已发货，无法直接取消。建议联系物流拦截。"
order["status"] = "已取消"
return f"✅ 订单 {order_id} 已成功取消。{reason}"
tools/ticket_tool.py
# tools/ticket_tool.py
"""工单创建工具"""
import uuid
from datetime import datetime
tickets_db = [] # 模拟数据库
def create_ticket(
customer_name: str,
category: str,
description: str,
contact: str = "",
) -> str:
"""创建客服工单（Agent 工具函数）"""
ticket_id = f"TK-{uuid.uuid4().hex[:8].upper()}"
ticket = {
"ticket_id": ticket_id,
"customer": customer_name,
"category": category,
"description": description,
"contact": contact,
"status": "待处理",
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"priority": "一般",
}
tickets_db.append(ticket)
return (
f"✅ 工单创建成功！\n"
f"工单号: {ticket_id}\n"
f"类型: {category}\n"
f"状态: {ticket['status']}\n"
f"创建时间: {ticket['created_at']}\n\n"
f"我们将尽快处理您的问题，请在24小时内关注处理状态。"
)
def query_ticket_status(ticket_id: str) -> str:
"""查询工单状态（Agent 工具函数）"""
for t in tickets_db:
if t["ticket_id"] == ticket_id:
return (
f"📋 工单号: {t['ticket_id']}\n"
f"状态: {t['status']}\n"
f"类型: {t['category']}\n"
f"描述: {t['description']}\n"
f"创建时间: {t['created_at']}"
)
return f"未找到工单 {ticket_id}"
agents/router_agent.py
# agents/router_agent.py
"""路由 Agent：判断用户意图，分配到对应 Agent"""
from autogen import AssistantAgent, UserProxyAgent, register_function
from config import LLM_CONFIG
# 分类 Agent
router_agent = AssistantAgent(
name="Router",
system_message=(
"你是客服路由 Agent。你的任务 ONLY 是判断用户问题的类别。\n\n"
"类别规则：\n"
"- 如果用户问的是产品功能、使用、安装、参数等问题 → 回复 'CATEGORY:产品'\n"
"- 如果用户问的是订单状态、物流、配送等问题 → 回复 'CATEGORY:订单'\n"
"- 如果用户问的是售后服务、退货、退款等问题 → 回复 'CATEGORY:售后'\n"
"- 如果不符合以上 → 回复 'CATEGORY:其他'\n\n"
"不要回答用户问题，只输出 CATEGORY 行！"
),
llm_config=LLM_CONFIG,
)
agents/knowledge_agent.py
# agents/knowledge_agent.py
"""知识问答 Agent：基于 RAG 回答产品问题"""
from autogen import AssistantAgent, UserProxyAgent, register_function
from config import LLM_CONFIG
from tools.rag_tool import search_faq
knowledge_agent = AssistantAgent(
name="KnowledgeAgent",
system_message=(
"你是产品知识问答 Agent。\n"
"1. 先用 search_faq 检索知识库\n"
"2. 基于检索结果回答用户问题\n"
"3. 如果知识库没有答案，用自己的知识回答并说明"
),
llm_config=LLM_CONFIG,
)
# 在 main.py 中注册工具
agents/service_agent.py
# agents/service_agent.py
"""业务处理 Agent：处理订单、售后、工单"""
from autogen import AssistantAgent, UserProxyAgent, register_function
from config import LLM_CONFIG
from tools.order_tool import query_order, cancel_order
from tools.ticket_tool import create_ticket, query_ticket_status
service_agent = AssistantAgent(
name="ServiceAgent",
system_message=(
"你是客服业务处理 Agent。\n"
"你可以处理以下事务：\n"
"1. 订单查询（query_order）\n"
"2. 订单取消（cancel_order）\n"
"3. 创建工单（create_ticket）\n"
"4. 查询工单状态（query_ticket_status）\n\n"
"礼貌、耐心、专业地处理用户的业务需求。\n"
"对于无法处理的情况，请引导用户联系人工客服。"
),
llm_config=LLM_CONFIG,
)
main.py
# main.py
"""智能客服 Agent 主程序"""
import autogen
from autogen import (
AssistantAgent,
UserProxyAgent,
GroupChat,
GroupChatManager,
register_function,
)
from config import LLM_CONFIG
from tools.rag_tool import search_faq, kb
from tools.order_tool import query_order, cancel_order
from tools.ticket_tool import create_ticket, query_ticket_status
def setup_agents():
"""初始化所有 Agent 并注册工具"""
# ===== 路由 Agent =====
router = AssistantAgent(
name="Router",
system_message=(
"你只负责分类用户问题，回复格式必须是以下之一：\n"
"'CATEGORY:产品' — 产品功能、使用、参数等\n"
"'CATEGORY:订单' — 订单、物流、配送等\n"
"'CATEGORY:售后' — 退货、退款、维修等\n"
"'CATEGORY:其他' — 其他问题\n"
"不要输出其他任何内容！"
),
llm_config=LLM_CONFIG,
)
# ===== 知识问答 Agent =====
knowledge = AssistantAgent(
name="KnowledgeAgent",
system_message=(
"你是产品知识问答专家。\n"
"请使用 search_faq 工具检索知识库后回答。\n"
"如果知识库没有答案，用自己的知识回答并标注。"
),
llm_config=LLM_CONFIG,
)
# ===== 业务处理 Agent =====
service = AssistantAgent(
name="ServiceAgent",
system_message=(
"你是客服业务处理专家。可使用以下工具：\n"
"- query_order: 查询订单\n"
"- cancel_order: 取消订单\n"
"- create_ticket: 创建工单\n"
"- query_ticket_status: 查询工单状态\n"
"处理完毕后回复'[COMPLETE]'表示完成。"
),
llm_config=LLM_CONFIG,
)
# ===== 汇总 Agent =====
summary = AssistantAgent(
name="SummaryAgent",
system_message=(
"你负责将其他 Agent 的回答整理为最终回复给用户。\n"
"要求：语言友好、结构清晰、包含关键信息。\n"
"整理完成后回复'TERMINATE'结束对话。"
),
llm_config=LLM_CONFIG,
)
# ===== 注册工具 =====
for agent, tools in [
(knowledge, [search_faq]),
(service, [query_order, cancel_order, create_ticket, query_ticket_status]),
]:
for tool_func in tools:
register_function(
tool_func,
caller=agent,
executor=UserProxyAgent(
name="Executor",
human_input_mode="NEVER",
code_execution_config=False,
),
name=tool_func.__name__,
description=tool_func.__doc__ or "",
)
return router, knowledge, service, summary
def create_group_chat(agents):
"""创建群聊"""
router, knowledge, service, summary = agents
# 用户代理
user = UserProxyAgent(
name="User",
human_input_mode="NEVER",
code_execution_config=False,
)
all_agents = [router, knowledge, service, summary, user]
# 自定义发言顺序
def custom_speaker_selector(last_speaker, groupchat):
"""根据上一条消息决定谁发言"""
messages = groupchat.messages
if not messages:
return router
last_msg = messages[-1].get("content", "")
# 路由 Agent 刚分类完
if last_speaker == router:
if "CATEGORY:产品" in last_msg:
return knowledge
elif "CATEGORY:订单" in last_msg or "CATEGORY:售后" in last_msg:
return service
else:
return summary
# 知识/服务 Agent 完成工作
if last_speaker in [knowledge, service]:
return summary
# 汇总完毕
if last_speaker == summary:
return None # 结束
return router
group_chat = GroupChat(
agents=all_agents,
messages=[],
max_round=10,
speaker_selection_method=custom_speaker_selector,
)
manager = GroupChatManager(
groupchat=group_chat,
llm_config=LLM_CONFIG,
)
return user, manager
def run_customer_service():
"""运行客服系统"""
print("=" * 50)
print("🤖 智能客服 Agent 系统 v2.0")
print("=" * 50)
print("支持：产品问答 / 订单查询 / 售后处理 / 工单创建")
print("输入 'exit' 退出")
print("-" * 50)
# 初始化
agents = setup_agents()
user, manager = create_group_chat(agents)
# 首次导入知识库
print("\n📚 正在初始化知识库...")
try:
count = kb.add_faq_documents("./knowledge")
print(f"✅ 已导入 {count} 条 FAQ\n")
except Exception as e:
print(f"⚠️ 知识库导入跳过: {e}\n")
while True:
query = input("\n🙋 用户: ").strip()
if query.lower() in ("exit", "quit", "退出"):
print("👋 感谢使用，再见！")
break
user.initiate_chat(
manager,
message=query,
max_turns=10,
silent=True, # 不打印内部对话，只输出最终结果
)
# 获取最后一条消息
last_msg = manager.groupchat.messages[-1]["content"]
print(f"\n🤖 客服: {last_msg}")
if __name__ == "__main__":
run_customer_service()
运行效果
==================================================
🤖 智能客服 Agent 系统 v2.0
==================================================
支持：产品问答 / 订单查询 / 售后处理 / 工单创建
输入 'exit' 退出
--------------------------------------------------
📚 正在初始化知识库...
✅ 已导入 15 条 FAQ
🙋 用户: 帮我查一下订单 ORD20250701001 的状态
🤖 客服: 📦 为您查询到订单信息如下：
**订单号：** ORD20250701001
**商品：** AI智能音箱 Pro
**金额：** ¥599.00
**状态：** ✅ 已发货
**物流：** 顺丰快递 SF1234567890
**预计送达：** 2025-07-12
您的商品已在路上，预计将于明天送达，请保持手机畅通以便快递员联系您！📞
🙋 用户: 这个音箱怎么连接WiFi？
🤖 客服: 根据知识库信息，AI智能音箱 Pro 连接 WiFi 步骤如下：
1️⃣ **开机**：长按电源键3秒
2️⃣ **进入配网模式**：首次开机自动进入，指示灯会橙色闪烁
3️⃣ **打开APP**：手机下载「AI Smart」App
4️⃣ **添加设备**：App首页点击「+」→ 扫描音箱底部的二维码
5️⃣ **选择WiFi**：输入您的WiFi密码（仅支持2.4GHz频段）
6️⃣ **等待连接**：约10-30秒，指示灯变蓝色常亮即连接成功
💡 提示：如果连接失败，请确保手机和音箱在同一个路由器下，且WiFi为2.4G频段。
🙋 用户: 我要退货
🤖 客服: 了解您的需求。我来帮您处理退货事宜。
已为您创建退货工单：
✅ 工单创建成功！
**工单号：** TK-A1B2C3D4
**类型：** 售后
**状态：** 待处理
**创建时间：** 2025-07-10 11:25:30
⏰ 我们的售后团队将在24小时内联系您处理退货事宜。
如需加急，请拨打客服热线 400-888-xxxx 并告知工单号。
---
<h2 id="8">8. 🚨 避坑指南（2025年最新踩坑总结）</h2>
### 坑 1：AutoGen v0.2 → v0.35 API 不兼容
```python
# ❌ 错的（v0.2 旧写法，2024年教程全是这个）
from autogen import AssistantAgent
assistant = AssistantAgent(
name="assistant",
llm_config={"model": "gpt-4"},
function_map={"get_weather": get_weather}, # 已废弃！
)
# ✅ 对的（v0.30+ 新写法）
from autogen import register_function
assistant = AssistantAgent(name="assistant", llm_config=llm_config)
user_proxy = UserProxyAgent(name="user", human_input_mode="NEVER", ...)
register_function(get_weather, caller=assistant, executor=user_proxy, name="get_weather")
坑 2：Agent 陷入死循环
症状：Agent 不停调用工具，重复相同模式，永不结束。
解决方案：
# 1. 限制最大轮数
user_proxy.initiate_chat(assistant, message=query, max_turns=5)
# 2. 设置合理的终止条件
is_termination_msg=lambda x: (
"TERMINATE" in (x.get("content", "") or "")
or "FINAL ANSWER" in (x.get("content", "") or "")
)
# 3. 系统提示词中明确终止指令
system_message="回答完毕后在末尾添加 TERMINATE"
# 4. LangGraph 使用 interrupt_before
from langgraph.checkpoint.memory import MemorySaver
agent = create_react_agent(
model=llm,
tools=tools,
interrupt_before=["tools"], # 每次调工具前暂停
checkpointer=MemorySaver(),
)
坑 3：Agent 产生幻觉（事实性错误）
# 🔧 解决：给 Agent 添加验证步骤
system_message = """回答规则：
1. 如果从工具获取了数据，基于数据回答
2. 如果没有工具可用，明确说明"这是我基于训练数据的回答，建议核实"
3. 不确定时不要编造，说"我需要查询更多信息"
4. 涉及数字、日期、统计数据，必须标注来源"""
坑 4：国产模型兼容性问题
# ✅ 已验证可用的国产模型（2025.07）
LLM_CONFIGS = {
# DeepSeek V3 — 性价比之王
"deepseek": {
"model": "deepseek-chat",
"api_key": "sk-xxx",
"base_url": "https://api.deepseek.com/v1",
},
# 通义千问 Qwen3 — 中文表现优秀
"qwen": {
"model": "qwen-plus",
"api_key": "sk-xxx",
"base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
},
# 硅基流动 — 第三方聚合，免翻墙
"siliconflow": {
"model": "Qwen/Qwen3-72B-Instruct",
"api_key": "sk-xxx",
"base_url": "https://api.siliconflow.cn/v1",
},
# GLM-4 — 适合复杂推理
"glm": {
"model": "glm-4-plus",
"api_key": "sk-xxx",
"base_url": "https://open.bigmodel.cn/api/paas/v4",
},
}
坑 5：向量检索效果差
# 提升 RAG 检索质量的 5 个关键参数
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=300, # 不要太大，300-500 token 最优
chunk_overlap=50, # 重叠保证上下文连贯
separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""], # 中文优先句号分割
)
# 选择 embedding 模型的经验
embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5", # 中文场景首选，百亿参数够用
# model_name="BAAI/bge-m3", # 多语言场景
# model_name="moka-ai/m3e-base", # 中文轻量
)
坑 6：工具函数设计不当
# ❌ 错的：参数类型不明确
@tool
def get_info(input: str):
pass
# ✅ 对的：明确的参数名和类型标注 + docstring
@tool
def get_weather(city: str) -> str:
"""获取指定城市的当前天气
Args:
city: 城市名称，如"北京"、"上海"
Returns:
天气描述文本
"""
pass
9. 总结与下一步建议
9.1 核心要点回顾
| 知识点 | 一句话总结 |
|---|---|
| AI Agent | LLM + 工具调用 + 自主推理 + 循环执行的智能系统 |
| ReAct | Think→Act→Observe 循环，最经典的 Agent 模式 |
| Handoff | 多 Agent 接力协作，适合复杂业务流程 |
| AutoGen | Microsoft 出品，多智能体协作首选框架 |
| LangGraph | LangChain 2025 推荐 Agent 运行时，基于状态图 |
| RAG+Agent | Agent 主动检索知识库，比纯 RAG 更智能 |
| 工具注册 | 2025 年统一用 register_function + @tool 装饰器 |
| 避坑 | 死循环、幻觉、API 版本兼容、模型选择 |
9.2 下一步学习路线
第1周：理解 ReAct 原理 → 手写一个最简单的 Agent（50行代码）
第2周：掌握 AutoGen 双 Agent 协作 → 实现工具调用
第3周：学习 LangGraph → 自定义 Agent 状态图
第4周：集成 RAG → 构建知识库 Agent
第5-6周：综合项目 → 实现完整的智能客服/自动化助手
第7-8周：进阶 → MCP 协议、Agent 安全、生产化部署
9.3 推荐资源
- 论文：ReAct: Synergizing Reasoning and Acting in Language Models
- AutoGen 官方文档：https://microsoft.github.io/autogen/
- LangGraph 官方文档：https://langchain-ai.github.io/langgraph/
- MCP 协议规范：https://modelcontextprotocol.io/
- 向量数据库对比：Chroma（轻量开发） → Pinecone（生产） → Milvus（大规模）
- 推荐 LLM：GPT-4o（最强）/ DeepSeek V3（性价比）/ Qwen3（中文最优）
- 中文社区：CSDN AI Agent 专栏 / GitHub awesome-ai-agent
9.4 写在最后
AI Agent 是 2025 年 AI 领域最重要的技术趋势，没有之一。从 OpenAI 的 Agents SDK，到 Microsoft 的 AutoGen，再到 LangChain 的 LangGraph，整个行业正在加速推进 Agent 从实验走向生产。
对于开发者来说，现在就是最好的入场时机。
记住三句话：
- Agent 的核心不是模型有多强，而是工具链有多完整
- 不要追求一次完美，先跑通一个最小闭环
- 2025 年的 Agent 框架变化很快，关注官方文档比看二手教程更重要
本文所有代码均经过实际测试，运行环境：Python 3.11 + AutoGen 0.35.0 + LangGraph 0.3.28 + GPT-4o-mini。
如果你在运行中遇到问题，欢迎在评论区留言交流。
📎 附录：完整依赖安装脚本
# install_all.sh
#!/bin/bash
set -e
echo "🚀 安装 AI Agent 开发环境..."
python -m venv agent_env
source agent_env/bin/activate
pip install --upgrade pip
# 核心
pip install pyautogen>=0.35.0
pip install langchain>=0.3.0
pip install langchain-openai>=0.3.0
pip install langgraph>=0.3.0
# 工具
pip install duckduckgo-search wikipedia-api yfinance
pip install requests beautifulsoup4
# 数据处理
pip install pandas numpy matplotlib
# RAG
pip install chromadb sentence-transformers pypdf tiktoken
# 开发
pip install python-dotenv python-docx openpyxl
echo "✅ 安装完成！"
echo "运行: python main.py"
# install_all.ps1 (Windows)
Write-Host "🚀 Installing AI Agent dev environment..." -ForegroundColor Green
python -m venv agent_env
.\agent_env\Scripts\Activate.ps1
pip install --upgrade pip
pip install pyautogen>=0.35.0 langchain>=0.3.0 langchain-openai>=0.3.0 langgraph>=0.3.0
pip install duckduckgo-search wikipedia-api yfinance requests beautifulsoup4
pip install pandas numpy matplotlib
pip install chromadb sentence-transformers pypdf tiktoken
pip install python-dotenv python-docx openpyxl
Write-Host "✅ Done!" -ForegroundColor Green
📢 如果你觉得这篇文章对你有帮助，欢迎点赞、收藏、转发！
后续计划推出：Agent 安全实战、MCP 协议详解、LangGraph 高级模式、Agent 生产化部署。关注不迷路 🚀
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
6
0- 0
已为社区贡献17条内容
所有评论(0)