系统讲解基于 Python 构建生产级 AI Agent 的方法。涵盖核心架构（LLM+ 规划 + 工具 + 记忆）、ReAct 循环原理、手写 ReAct Agent 实现、短期与长期记忆系统设计、Function Calling 工具调用机制、多 Agent 协作模式以及完整实战案例。重点介绍了如何利用 LangGraph、ChromaDB 等组件搭建具备自主决策能力的智能助手，并提供了安全性、可控性及性能优化的关键设计考量。
AI Agent（智能体）是大模型落地应用的核心范式。与传统的'一问一答'不同，Agent 能够自主规划任务、调用外部工具、管理记忆上下文，甚至与其他 Agent 协作。本文将基于 Python 生态，从原理到实战，系统讲解如何构建一个生产级 AI Agent。
AI Agent = LLM（大脑）+ Planning（规划）+ Tools（工具）+ Memory（记忆）
系统主要包含以下组件和流程：
根据 2026 年预测，Python Agent 开发框架使用占比如下：
| 框架 | 占比 |
|---|---|
| LangChain / LangGraph | 35% |
| OpenAI Agents SDK | 20% |
| CrewAI | 12% |
| AutoGen | 10% |
| Dify (Python SDK) | 10% |
| 自研 / 裸调 API | 8% |
| 其他 | 5% |
推荐技术栈方案：
| 组件 | 推荐方案 | 说明 |
|---|---|---|
| LLM 接口 | OpenAI API / vLLM | 兼容 OpenAI 协议即可 |
| Agent 框架 | LangGraph / OpenAI Agents SDK | 状态机编排，可控性强 |
| 向量数据库 | Chroma / Milvus / Qdrant | 长期记忆与 RAG |
| 工具协议 | Function Calling / MCP | 工具注册与调用 |
| 多 Agent | CrewAI / AutoGen | 角色分工与协作 |
Agent 的核心是 ReAct（Reasoning + Acting） 循环：
"""
最小 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):
:
result = (expression)
Exception e:
TOOLS = {
: {
: get_weather,
: ,
: {: },
},
: {
: search_web,
: ,
: {: },
},
: {
: calculate,
: ,
: {: },
},
}
SYSTEM_PROMPT =
:
():
.max_iterations = max_iterations
() -> :
messages = [
{: , : SYSTEM_PROMPT},
{: , : user_query},
]
i (.max_iterations):
response = client.chat.completions.create(
model=,
messages=messages,
temperature=,
max_tokens=,
)
reply = response.choices[].message.content
messages.append({: , : reply})
()
(reply)
reply:
._extract_final_answer(reply)
action, action_input = ._parse_action(reply)
action:
action TOOLS:
tool_result = TOOLS[action][](**action_input)
:
tool_result =
observation =
(observation)
messages.append({: , : observation})
() -> :
action_match = re.search(, text)
input_match = re.search(, text, re.DOTALL)
action_match input_match:
,
action = action_match.group()
:
action_input = json.loads(input_match.group())
json.JSONDecodeError:
action_input = {}
action, action_input
() -> :
= re.search(, text, re.DOTALL)
.group().strip() text
__name__ == :
agent = ReactAgent(max_iterations=)
result = agent.run()
()
()
运行示例输出：
--- 第 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: 已获取所有信息，可以回复了。
"""
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"]} m .history]
():
.history.clear()
:
():
.embedder = SentenceTransformer()
.client = chromadb.PersistentClient(path=)
.collection = .client.get_or_create_collection(
name=collection_name,
metadata={: },
)
() -> []:
.embedder.encode(text).tolist()
():
doc_id = hashlib.md5(content.encode()).hexdigest()[:]
.collection.upsert(
ids=[doc_id],
documents=[content],
embeddings=[._embed(content)],
metadatas=[metadata {: datetime.now().isoformat()}],
)
() -> []:
results = .collection.query(
query_embeddings=[._embed(query)],
n_results=top_k,
)
results[][] results[] []
():
conversation = json.dumps([{: m[], : m[][:]} m messages], ensure_ascii=)
.store(
content=conversation,
metadata={: , : (messages), : datetime.now().isoformat()},
)
:
():
.short_term = ShortTermMemory(max_rounds=)
.long_term = LongTermMemory()
() -> []:
relevant_memories = .long_term.retrieve(user_input, top_k=)
context = []
relevant_memories:
memory_text = .join(relevant_memories)
context.append({: , : })
context.extend(.short_term.get_messages())
context
():
.short_term.add(role, content)
():
messages = .short_term.get_messages()
(messages) >= :
.long_term.store_conversation_summary(messages)
.short_term.clear()
()
tool_calls JSON。"""
结构化工具调用系统
使用 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 () -> []:
(._tools.values())
() -> :
handler = ._handlers.get(name)
handler:
json.dumps({: }, ensure_ascii=)
:
result = handler(**arguments)
json.dumps(result, ensure_ascii=) (result, ) result
Exception e:
json.dumps({: (e)}, ensure_ascii=)
registry = ToolRegistry()
() -> :
sql.strip().upper().startswith():
json.dumps({
: [, , ],
: [[, , ], [, , ], [, , ]],
: ,
}, ensure_ascii=)
() -> :
() -> :
:
(path, , encoding=encoding) f:
content = f.read()
content
FileNotFoundError:
Exception e:
() -> :
messages = [
{: , : },
{: , : user_query},
]
round_num (max_rounds):
response = client.chat.completions.create(
model=,
messages=messages,
tools=registry.get_schemas(),
tool_choice=,
temperature=,
)
msg = response.choices[].message
msg.tool_calls:
msg.content
messages.append(msg)
tool_call msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
()
result = registry.execute(func_name, func_args)
()
messages.append({: , : tool_call., : result})
__name__ == :
result = run_agent()
()
典型的多 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[].message.content
.history.append({: , : reply})
reply
() -> [, Agent]:
{
: Agent(
name=,
role=AgentRole.ORCHESTRATOR,
system_prompt=,
),
: Agent(
name=,
role=AgentRole.RESEARCHER,
system_prompt=,
),
: Agent(
name=,
role=AgentRole.CODER,
system_prompt=,
),
: Agent(
name=,
role=AgentRole.REVIEWER,
system_prompt=,
),
}
() -> :
agents = create_agents()
orchestrator = agents[]
current_task = task
task_history = []
i (max_rounds):
()
()
()
context = + .join(task_history) i >
decision_text = orchestrator.chat(context)
()
:
re
json_match = re.search(, decision_text, re.DOTALL)
json_match:
decision = json.loads(json_match.group())
(json.JSONDecodeError, AttributeError):
action = decision.get(, )
target_agent = decision.get(, )
sub_task = decision.get(, )
action == :
sub_task
action (, ) target_agent agents:
agent = agents[target_agent]
result = agent.chat(sub_task)
()
task_history.append()
current_task = result
__name__ == :
task =
result = run_multi_agent(task)
()
()
"""
完整 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": {: ._tool_save_note, : , : {: }},
}
() -> :
() -> :
allowed = ()
(c allowed c expression):
((expression))
() -> :
datetime.now().strftime()
() -> :
.memory.long_term.store(content, metadata={: })
() -> :
context = .memory.build_context(user_input)
context.append({: , : user_input})
tools_schema = [
{
: ,
: {
: name,
: tool[],
: {
: ,
: {k: {: , : v} k, v tool[].items()},
: (tool[].keys()),
},
},
}
name, tool .registry.items()
]
response = client.chat.completions.create(
model=,
messages=context,
tools=tools_schema,
tool_choice=,
temperature=,
)
msg = response.choices[].message
msg.tool_calls:
context.append(msg)
tc msg.tool_calls:
tool_name = tc.function.name
tool_args = json.loads(tc.function.arguments)
tool_result = .registry[tool_name][](**tool_args)
context.append({: , : tc., : tool_result})
response = client.chat.completions.create(
model=,
messages=context,
temperature=,
)
msg = response.choices[].message
.memory.save_turn(, user_input)
.memory.save_turn(, msg.content)
msg.content
__name__ == :
agent = SmartAgent(name=)
conversations = [
,
,
,
,
]
user_input conversations:
()
reply = agent.chat(user_input)
()
| 场景 | 占比 |
|---|---|
| 代码开发辅助 | 22% |
| 数据分析与报表 | 18% |
| 客服与工单处理 | 15% |
| 内容创作与 SEO | 12% |
| 自动化运维 | 10% |
| 研究与知识管理 | 10% |
| 工作流自动化 | 8% |
| 其他 | 5% |
| 维度 | 要点 | 最佳实践 |
|---|---|---|
| 安全性 | 工具执行隔离 | 沙箱环境、权限最小化、输入校验 |
| 可控性 | 限制 Agent 行为边界 | 设定白名单工具、最大迭代次数 |
| 可观测性 | 记录 Agent 决策过程 | 日志记录每轮 Thought/Action/Observation |
| 成本 | Token 消耗控制 | 压缩历史、摘要记忆、缓存频繁查询 |
| 延迟 | 减少不必要的工具调用 | 优化 ReAct 轮数、并行执行独立工具 |
| 可靠性 | 处理 LLM 输出不稳定 | 结构化输出（JSON Schema）、重试机制 |
本文从零到一构建了一个完整的 AI Agent 系统，核心要点：
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online