AI Agent 开发实践
大约 10 分钟约 2984 字
AI Agent 开发实践
简介
AI Agent 是能够自主感知环境、做出决策并执行行动的智能体。相比简单的问答交互，Agent 可以使用工具、规划任务、反思结果，实现复杂的多步骤自动化。本篇介绍 Agent 的核心架构和开发实践。
AI Agent 的本质是将大语言模型的推理能力与外部世界的操作能力连接起来。传统软件是"人定义规则，机器执行"；AI Agent 则是"人定义目标和边界，AI 规划并执行"。这种范式转变带来了巨大的潜力，同时也带来了新的挑战：如何保证 Agent 行为的可预测性、如何处理执行失败、如何控制成本和延迟、如何确保安全性。
特点
Agent 基础架构
ReAct 模式
from openai import OpenAI
import json
client = OpenAI(api_key="your-api-key")
# 工具定义
tools = [
{
"type": "function",
"function": {
"name": "search_database",
"description": "搜索数据库获取数据",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "SQL 查询语句"},
"database": {"type": "string", "description": "数据库名称"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "发送邮件通知",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string", "description": "收件人"},
"subject": {"type": "string", "description": "主题"},
"body": {"type": "string", "description": "邮件内容"}
},
"required": ["to", "subject", "body"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
}
}
}
]
# 工具执行器
def execute_tool(name, args):
if name == "search_database":
return f"查询结果: 查到 {args['query']} 的 15 条记录"
elif name == "send_email":
return f"邮件已发送至 {args['to']}"
elif name == "calculate":
return f"计算结果: {eval(args['expression'])}"
return "未知工具"
# Agent 循环
def run_agent(user_message, max_steps=5):
messages = [
{"role": "system", "content": "你是一个数据分析助手，可以使用工具完成任务。"},
{"role": "user", "content": user_message}
]
for step in range(max_steps):
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools,
temperature=0.1
)
msg = response.choices[0].message
messages.append(msg)
# 检查是否需要调用工具
if msg.tool_calls:
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"[步骤{step+1}] 调用工具: {func_name}({func_args})")
result = execute_tool(func_name, func_args)
print(f"[结果] {result}")
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
else:
# 没有工具调用，返回最终回答
return msg.content
return "达到最大步骤限制"
# 使用
result = run_agent("查询上月销售额，如果超过100万就给老板发邮件通知")
print(f"\n最终回答: {result}")带错误处理和重试的 Agent
class RobustAgent:
"""
健壮的 Agent 实现：
- 错误处理和重试
- 工具执行超时
- 执行日志记录
- 人工确认机制
"""
def __init__(self, client, tools, max_steps=5, timeout=30):
self.client = client
self.tools = tools
self.max_steps = max_steps
self.timeout = timeout
self.execution_log = []
def run(self, user_message, require_approval=False):
messages = [
{"role": "system", "content": self._build_system_prompt()},
{"role": "user", "content": user_message}
]
for step in range(self.max_steps):
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=self.tools,
temperature=0.1
)
msg = response.choices[0].message
messages.append(msg)
if msg.tool_calls:
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
# 高风险工具需要人工确认
if require_approval and func_name in ["send_email", "delete_data"]:
approved = self._request_approval(func_name, func_args)
if not approved:
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": "用户拒绝了此操作"
})
continue
# 执行工具（带错误处理）
try:
result = self._safe_execute(func_name, func_args)
self.execution_log.append({
"step": step + 1,
"tool": func_name,
"args": func_args,
"result": result,
"status": "success"
})
except Exception as e:
result = f"工具执行失败: {str(e)}"
self.execution_log.append({
"step": step + 1,
"tool": func_name,
"args": func_args,
"result": result,
"status": "error"
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
else:
return msg.content
return "达到最大步骤限制，任务未完成"
def _build_system_prompt(self):
return """你是一个数据分析助手。
使用工具完成任务，如果工具执行失败，尝试替代方案。
如果无法完成任务，明确说明原因。"""
def _safe_execute(self, name, args):
"""带超时的安全工具执行"""
import signal
def timeout_handler(signum, frame):
raise TimeoutError(f"工具 {name} 执行超时")
# 设置超时
try:
result = execute_tool(name, args)
return result
except TimeoutError:
return f"工具执行超时（{self.timeout}秒）"
def _request_approval(self, func_name, func_args):
"""请求人工确认（简化实现）"""
print(f"\n[需要确认] 工具: {func_name}, 参数: {func_args}")
# 在实际系统中，这里应该发送通知并等待响应
return True # 默认允许
print("健壮 Agent 框架已定义")多 Agent 协作
角色分工
class Agent:
def __init__(self, name, role, system_prompt):
self.name = name
self.role = role
self.system_prompt = system_prompt
def run(self, message):
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": message}
],
temperature=0.3
)
return response.choices[0].message.content
# 定义专业 Agent
researcher = Agent(
name="研究员",
role="researcher",
system_prompt="你是一个技术研究员，负责收集和分析技术信息，给出详细的技术报告。"
)
coder = Agent(
name="开发者",
role="coder",
system_prompt="你是一个高级 .NET 开发者，根据需求编写高质量的 C# 代码。"
)
reviewer = Agent(
name="审查员",
role="reviewer",
system_prompt="你是一个代码审查专家，检查代码质量、安全性和性能问题，给出改进建议。"
)
# 协作流程
def collaborative_task(task):
# 1. 研究员分析需求
research = researcher.run(f"分析以下任务的技术方案：{task}")
print(f"[研究员] {research[:200]}...")
# 2. 开发者编写代码
code = coder.run(f"基于以下技术方案编写代码：\n{research}\n\n任务：{task}")
print(f"[开发者] 代码已生成")
# 3. 审查员检查代码
review = reviewer.run(f"审查以下代码：\n{code}")
print(f"[审查员] {review[:200]}...")
# 4. 如果有问题，开发者修改
if "问题" in review or "建议" in review:
fixed_code = coder.run(f"根据审查意见修改代码：\n审查意见：{review}\n原始代码：{code}")
return fixed_code
return code多 Agent 通信模式
from enum import Enum
from typing import Optional
class AgentRole(Enum):
COORDINATOR = "coordinator"
RESEARCHER = "researcher"
CODER = "coder"
REVIEWER = "reviewer"
TESTER = "tester"
@dataclass
class AgentMessage:
"""Agent 间通信消息"""
sender: AgentRole
receiver: AgentRole
content: str
task_id: str
is_final: bool = False
class AgentOrchestrator:
"""
Agent 编排器：
协调多个 Agent 的执行顺序和通信
"""
def __init__(self):
self.agents = {}
self.message_queue = []
def register_agent(self, role: AgentRole, agent):
self.agents[role] = agent
def execute_pipeline(self, task: str, pipeline: list[AgentRole]):
"""按管道顺序执行 Agent"""
context = task
for role in pipeline:
if role not in self.agents:
continue
agent = self.agents[role]
result = agent.run(context)
context = result
self.message_queue.append(AgentMessage(
sender=role,
receiver=AgentRole.COORDINATOR,
content=result,
task_id="pipeline-001"
))
return context
def execute_with_review_loop(self, task: str, max_iterations=3):
"""带审查循环的执行"""
context = task
for i in range(max_iterations):
# 研究员分析
context = self.agents[AgentRole.RESEARCHER].run(context)
# 开发者实现
context = self.agents[AgentRole.CODER].run(context)
# 审查员审查
review = self.agents[AgentRole.REVIEWER].run(context)
if "通过" in review or "没有问题" in review:
return context
# 如果有问题，将审查意见反馈给开发者
context = f"审查意见：{review}\n\n请根据以上意见修改。"
return context
print("Agent 编排器已定义")Agent 框架
使用 LangGraph 构建
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def search_docs(query: str) -> str:
"""搜索技术文档"""
return f"找到关于 '{query}' 的 3 篇相关文档"
@tool
def run_code(code: str) -> str:
"""执行代码并返回结果"""
return "代码执行成功，输出: Hello World"
@tool
def create_ticket(title: str, description: str) -> str:
"""创建工单"""
return f"工单已创建: {title}"
# 创建 Agent
llm = ChatOpenAI(model="gpt-4o-mini")
tools = [search_docs, run_code, create_ticket]
agent = create_react_agent(llm, tools)
# 运行
result = agent.invoke({
"messages": [{"role": "user", "content": "查一下 Redis 集群配置文档，并创建一个配置工单"}]
})Agent 模式对比
| 模式 | 说明 | 适用场景 |
|---|---|---|
| ReAct | 推理+行动循环 | 通用任务 |
| Plan-Execute | 先规划后执行 | 复杂多步骤 |
| Reflection | 执行后反思修正 | 需要高质量输出 |
| Multi-Agent | 多角色协作 | 大型任务 |
| Tool-Use | 工具调用 | 需要外部能力 |
| Autonomous | 自主决策运行 | 长时间任务 |
Agent 设计原则
agent_design_principles = {
"最小权限": "Agent 只应拥有完成任务所需的最小工具权限",
"可观测性": "每一步工具调用和决策都应被记录和可追踪",
"可中断性": "必须支持在任意步骤中断和恢复",
"失败优雅": "工具调用失败时应有明确的降级策略",
"成本可控": "设置 Token 预算和步数限制，防止成本失控",
"人工兜底": "高风险操作必须有人工确认或回退机制",
}
for principle, desc in agent_design_principles.items():
print(f"{principle}: {desc}")优点
缺点
总结
AI Agent 核心：LLM 大脑 + 工具集 + 规划能力。ReAct 模式是最常用的 Agent 架构（推理→行动→观察→循环）。多 Agent 用角色分工处理复杂任务。LangGraph 是构建 Agent 的推荐框架。关键设计原则：工具定义要清晰、错误处理要完善、步骤限制防无限循环。企业应用中需要加入权限控制和审计日志。
关键知识点
- 先分清模型能力边界、数据边界和工程边界。
- 任何 AI 主题都不只看效果，还要看延迟、成本、可解释性和安全性。
- 评估方式和失败样例往往比"换哪个模型"更重要。
- 要先分清 Agent 的规划、工具调用、状态存储和执行循环分别由谁负责。
- Agent 的成功标准不是"能跑通 Demo"，而是"能稳定完成任务"。
项目落地视角
- 给数据来源、Prompt 模板、Embedding 版本、评估集和实验结果做版本管理。
- 上线前准备兜底策略，例如拒答、回退、人工审核或缓存降级。
- 观察错误类型时，区分数据问题、召回问题、提示词问题和模型问题。
- 先做一个只有单条主路径的闭环，再逐步增加分支和反思环节。
常见误区
- 只关注 Demo 效果，不考虑线上稳定性和可复现性。
- 没有评估集就频繁调参，最后无法解释为什么变好或变差。
- 忽略权限、审计、隐私和模型输出的安全边界。
- 一上来就做多 Agent，导致问题难定位。
进阶路线
- 继续补齐训练、推理、评估、MLOps 和治理链路。
- 把主题放回真实业务流程，思考谁提供数据、谁消费结果、谁负责兜底。
- 把 PoC 逐步升级到可观测、可回滚、可演进的生产方案。
- 继续深入状态图、工作流编排、Memory 管理和 Agent 评估。
适用场景
- 当你准备把《AI Agent 开发实践》真正落到项目里时，最适合先在一个独立模块或最小样例里验证关键路径。
- 适合企业知识问答、内容生成、分类抽取和智能助手等场景。
- 当需求同时关注效果、时延、成本和安全边界时，这类主题最有价值。
落地建议
- 先定义评估集、成功标准和失败样例，再开始调模型或调提示。
- 把数据来源、分块方式、Embedding 版本和 Prompt 模板纳入版本管理。
- 上线前准备兜底策略，例如拒答、回退、人工审核或检索降级。
排错清单
- 先判断问题出在数据、检索、Prompt、模型还是后处理。
- 检查上下文是否过长、分块是否过碎或召回是否偏题。
- 对错误回答做分类，区分幻觉、事实过时、指令误解和格式错误。
复盘问题
- 如果把《AI Agent 开发实践》放进你的当前项目，最先要验证的输入、输出和失败路径分别是什么？
- 《AI Agent 开发实践》最容易在什么规模、什么边界条件下暴露问题？你会用什么指标或日志去确认？
- 相比默认实现或替代方案，采用《AI Agent 开发实践》最大的收益和代价分别是什么？