AI Agent（智能体）是大模型落地应用的核心范式。与传统的'一问一答'不同，Agent 能够自主规划任务、调用外部工具、管理记忆上下文、甚至与其他 Agent 协作。本文将基于 Python 生态，从原理到实战，系统讲解如何构建一个生产级 AI Agent。
一、AI Agent 核心架构
1.1 什么是 AI Agent？
AI Agent = LLM（大脑）+ Planning（规划）+ Tools（工具）+ Memory（记忆）
传统 LLM 调用：用户 → Prompt → LLM → 文本回复
AI Agent 调用：用户 → Agent → 规划 → [选工具 → 执行 → 观察] × N → 最终回复
1.2 整体架构图
整体架构图展示了从用户输入到任务规划、工具选择、执行、记忆管理及最终回复的完整流程。主要组件包括任务规划器（ReAct / Plan-and-Execute）、大语言模型（LLM）、工具选择模块（Web Search API, Code Interpreter, SQL/API 调用等）、记忆模块以及执行结果观察机制。
二、技术栈与生态
| 组件 | 推荐方案 | 说明 |
|---|---|---|
| LLM 接口 | OpenAI API / vLLM | 兼容 OpenAI 协议即可 |
| Agent 框架 | LangGraph / OpenAI Agents SDK | 状态机编排，可控性强 |
| 向量数据库 | Chroma / Milvus / Qdrant | 长期记忆与 RAG |
| 工具协议 | Function Calling / MCP | 工具注册与调用 |
| 多 Agent | CrewAI / AutoGen | 角色分工与协作 |
三、从零实现：最小可用 Agent
3.1 ReAct 循环
Agent 的核心是 ReAct（Reasoning + Acting） 循环：
- Thought: 思考下一步
- Action: 选择并执行工具
- Observation: 观察执行结果
- 任务是否完成？
- Final Answer: 生成最终回复
3.2 手写 ReAct Agent（不依赖框架）
""" 最小 ReAct Agent 实现 依赖：pip install openai """
import json
import re
from typing import Callable
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")
# ========== 工具定义 ==========
def get_weather(city:str)->str:
"""查询城市天气（模拟）。"""
weather_db = {
"北京":"晴，气温 18°C，空气质量良好",
"上海":"多云，气温 22°C，有轻微雾霾",
"深圳":"阵雨，气温 28°C，湿度 85%",
"成都":"阴，气温 16°C，预计下午转晴",
}
return weather_db.get(city,f"未找到 {city} 的天气数据")
def search_web(query:str)->str:
"""网络搜索（模拟）。"""
return f"搜索结果：关于「{query}」，以下是最相关的信息...（模拟数据）"
def calculate(expression:str)->str:
"""安全计算数学表达式。"""
allowed = set("0123456789+-*/().% ")
if not all(c in allowed for c in expression):
return "错误：表达式包含非法字符"
try:
result = eval(expression) # 仅允许数学字符
return f"计算结果：{result}"
except Exception as e:
return f"计算错误：{e}"
# ========== 工具注册表 ==========
TOOLS = {
"get_weather": {
"func": get_weather,
"description": "查询指定城市的天气情况",
"params": {"city": "城市名称，如：北京、上海"},
},
"search_web": {
"func": search_web,
"description": "在网络上搜索信息",
"params": {"query": "搜索关键词"},
},
"calculate": {
"func": calculate,
"description": "计算数学表达式",
"params": {"expression": "数学表达式，如：(25 + 37) * 2"},
},
}
SYSTEM_PROMPT = f"""你是一个 AI Agent，可以使用工具来完成任务。
可用工具：{json.dumps({k:{"description": v["description"],"params": v["params"]}for k, v in TOOLS.items()}, ensure_ascii=False, indent=2)}
请严格按照以下格式回复:
Thought: <分析当前情况，思考下一步>
Action: <工具名称>
Action Input: <JSON 格式参数>
或者当你认为任务已完成时:
Thought: <总结分析>
Final Answer: <最终答案>
注意：每次只执行一个工具调用。"""
# ========== Agent 主循环 ==========
class ReactAgent:
def __init__(self, max_iterations:int=10):
self.max_iterations = max_iterations
def run(self, user_query:str)->str:
messages = [
{"role":"system","content": SYSTEM_PROMPT},
{"role":"user","content": user_query},
]
for i in range(self.max_iterations):
response = client.chat.completions.create(
model="qwen-72b",
messages=messages,
temperature=0.3,
max_tokens=1024,
)
reply = response.choices[0].message.content
messages.append({"role":"assistant","content": reply})
print(f"\n--- 第 {i +1} 轮 ---")
print(reply)
# 检查是否给出最终答案
if "Final Answer:" in reply:
return self._extract_final_answer(reply)
# 解析工具调用
action, action_input = self._parse_action(reply)
if not action:
continue
# 执行工具
if action in TOOLS:
tool_result = TOOLS[action]["func"](**action_input)
else:
tool_result = f"错误：未知工具 '{action}'"
observation = f"Observation: {tool_result}"
print(observation)
messages.append({"role":"user","content": observation})
return "Agent 达到最大迭代次数，任务未完成。"
def _parse_action(self, text:str)->tuple:
"""从回复中解析 Action 和 Action Input。"""
action_match = re.search(r"Action:\s*(\w+)", text)
input_match = re.search(r"Action Input:\s*(\{.*?\})", text, re.DOTALL)
if not action_match or not input_match:
return None, None
action = action_match.group(1)
try:
action_input = json.loads(input_match.group(1))
except json.JSONDecodeError:
action_input = {}
return action, action_input
def _extract_final_answer(self, text:str)->str:
match = re.search(r"Final Answer:\s*(.*)", text, re.DOTALL)
return match.group(1).strip() if match else text
# ========== 运行 ==========
if __name__ == "__main__":
agent = ReactAgent(max_iterations=10)
result = agent.run("帮我查一下北京和深圳的天气，然后算一下两个城市的温差是多少。")
print(f"\n{'='*50}")
print(f"最终结果：{result}")
运行示例输出：
--- 第 1 轮 ---
Thought: 用户需要查询两个城市的天气，我先查北京的。
Action: get_weather
Action Input: {"city": "北京"}
Observation: 晴，气温 18°C，空气质量良好
--- 第 2 轮 ---
Thought: 已获取北京天气，再查深圳。
Action: get_weather
Action Input: {"city": "深圳"}
Observation: 阵雨，气温 28°C，湿度 85%
--- 第 3 轮 ---
Thought: 北京 18°C，深圳 28°C，温差 = 28 - 18 = 10°C
Action: calculate
Action Input: {"expression": "28 - 18"}
Observation: 计算结果：10
--- 第 4 轮 ---
Thought: 已获取所有信息，可以回复了。
Final Answer: 北京：晴，18°C；深圳：阵雨，28°C。两城市温差为 10°C。
最终结果：北京：晴，18°C；深圳：阵雨，28°C。两城市温差为 10°C。
四、记忆系统：短期记忆 + 长期记忆
4.1 记忆架构
相似度检索存储对话提取关键事实。短期记忆维护对话历史/滑动窗口，长期记忆通过向量数据库（ChromaDB）进行摘要压缩和相关上下文检索。
4.2 记忆模块实现
""" Agent 记忆系统：短期对话记忆 + 长期向量记忆 依赖：pip install chromadb sentence-transformers """
import hashlib
import json
from datetime import datetime
from typing import Optional
import chromadb
from sentence_transformers import SentenceTransformer
class ShortTermMemory:
"""短期记忆：维护最近 N 轮对话的滑动窗口。"""
def __init__(self, max_rounds:int=20):
self.max_rounds = max_rounds
self.history:list[dict]=[]
def add(self, role:str, content:str):
self.history.append({"role": role,"content": content,"timestamp": datetime.now().isoformat(),})
# 超出窗口时保留系统提示 + 最近对话
if len(self.history)> self.max_rounds:
self.history = self.history[-self.max_rounds:]
def get_messages(self)->list[dict]:
return [{"role": m["role"],"content": m["content"]}for m in self.history]
def clear(self):
self.history.clear()
class LongTermMemory:
"""长期记忆：基于向量数据库的语义检索。"""
def __init__(self, collection_name:str="agent_memory"):
self.embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5")
self.client = chromadb.PersistentClient(path="./chroma_memory")
self.collection = self.client.get_or_create_collection(
name=collection_name,
metadata={"hnsw:space":"cosine"},
)
def _embed(self, text:str)->list[float]:
return self.embedder.encode(text).tolist()
def store(self, content:str, metadata: Optional[dict]=None):
"""存储一条记忆。"""
doc_id = hashlib.md5(content.encode()).hexdigest()[:12]
self.collection.upsert(
ids=[doc_id],
documents=[content],
embeddings=[self._embed(content)],
metadatas=[metadata or{"stored_at": datetime.now().isoformat()}],
)
def retrieve(self, query:str, top_k:int=5)->list[str]:
"""检索与查询最相关的记忆。"""
results = self.collection.query(
query_embeddings=[self._embed(query)],
n_results=top_k,
)
return results["documents"][0] if results["documents"] else []
def store_conversation_summary(self, messages:list[dict]):
"""将一轮完整对话提取为摘要后存储。"""
conversation = json.dumps([{"role": m["role"],"content": m["content"][:200]}for m in messages], ensure_ascii=False,)
self.store(
content=conversation,
metadata={"type":"conversation","rounds":len(messages),"stored_at": datetime.now().isoformat(),},
)
# ========== 组合使用 ==========
class AgentMemory:
"""Agent 统一记忆管理器。"""
def __init__(self):
self.short_term = ShortTermMemory(max_rounds=20)
self.long_term = LongTermMemory()
def build_context(self, user_input:str)->list[dict]:
"""构建注入给 LLM 的完整上下文。"""
# 从长期记忆中检索相关内容
relevant_memories = self.long_term.retrieve(user_input, top_k=3)
context = []
# 注入长期记忆作为系统上下文
if relevant_memories:
memory_text = "\n---\n".join(relevant_memories)
context.append({"role":"system","content":f"以下是与当前对话相关的历史记忆:\n{memory_text}",})
# 注入短期对话历史
context.extend(self.short_term.get_messages())
return context
def save_turn(self, role:str, content:str):
"""保存一轮对话到短期记忆。"""
self.short_term.add(role, content)
def consolidate(self):
"""将短期记忆中的重要信息转入长期记忆。"""
messages = self.short_term.get_messages()
if len(messages)>=10:
self.long_term.store_conversation_summary(messages)
self.short_term.clear()
print("[Memory] 短期记忆已整合到长期记忆")
五、Function Calling：结构化工具调用
5.1 工具注册与调用流程
Function Calling 流程包括用户请求、LLM 判断是否需要工具、直接回复或生成 tool_calls JSON、解析并执行工具、将结果回传 LLM、LLM 生成最终回复。
5.2 基于 OpenAI Function Calling 的工具系统
""" 结构化工具调用系统 使用 OpenAI Function Calling 协议，兼容 vLLM / Ollama / 任何兼容 API """
import json
from typing import Any, Callable
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")
# ========== 工具注册 ==========
class ToolRegistry:
"""工具注册中心，自动生成 OpenAI Function Calling schema。"""
def __init__(self):
self._tools:dict[str,dict]={}
self._handlers:dict[str, Callable]={}
def register(self, name:str, description:str, parameters:dict):
"""装饰器：注册一个工具函数。"""
def decorator(func: Callable):
self._tools[name]={"type":"function","function":{"name": name,"description": description,"parameters": parameters,},}
self._handlers[name]= func
return func
return decorator
def get_schemas(self)->list[dict]:
return list(self._tools.values())
def execute(self, name:str, arguments:dict)->str:
handler = self._handlers.get(name)
if not handler:
return json.dumps({"error":f"未知工具：{name}"}, ensure_ascii=False)
try:
result = handler(**arguments)
return json.dumps(result, ensure_ascii=False) if not isinstance(result,str) else result
except Exception as e:
return json.dumps({"error":str(e)}, ensure_ascii=False)
registry = ToolRegistry()
# ========== 注册具体工具 ==========
@registry.register(
name="query_database",
description="查询 SQL 数据库，返回查询结果",
parameters={"type":"object","properties":{"sql":{"type":"string","description":"SQL 查询语句（仅支持 SELECT）",},"database":{"type":"string","description":"数据库名称","enum":["orders","users","products"],},},"required":["sql","database"],},
)
def query_database(sql:str, database:str)->str:
"""查询数据库（模拟）。"""
# 生产环境应使用 SQLAlchemy 等安全执行
if not sql.strip().upper().startswith("SELECT"):
return "错误：仅允许 SELECT 查询"
return json.dumps({"columns":["id","name","amount"],"rows":[[1,"订单 A",299.00],[2,"订单 B",1599.00],[3,"订单 C",89.50],],"total":3,}, ensure_ascii=False)
@registry.register(
name="send_email",
description="发送电子邮件",
parameters={"type":"object","properties":{"to":{"type":"string","description":"收件人邮箱"},"subject":{"type":"string","description":"邮件主题"},"body":{"type":"string","description":"邮件正文"},},"required":["to","subject","body"],},
)
def send_email(to:str, subject:str, body:str)->str:
"""发送邮件（模拟）。"""
return f"邮件已发送至 {to}，主题：{subject}"
@registry.register(
name="read_file",
description="读取指定文件的内容",
parameters={"type":"object","properties":{"path":{"type":"string","description":"文件路径"},"encoding":{"type":"string","description":"文件编码","default":"utf-8",},},"required":["path"],},
)
def read_file(path:str, encoding:str="utf-8")->str:
"""读取文件内容。"""
try:
with open(path,"r", encoding=encoding)as f:
content = f.read(5000)# 限制读取长度
return content
except FileNotFoundError:
return f"错误：文件不存在 - {path}"
except Exception as e:
return f"错误：{e}"
# ========== Agent 运行 ==========
def run_agent(user_query:str, max_rounds:int=5)->str:
"""运行基于 Function Calling 的 Agent。"""
messages = [
{"role":"system","content":"你是一个智能助手，可以使用工具帮助用户完成任务。请使用中文回复。",},
{"role":"user","content": user_query},
]
for round_num in range(max_rounds):
response = client.chat.completions.create(
model="qwen-72b",
messages=messages,
tools=registry.get_schemas(),
tool_choice="auto",
temperature=0.3,
)
msg = response.choices[0].message
# 没有工具调用 → 直接回复
if not msg.tool_calls:
return msg.content
# 有工具调用 → 逐个执行
messages.append(msg)
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f" [调用工具] {func_name}({json.dumps(func_args, ensure_ascii=False)})")
result = registry.execute(func_name, func_args)
print(f" [执行结果] {result[:200]}")
messages.append({"role":"tool","tool_call_id": tool_call.id,"content": result,})
return "Agent 达到最大调用轮数。"
if __name__ == "__main__":
result = run_agent("帮我查一下 orders 数据库中最近的订单数据，然后把结果发送到 [email protected]，""邮件主题写'本周订单汇总'。")
print(f"\n最终回复:\n{result}")
六、多 Agent 协作
6.1 多 Agent 协作架构
多 Agent 协作架构包含编排者、研究员、程序员和审核员的角色分工。编排者负责任务拆解与分配，研究员产出调研结果，程序员产出代码，审核员提供审核意见，最终输出成果。
6.2 多 Agent 实现
""" 多 Agent 协作系统：研究员 + 程序员 + 审核员 """
import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")
class AgentRole(Enum):
ORCHESTRATOR = "orchestrator"
RESEARCHER = "researcher"
CODER = "coder"
REVIEWER = "reviewer"
@dataclass
class Agent:
name:str
role: AgentRole
system_prompt:str
model:str="qwen-72b"
history:list[dict]= field(default_factory=list)
def chat(self, message:str)->str:
self.history.append({"role":"user","content": message})
messages = [{"role":"system","content": self.system_prompt}]+ self.history
response = client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.3,
max_tokens=2048,
)
reply = response.choices[0].message.content
self.history.append({"role":"assistant","content": reply})
return reply
# ========== 创建 Agent ==========
def create_agents()->dict[str, Agent]:
return {
"orchestrator": Agent(
name="编排者",
role=AgentRole.ORCHESTRATOR,
system_prompt="""你是任务编排者。负责：
1. 将用户任务拆解为子任务
2. 分配给合适的 Agent 执行
3. 汇总结果并判断是否需要修改
4. 以 JSON 格式输出指令：{"action": "assign|finish|revise", "agent": "目标 agent", "task": "任务描述", "summary": "当前进度摘要"}
可用 Agent: researcher(调研), coder(编程), reviewer(审核)
任务完成后 action 设为 "finish" 并在 task 中给出最终结果。""",
),
"researcher": Agent(
name="研究员",
role=AgentRole.RESEARCHER,
system_prompt="你是一位技术研究员，负责对给定主题进行深入调研，给出技术方案、最佳实践和注意事项。回复要结构清晰、有理有据。",
),
"coder": Agent(
name="程序员",
role=AgentRole.CODER,
system_prompt="你是一位资深 Python 工程师。根据需求编写高质量代码，包含完整的类型注解、错误处理和文档字符串。只输出代码和必要的说明。",
),
"reviewer": Agent(
name="审核员",
role=AgentRole.REVIEWER,
system_prompt="""你是代码审核专家。审核代码时关注：
1. 正确性：逻辑是否正确
2. 安全性：是否存在安全漏洞
3. 性能：是否有明显性能问题
4. 可读性：命名、结构是否清晰
回复格式:
- status: "approved" 或 "needs_revision"
- issues: 问题列表（如有）
- suggestions: 改进建议""",
),
}
# ========== 协作流程 ==========
def run_multi_agent(task:str, max_rounds:int=12)->str:
agents = create_agents()
orchestrator = agents["orchestrator"]
current_task = task
task_history = []
for i in range(max_rounds):
print(f"\n{'='*60}")
print(f"第 {i +1} 轮")
print(f"{'='*60}")
# 编排者决策
context = f"原始任务：{task}\n\n执行历史:\n"+"\n".join(task_history) if i >0 else f"原始任务：{task}"
decision_text = orchestrator.chat(context)
print(f"[编排者] {decision_text[:300]}")
# 解析决策
try:
# 从回复中提取 JSON
import re
json_match = re.search(r'\{.*\}', decision_text, re.DOTALL)
if not json_match:
continue
decision = json.loads(json_match.group())
except (json.JSONDecodeError, AttributeError):
continue
action = decision.get("action","")
target_agent = decision.get("agent","")
sub_task = decision.get("task","")
if action == "finish":
return sub_task
if action in("assign","revise")and target_agent in agents:
agent = agents[target_agent]
result = agent.chat(sub_task)
print(f"[{agent.name}] {result[:300]}")
task_history.append(f"- {agent.name} 执行：{sub_task[:100]}... → 结果：{result[:200]}...")
current_task = result
return "多 Agent 协作达到最大轮数。"
# ========== 运行 ==========
if __name__ == "__main__":
task = "开发一个 Python 爬虫，爬取某新闻网站的最新科技新闻标题和摘要，保存为 CSV 文件"
result = run_multi_agent(task)
print(f"\n{'='*60}")
print(f"最终结果:\n{result}")
七、完整实战：带记忆和工具的智能助手
""" 完整 AI Agent：集成记忆 + 工具调用 + ReAct 循环 """
import json
import re
from datetime import datetime
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")
class SmartAgent:
"""集记忆、工具、多轮对话于一体的智能 Agent。"""
def __init__(self, name:str="AI 助手"):
self.name = name
self.memory = AgentMemory()
self.registry = self._setup_tools()
def _setup_tools(self)->dict:
"""注册可用工具。"""
return {
"search": {"handler": self._tool_search,"description":"搜索网络信息","params":{"query":"str"},},
"calculate": {"handler": self._tool_calculate,"description":"执行数学计算","params":{"expression":"str"},},
"get_time": {"handler": self._tool_get_time,"description":"获取当前时间","params":{},},
"save_note": {"handler": self._tool_save_note,"description":"保存笔记到记忆","params":{"content":"str"},},
}
# --- 工具实现 ---
def _tool_search(self, query:str)->str:
return f"搜索「{query}」的结果：...（模拟数据）"
def _tool_calculate(self, expression:str)->str:
allowed = set("0123456789+-*/().% ")
if all(c in allowed for c in expression):
return str(eval(expression))
return "错误：表达式不安全"
def _tool_get_time(self)->str:
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def _tool_save_note(self, content:str)->str:
self.memory.long_term.store(content, metadata={"type":"note"})
return f"已保存笔记（{len(content)} 字）"
# --- 核心对话 ---
def chat(self, user_input:str)->str:
# 构建上下文：长期记忆 + 短期记忆
context = self.memory.build_context(user_input)
context.append({"role":"user","content": user_input})
# 第一轮 LLM 调用（含工具定义）
tools_schema = [
{"type":"function","function":{"name": name,"description": tool["description"],"parameters":{"type":"object","properties":{ k:{"type":"string","description": v}for k, v in tool["params"].items()},"required":list(tool["params"].keys()),},},}
for name, tool in self.registry.items()
]
response = client.chat.completions.create(
model="qwen-72b",
messages=context,
tools=tools_schema,
tool_choice="auto",
temperature=0.5,
)
msg = response.choices[0].message
# 处理工具调用
if msg.tool_calls:
context.append(msg)
for tc in msg.tool_calls:
tool_name = tc.function.name
tool_args = json.loads(tc.function.arguments)
tool_result = self.registry[tool_name]["handler"](**tool_args)
context.append({"role":"tool","tool_call_id": tc.id,"content": tool_result,})
# 第二轮调用，让 LLM 基于工具结果生成回复
response = client.chat.completions.create(
model="qwen-72b",
messages=context,
temperature=0.5,
)
msg = response.choices[0].message
# 保存到记忆
self.memory.save_turn("user", user_input)
self.memory.save_turn("assistant", msg.content)
return msg.content
if __name__ == "__main__":
agent = SmartAgent(name="小智")
# 多轮对话示例
conversations = [
"现在几点了？",
"帮我搜索一下 Python 3.13 的新特性",
"把刚才搜索到的内容保存为笔记",
"我之前保存了什么笔记？", # 会触发长期记忆检索
]
for user_input in conversations:
print(f"\n👤 用户：{user_input}")
reply = agent.chat(user_input)
print(f"🤖 {agent.name}: {reply}")
八、Agent 应用场景与性能对比
关键设计考量
| 维度 | 要点 | 最佳实践 |
|---|---|---|
| 安全性 | 工具执行隔离 | 沙箱环境、权限最小化、输入校验 |
| 可控性 | 限制 Agent 行为边界 | 设定白名单工具、最大迭代次数 |
| 可观测性 | 记录 Agent 决策过程 | 日志记录每轮 Thought/Action/Observation |
| 成本 | Token 消耗控制 | 压缩历史、摘要记忆、缓存频繁查询 |
| 延迟 | 减少不必要的工具调用 | 优化 ReAct 轮数、并行执行独立工具 |
| 可靠性 | 处理 LLM 输出不稳定 | 结构化输出（JSON Schema）、重试机制 |
九、总结
本文从零到一构建了一个完整的 AI Agent 系统，核心要点：
- ReAct 是 Agent 的灵魂 — Think → Act → Observe 循环使 LLM 具备自主推理能力
- 工具是 Agent 的双手 — Function Calling 提供了结构化、可靠的外部交互机制
- 记忆是 Agent 的经验 — 短期记忆维持对话连贯，长期记忆实现知识积累
- 多 Agent 是 Agent 的团队 — 角色分工、协作编排解决复杂任务
- 安全与可观测是底线 — 生产环境必须做工具隔离、日志追踪和成本控制
