【无标题】
·
AI Agent 开发完全教程
从零到一构建智能体系统 · 2026年版
涵盖核心概念、框架选型、实战代码、部署最佳实践
目录
- 什么是 AI Agent
- Agent 核心架构解析
- 主流开发框架对比与选型
- 开发环境搭建
- 实战：从零构建 ReAct Agent
- 进阶：多 Agent 协作系统
- Agent 记忆系统设计与实现
- 工具调用系统
- 部署与生产化
- 最佳实践与常见坑
1. 什么是 AI Agent
1.1 定义
AI Agent（智能体） 是一个以大语言模型（LLM）为核心控制器，能够自主感知环境、制定计划、调用工具、执行行动并记忆经验的软件系统。它不只是"问答机器人"，而是一个目标驱动的自主执行系统。
“LLM is the brain, but the agent is the body.”
1.2 Agent vs 传统 Chatbot
| 维度 | 传统 Chatbot | AI Agent |
|---|---|---|
| 交互方式 | 一问一答 | 多步推理 + 自主执行 |
| 工具使用 | 无 | 调用 API/代码/数据库等 |
| 记忆能力 | 当前对话 | 长期记忆 + 向量检索 |
| 规划能力 | 无 | 任务分解 + 自我反思 |
| 自主性 | 被动响应 | 主动推进目标 |
1.3 典型应用场景
- 代码开发助手：理解需求 → 写代码 → 调试 → 提 PR
- 自动化数据分析：连接数据源 → 清洗 → 分析 → 出报告
- 客服系统：理解问题 → 查知识库 → 调用工单系统 → 回复
- 研究助手：搜索文献 → 摘要 → 关联分析 → 生成综述
- 自动化运维：监控告警 → 排查 → 执行修复 → 验证
2. Agent 核心架构解析
一个完整的 AI Agent 系统由三大核心组件构成：
┌─────────────────────────────────────────────┐
│ AI Agent │
│ │
│ ┌──────────┐ ┌──────────┐ │
│ │ Planning │ │ Memory │ │
│ │ (规划) │ │ (记忆) │ │
│ │ · 任务分解 │ │ · 短期 │ │
│ │ · 反思 │ │ · 长期 │ │
│ │ · 自评 │ │ · 向量 │ │
│ └──────────┘ └──────────┘ │
│ │
│ ┌──────────┐ ┌──────────┐ │
│ │ Tool Use │ │ 感知 │ │
│ │ (工具) │ │ (Perception)│ │
│ │ · API │ │ · 环境输入│ │
│ │ · 代码 │ │ · 用户输入│ │
│ │ · 搜索 │ │ · 反馈 │ │
│ └──────────┘ └──────────┘ │
│ │
│ ┌──────────────────┐ │
│ │ LLM Core (大脑) │ │
│ └──────────────────┘ │
└─────────────────────────────────────────────┘
2.1 Planning（规划系统）
任务分解（Task Decomposition）
Agent 将复杂任务拆解为可执行的子步骤。常用方法：
- Chain of Thought (CoT)：引导模型逐步思考
- Tree of Thoughts (ToT)：在每一步探索多个推理分支
- Plan-and-Execute：先制定完整计划，再逐步执行
# 任务分解示例（伪代码）
prompt = f"""
用户目标: {user_goal}
请将上述目标分解为可执行的步骤，每一步应包含：
1. 步骤描述
2. 需要使用的工具
3. 预期输出
输出格式：
步骤1: ...
步骤2: ...
...
"""
自我反思（Self-Reflection）
Agent 在执行过程中不断评估自身行为并修正：
def reflect(history, last_action, last_result):
prompt = f"""
你刚刚执行了: {last_action}
得到结果: {last_result}
请评估：
1. 这个结果是否符合预期？
2. 是否有更好的执行方式？
3. 下一步应该做什么？
"""
return llm.invoke(prompt)
2.2 Memory（记忆系统）
| 记忆类型 | 实现方式 | 生命周期 | 用途 |
|---|---|---|---|
| 短期记忆 | In-Context（上下文窗口） | 单轮对话 | 当前任务上下文 |
| 长期记忆 | 向量数据库（如 Chroma/Pinecone） | 持久化 | 跨会话知识 |
| 工作记忆 | 文件 / Redis | 任务期间 | 中间结果暂存 |
2.3 Tool Use（工具使用）
Agent 通过工具与外部世界交互。工具的本质是一个函数 + Schema 描述：
class Tool:
name: str # 工具名称
description: str # 工具描述（LLM 理解用）
parameters: dict # 参数 JSON Schema
fn: Callable # 实际执行函数
3. 主流开发框架对比与选型
3.1 框架全景图
| 框架 | 开发者 | 星标 | 核心思想 | 适合场景 |
|---|---|---|---|---|
| LangChain | LangChain | 100k+ | 链式组合 + 工具集成 | 通用 Agent 开发 |
| LangGraph | LangChain | 15k+ | 有向图状态机 | 复杂状态、长任务 |
| CrewAI | CrewAI | 25k+ | 角色化多 Agent 团队 | 多角色协作 |
| AutoGen | Microsoft | 60k+ | 多 Agent 对话编排 | 研究、代码生成 |
| MetaGPT | 开源 | 45k+ | 软件公司模拟 | 自动化软件开发 |
| OpenAI Agents SDK | OpenAI | 少量 | 轻量 Agent 实现 | 快速原型 |
| Semantic Kernel | Microsoft | 25k+ | 企业级 .NET/Python | 企业集成 |
3.2 详细对比
| 维度 | LangChain | LangGraph | CrewAI | AutoGen |
|---|---|---|---|---|
| 上手难度 | ⭐⭐⭐ 中等 | ⭐⭐⭐⭐ 较难 | ⭐⭐ 简单 | ⭐⭐⭐ 中等 |
| 灵活性 | 高 | 极高 | 中 | 高 |
| 多 Agent | 有限 | 原生支持 | 原生支持 | 原生支持 |
| 状态管理 | 简单链 | 图状态机 | 流程控制 | 对话轮次 |
| 工具生态 | 最丰富 | 同 LangChain | 中等 | 中等 |
| 企业级 | 是 | 是 | 部分 | 是 |
| 适合场景 | 快速开发 | 生产级 | 团队协作 | 研究探索 |
3.3 选型决策树
你的任务是什么？
├── 单个 Agent + 工具调用 → LangChain / OpenAI SDK
├── 复杂状态/长任务流程 → LangGraph
├── 多角色团队协作 → CrewAI
├── 自动化软件开发 → MetaGPT
└── 研究性多 Agent 对话 → AutoGen
3.4 生态评分（2026 实测）
| 框架 | 上手难度(10分制) | 工具生态 | 部署友好度 | 社区活跃度 |
|---|---|---|---|---|
| CrewAI | 9.2 | ⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐ |
| LangGraph | 6.5 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
| AutoGen | 7.0 | ⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ |
| LangChain | 7.5 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
4. 开发环境搭建
4.1 基础环境
# Python 3.10+
python -m venv agent_env
source agent_env/bin/activate
# 安装核心依赖
pip install openai # LLM 调用
pip install langchain # LangChain 框架
pip install langgraph # LangGraph（可选）
pip install crewai # CrewAI（可选）
pip install pyautogen # AutoGen（可选）
4.2 LLM 配置
# config.py
import os
from openai import OpenAI
# OpenAI API
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 或使用本地模型（Ollama）
# client = OpenAI(
# base_url="http://localhost:11434/v1",
# api_key="ollama"
# )
4.3 向量数据库（记忆用）
# 轻量方案：ChromaDB
pip install chromadb
# 生产方案：Qdrant / Pinecone / Weaviate
5. 实战：从零构建 ReAct Agent
ReAct（Reasoning + Acting）是最经典的 Agent 模式，让模型交替进行推理和行动。
5.1 定义工具集
# tools.py
import json
import requests
from datetime import datetime
class Tool:
def __init__(self, name, description, parameters, fn):
self.name = name
self.description = description
self.parameters = parameters # JSON Schema
self.fn = fn
def run(self, **kwargs):
return self.fn(**kwargs)
# 工具1：获取当前时间
def get_current_time():
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
time_tool = Tool(
name="get_time",
description="获取当前日期和时间",
parameters={
"type": "object",
"properties": {},
"required": []
},
fn=get_current_time
)
# 工具2：网页搜索
def web_search(query: str):
response = requests.get(
f"https://api.duckduckgo.com/?q={query}&format=json"
)
return response.text[:2000]
search_tool = Tool(
name="web_search",
description="搜索网络信息，参数为搜索关键词",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
},
fn=web_search
)
# 工具3：计算器
def calculator(expression: str):
return str(eval(expression))
calc_tool = Tool(
name="calculator",
description="执行数学计算，传入数学表达式字符串",
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
},
fn=calculator
)
# 注册所有工具
TOOLS = [time_tool, search_tool, calc_tool]
TOOL_MAP = {t.name: t for t in TOOLS}
5.2 构建 Agent 核心循环
# agent.py
import json
from openai import OpenAI
client = OpenAI()
def build_system_prompt(tools):
"""构建系统提示词"""
tool_descriptions = "\n".join([
f"- {t.name}: {t.description}\n 参数: {json.dumps(t.parameters, indent=2)}"
for t in tools
])
return f"""你是一个 AI Agent，拥有以下工具可以使用：
{tool_descriptions}
工作流程：
1. 分析用户请求，决定是否需要使用工具
2. 如果需要工具，输出格式为：
THOUGHT: 思考当前情况
ACTION: {{"name": "工具名", "args": {{...}}}}
3. 得到工具结果后，继续推理
4. 最终输出答案给用户
规则：
- 一次只能调用一个工具
- 信息足够时直接回答，不要继续调用工具
- 如果工具返回错误，尝试其他方式解决
"""
def parse_action(text):
"""从模型输出中解析工具调用"""
if "ACTION:" not in text:
return None
action_part = text.split("ACTION:")[-1].strip()
try:
return json.loads(action_part)
except:
return None
def run_agent(user_input, max_steps=10):
"""运行 Agent 主循环"""
messages = [
{"role": "system", "content": build_system_prompt(TOOLS)},
{"role": "user", "content": user_input}
]
for step in range(max_steps):
# 调用 LLM
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0.2
)
content = response.choices[0].message.content
print(f"\n[Step {step}] LLM:\n{content}")
# 检查是否需要调用工具
action = parse_action(content)
if action is None:
# 没有工具调用，直接返回最终答案
return content
# 执行工具调用
tool_name = action["name"]
tool_args = action.get("args", {})
if tool_name in TOOL_MAP:
try:
result = TOOL_MAP[tool_name].run(**tool_args)
print(f" → 工具 '{tool_name}' 返回: {result[:200]}")
except Exception as e:
result = f"工具调用失败: {str(e)}"
else:
result = f"工具 '{tool_name}' 不存在"
# 将推理过程和工具结果加入对话
messages.append({"role": "assistant", "content": content})
messages.append({"role": "user", "content": f"工具 '{tool_name}' 返回结果: {result}"})
return "达到最大步数限制，任务结束。"
5.3 运行示例
# main.py
from agent import run_agent
# 示例1：多步推理
result = run_agent("今天是几号？搜索一下今天AI圈有什么大事")
print(f"\n最终结果:\n{result}")
# 示例2：复杂计算
result = run_agent("计算 (1452 * 378) + 89234 等于多少")
print(f"\n最终结果:\n{result}")
5.4 使用 LangChain 快速实现（16 行代码）
from langchain.tools import tool
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
@tool
def get_time() -> str:
"""获取当前日期时间"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@tool
def calculator(expression: str) -> str:
"""执行数学计算"""
return str(eval(expression))
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [get_time, calculator]
agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "今天几号？1452*378+89234等于多少？"})
print(result["output"])
6. 进阶：多 Agent 协作系统
6.1 多 Agent 架构模式
┌──────────────────────────────────────────┐
│ Orchestrator（协调者） │
│ 分解任务 → 分配 → 聚合结果 → 输出 │
└──────┬──────────┬──────────┬────────────┘
│ │ │
┌────▼───┐ ┌────▼───┐ ┌────▼───┐
│ Agent A │ │ Agent B │ │ Agent C │
│ 搜索专家 │ │ 分析专家 │ │ 写作专家 │
└────────┘ └────────┘ └────────┘
6.2 CrewAI 实现（推荐入门）
from crewai import Agent, Task, Crew, Process
# 定义 Agent
researcher = Agent(
role="研究员",
goal="搜索和收集相关信息",
backstory="你是一名专业研究员，擅长从网络中快速找到准确信息",
tools=[search_tool],
verbose=True
)
analyst = Agent(
role="分析师",
goal="分析数据并得出结论",
backstory="你是一名资深分析师，擅长从数据中发现洞察",
verbose=True
)
writer = Agent(
role="写作者",
goal="撰写清晰易懂的报告",
backstory="你是一名专业撰稿人，擅长将复杂信息转化为易懂的文字",
verbose=True
)
# 定义任务
research_task = Task(
description="搜索并整理关于'{topic}'的最新信息",
agent=researcher,
expected_output="一份包含关键事实和数据的摘要"
)
analysis_task = Task(
description="基于搜索结果进行分析，给出核心洞察",
agent=analyst,
expected_output="一份包含分析和建议的报告"
)
writing_task = Task(
description="将分析结果写成一篇完整的文章",
agent=writer,
expected_output="一篇结构清晰、内容完整的文章"
)
# 组建 Crew
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.sequential # 顺序执行
)
# 启动
result = crew.kickoff(inputs={"topic": "2026年AI Agent发展趋势"})
print(result)
6.3 LangGraph 实现（生产级）
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
# 定义状态
class AgentState(TypedDict):
messages: List
next_agent: str
# 定义 Agent 节点
def researcher_node(state: AgentState):
llm = ChatOpenAI(model="gpt-4o")
# ... 研究逻辑
return {"messages": state["messages"] + [new_msg], "next_agent": "analyst"}
def analyst_node(state: AgentState):
# ... 分析逻辑
return {"messages": state["messages"] + [new_msg], "next_agent": "writer"}
def writer_node(state: AgentState):
# ... 写作逻辑
return {"messages": state["messages"] + [new_msg], "next_agent": "END"}
# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("analyst", analyst_node)
workflow.add_node("writer", writer_node)
workflow.set_entry_point("researcher")
workflow.add_conditional_edges("researcher", lambda s: s["next_agent"])
workflow.add_conditional_edges("analyst", lambda s: s["next_agent"])
workflow.add_edge("writer", END)
app = workflow.compile()
6.4 AutoGen 实现（研究型）
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# 定义 Agent
researcher = AssistantAgent(
name="Researcher",
system_message="你是研究专家，负责搜索和收集信息。",
llm_config={"config_list": [{"model": "gpt-4o", "api_key": "..."}]}
)
analyst = AssistantAgent(
name="Analyst",
system_message="你是分析专家，负责分析数据。",
llm_config={"config_list": [{"model": "gpt-4o", "api_key": "..."}]}
)
writer = AssistantAgent(
name="Writer",
system_message="你是写作专家，负责撰写报告。",
llm_config={"config_list": [{"model": "gpt-4o", "api_key": "..."}]}
)
# 构建群聊
group_chat = GroupChat(
agents=[researcher, analyst, writer],
messages=[],
max_round=10
)
manager = GroupChatManager(groupchat=group_chat)
# 启动
user = UserProxyAgent(name="User", human_input_mode="NEVER")
user.initiate_chat(manager, message="分析2026年AI Agent发展趋势")
7. Agent 记忆系统设计与实现
7.1 记忆架构
短期记忆（上下文窗口）
↓ 重要信息提取
工作记忆（当前任务缓存）
↓ 持久化
长期记忆（向量数据库）
7.2 使用 ChromaDB 实现长期记忆
# memory.py
import chromadb
from chromadb.utils import embedding_functions
class AgentMemory:
def __init__(self, collection_name="agent_memory"):
self.client = chromadb.Client()
self.collection = self.client.get_or_create_collection(
name=collection_name,
embedding_function=embedding_functions.DefaultEmbeddingFunction()
)
def remember(self, content: str, metadata: dict = None):
"""存储记忆"""
self.collection.add(
documents=[content],
metadatas=[metadata or {}],
ids=[f"mem_{int(time.time())}_{hash(content) % 10000}"]
)
def recall(self, query: str, n_results: int = 5):
"""根据查询召回相关记忆"""
results = self.collection.query(
query_texts=[query],
n_results=n_results
)
return results['documents'][0] if results['documents'] else []
def forget(self, older_than_days: int = 30):
"""清理过期记忆"""
# 可按时间戳过滤删除
pass
# 使用示例
memory = AgentMemory()
memory.remember("用户偏好Python语言", {"type": "preference"})
memory.remember("已完成项目：数据分析Agent", {"type": "project"})
# 召回
relevant = memory.recall("用户喜欢什么语言")
print(relevant)
7.3 记忆增强的 Agent
class MemoryAgent:
def __init__(self):
self.memory = AgentMemory()
self.short_term = [] # 短期记忆
def process(self, user_input: str):
# 1. 检索相关长期记忆
past_memories = self.memory.recall(user_input)
# 2. 构建上下文（短期 + 长期）
context = {
"short_term": self.short_term[-5:],
"long_term": past_memories
}
# 3. 执行推理（略）
response = self._llm_inference(user_input, context)
# 4. 更新记忆
self.short_term.append({"role": "user", "content": user_input})
self.short_term.append({"role": "assistant", "content": response})
self.memory.remember(f"对话: {user_input} -> {response}")
return response
8. 工具调用系统
8.1 工具注册模式
# tool_registry.py
from functools import wraps
class ToolRegistry:
"""全局工具注册器"""
_tools = {}
@classmethod
def register(cls, name=None, description=None, schema=None):
def decorator(func):
tool_name = name or func.__name__
cls._tools[tool_name] = {
"fn": func,
"name": tool_name,
"description": description or func.__doc__ or "",
"parameters": schema or cls._infer_schema(func)
}
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return decorator
@classmethod
def get_all(cls):
return list(cls._tools.values())
@classmethod
def call(cls, name, **kwargs):
if name not in cls._tools:
raise ValueError(f"工具 '{name}' 未注册")
return cls._tools[name]["fn"](**kwargs)
# 使用
@ToolRegistry.register(description="发送HTTP GET请求")
def http_get(url: str) -> str:
import requests
return requests.get(url, timeout=10).text
@ToolRegistry.register(description="执行Python代码")
def run_python(code: str) -> str:
try:
exec_globals = {}
exec(code, exec_globals)
return str(exec_globals.get("result", "执行完成"))
except Exception as e:
return f"执行错误: {e}"
8.2 工具安全沙箱
# safe_executor.py
import ast
import sys
class SandboxedExecutor:
"""安全执行环境"""
ALLOWED_MODULES = {"math", "datetime", "json", "random", "statistics"}
@classmethod
def execute(cls, code: str, timeout: int = 5):
# 1. AST 安全检查
try:
tree = ast.parse(code)
for node in ast.walk(tree):
if isinstance(node, ast.Call):
if isinstance(node.func, ast.Attribute):
if 'system' in node.func.attr.lower() or 'exec' in node.func.attr.lower():
return "安全限制：禁止调用系统相关函数"
except SyntaxError as e:
return f"语法错误: {e}"
# 2. 受限环境执行
restricted_globals = {
"__builtins__": {
"print": print, "len": len, "range": range,
"int": int, "float": float, "str": str, "list": list,
"dict": dict, "tuple": tuple, "set": set, "bool": bool,
"max": max, "min": min, "sum": sum, "abs": abs,
"sorted": sorted, "reversed": reversed, "enumerate": enumerate,
"zip": zip, "map": map, "filter": filter, "type": type,
"isinstance": isinstance, "hasattr": hasattr, "getattr": getattr,
"True": True, "False": False, "None": None,
}
}
try:
exec(code, restricted_globals)
return restricted_globals.get("result", "执行完成")
except Exception as e:
return f"执行错误: {e}"
9. 部署与生产化
9.1 项目结构
agent_project/
├── agent/
│ ├── __init__.py
│ ├── core.py # Agent 主循环
│ ├── tools/ # 工具模块
│ │ ├── __init__.py
│ │ ├── search.py
│ │ ├── code_exec.py
│ │ └── database.py
│ ├── memory/ # 记忆模块
│ │ ├── __init__.py
│ │ ├── short_term.py
│ │ └── long_term.py
│ └── llm/ # LLM 客户端
│ ├── __init__.py
│ └── client.py
├── api/ # API 服务
│ ├── __init__.py
│ ├── routes.py
│ └── schemas.py
├── config/
│ ├── settings.py
│ └── .env.example
├── tests/
├── Dockerfile
├── docker-compose.yml
└── requirements.txt
9.2 API 服务（FastAPI）
# api/routes.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent.core import Agent
app = FastAPI(title="Agent API")
agent = Agent()
class ChatRequest(BaseModel):
message: str
session_id: str = "default"
stream: bool = False
class ChatResponse(BaseModel):
response: str
steps: list = []
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
try:
result = agent.run(
request.message,
session_id=request.session_id
)
return ChatResponse(response=result["output"])
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/reset")
async def reset_session(session_id: str = "default"):
agent.clear_memory(session_id)
return {"status": "ok"}
@app.get("/tools")
async def list_tools():
return agent.get_available_tools()
9.3 Docker 部署
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制代码
COPY . .
# 暴露端口
EXPOSE 8000
# 启动
CMD ["uvicorn", "api.routes:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
version: '3.8'
services:
agent-api:
build: .
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- CHROMA_DB_URL=http://chromadb:8000
depends_on:
- chromadb
chromadb:
image: chromadb/chroma:latest
ports:
- "8001:8000"
volumes:
- chroma_data:/chroma/data
volumes:
chroma_data:
9.4 监控与日志
# monitoring.py
import logging
import time
from datetime import datetime
class AgentMonitor:
"""Agent 监控器"""
def __init__(self):
self.logger = logging.getLogger("agent")
self.metrics = {
"total_calls": 0,
"total_tokens": 0,
"tool_calls": {},
"errors": []
}
def log_call(self, func_name: str, duration: float, tokens: int):
self.metrics["total_calls"] += 1
self.metrics["total_tokens"] += tokens
self.metrics["tool_calls"][func_name] = \
self.metrics["tool_calls"].get(func_name, 0) + 1
def log_error(self, error: str):
self.metrics["errors"].append({
"time": datetime.now().isoformat(),
"error": error
})
self.logger.error(error)
def report(self):
return {
**self.metrics,
"avg_tokens_per_call":
self.metrics["total_tokens"] / max(self.metrics["total_calls"], 1)
}
10. 最佳实践与常见坑
✅ 最佳实践
10.1 提示词工程
✅ 好的 Agent 提示词模板：
[角色定义]
你是一个 [具体角色]，擅长 [具体能力]。
[工作流程]
1. 分析：理解用户真实需求
2. 计划：制定执行步骤
3. 执行：调用合适的工具
4. 综合：整合所有结果
[工具说明]
工具列表及使用规则...
[约束条件]
- 一次只调用一个工具
- 信息足够时直接回答
- 出错时尝试其他方案
10.2 错误处理
def robust_tool_call(tool_fn, max_retries=3, **kwargs):
"""带重试机制的工具调用"""
for i in range(max_retries):
try:
return tool_fn(**kwargs)
except RateLimitError:
wait = 2 ** i
time.sleep(wait)
except TemporaryError:
time.sleep(1)
except PermanentError:
return f"工具不可用: {kwargs}"
return "多次重试后仍然失败"
10.3 成本控制
class CostManager:
"""Token 成本管理"""
PRICING = {
"gpt-4o": {"input": 2.5/1e6, "output": 10/1e6},
"gpt-4o-mini": {"input": 0.15/1e6, "output": 0.6/1e6},
}
def __init__(self, model="gpt-4o"):
self.model = model
self.total_input = 0
self.total_output = 0
def track(self, input_tokens: int, output_tokens: int):
self.total_input += input_tokens
self.total_output += output_tokens
@property
def cost(self):
p = self.PRICING[self.model]
return self.total_input * p["input"] + self.total_output * p["output"]
def summary(self):
return {
"model": self.model,
"input_tokens": self.total_input,
"output_tokens": self.total_output,
"total_cost_usd": round(self.cost, 4)
}
10.4 安全清单
| 项目 | 检查项 | 优先级 |
|---|---|---|
| 提示词注入 | 过滤用户输入中的系统指令覆盖 | 🔴 必做 |
| 工具沙箱 | 代码执行使用隔离环境 | 🔴 必做 |
| 速率限制 | 防止 API 滥用 | 🟡 建议 |
| 日志审计 | 记录所有 Agent 操作 | 🟡 建议 |
| 权限控制 | 工具最小权限原则 | 🔴 必做 |
| 数据脱敏 | 用户敏感信息过滤 | 🔴 必做 |
❌ 常见坑
| 坑 | 表现 | 解决方案 |
|---|---|---|
| 循环调用 | Agent 不停调用工具停不下来 | 设置最大步数 + 循环检测 |
| 幻觉工具 | LLM 调用不存在的工具 | strict tool binding + schema 校验 |
| 上下文爆炸 | Token 用尽，丢失上下文 | 窗口滑动 + 摘要压缩 |
| 提示词注入 | 用户让 Agent 忽略指令 | 输入过滤 + 系统提示加固 |
| 工具死锁 | 多 Agent 互相等待 | 设置超时 + 协调者仲裁 |
| 成本失控 | 复杂任务消耗大量 Token | 预算限制 + 模型分级 |
附录
A. 推荐学习资源
| 资源 | 类型 | 链接 |
|---|---|---|
| Lilian Weng Agent 综述 | 博客 | lilianweng.github.io |
| LangChain 官方教程 | 文档 | python.langchain.com |
| CrewAI 文档 | 文档 | docs.crewai.com |
| AutoGen 示例 | GitHub | github.com/microsoft/autogen |
| LangGraph 教程 | 文档 | langchain-ai.github.io/langgraph |
| OpenAI Function Calling | 文档 | platform.openai.com |
B. 常用工具模板
# 文件操作工具
@ToolRegistry.register(description="读取文件内容")
def read_file(path: str) -> str:
with open(path, 'r') as f:
return f.read()
# 数据库查询工具
@ToolRegistry.register(description="执行 SQL 查询")
def query_database(sql: str) -> list:
import sqlite3
conn = sqlite3.connect("data.db")
return conn.execute(sql).fetchall()
# 图像理解工具
@ToolRegistry.register(description="分析图片内容")
def analyze_image(url: str) -> str:
# 调用多模态模型
pass
C. 框架速查表
需要做什么？ → 选什么框架？
─────────────────────────────────────────────
单 Agent + 工具链 → LangChain / OpenAI SDK
复杂状态机 + 长任务 → LangGraph
多角色团队协作 → CrewAI
多 Agent 自由对话 → AutoGen
自动化软件开发 → MetaGPT
企业 .NET 集成 → Semantic Kernel
简单快速原型 → OpenAI Agents SDK
文档版本: v1.0 | 最后更新: 2026-07-15
本教程涵盖了从 Agent 基础概念到生产部署的完整链路。建议按照章节顺序学习，每章都配有可直接运行的代码示例。如有问题，欢迎交流。
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
1
0- 0
已为社区贡献2条内容
所有评论(0)