AI Agent 开发实战:从 0 到 1 构建你的第一个智能助手
本文带你完整经历 AI Agent 的开发流程,从架构设计到代码实现,手把手教你构建一个实用的智能助手。涵盖三层架构设计、工具系统实现、任务规划与反思模块,以及部署监控最佳实践。
AI Agent 开发实战:从 0 到 1 构建你的第一个智能助手
本文带你完整经历 AI Agent 的开发流程,从架构设计到代码实现,手把手教你构建一个实用的智能助手。
一、为什么需要 AI Agent
2024 年被认为是 AI Agent 的元年。从 AutoGPT 的爆火,到各种 Agent 框架的涌现,我们看到了 AI 从"对话工具"向"自主执行者"的进化。
传统的 AI 应用往往是被动响应:用户提问,AI 回答。但真实世界的工作流程要复杂得多——你需要查邮件、更新日历、操作数据库、调用 API、甚至控制浏览器。这就是 AI Agent 的价值所在:让 AI 不仅能思考,还能行动。
Agent 的核心能力
- 感知(Perception):理解用户意图,读取环境状态
- 规划(Planning):分解复杂任务,制定执行步骤
- 行动(Action):调用工具、执行代码、操作外部系统
- 反思(Reflection):评估结果,调整策略,从错误中学习
二、架构设计:三层模型
在开始编码之前,我们先设计一个清晰的架构。我推荐采用三层模型:
┌─────────────────────────────────────┐
│ 应用层 (Application) │
│ - 用户接口 (CLI/Web/API) │
│ - 会话管理 │
│ - 上下文管理 │
├─────────────────────────────────────┤
│ 代理层 (Agent Core) │
│ - 任务规划器 │
│ - 工具路由器 │
│ - 执行引擎 │
│ - 反思模块 │
├─────────────────────────────────────┤
│ 工具层 (Tool Layer) │
│ - 文件操作 │
│ - 网络请求 │
│ - 数据库访问 │
│ - 第三方 API │
└─────────────────────────────────────┘
各层职责
应用层负责与用户交互,管理对话历史,处理认证和授权。
代理层是大脑,负责理解意图、规划步骤、选择工具、执行任务。
工具层是手脚,封装所有可执行的操作,提供统一的调用接口。
三、技术选型
语言选择
- Python:生态丰富,AI 库最全,适合快速原型
- TypeScript/Node.js:适合 Web 集成,类型安全
- Go:高性能,适合生产环境
本文使用 Python 3.11+,因为它的 AI 生态最成熟。
核心依赖
pip install openai langchain python-dotenv
pip install playwright # 浏览器自动化
pip install aiohttp # 异步 HTTP
pip install pydantic # 数据验证
为什么不用现成框架?
你可能会问:为什么不直接用 LangChain、LlamaIndex 或 AutoGen?
答案是:学习阶段应该从底层开始。框架封装了太多细节,不利于理解 Agent 的工作原理。等你理解了核心概念,再用框架会事半功倍。
四、代码实现
4.1 项目结构
my_agent/
├── agent/
│ ├── __init__.py
│ ├── core.py # Agent 核心逻辑
│ ├── planner.py # 任务规划
│ └── reflector.py # 反思模块
├── tools/
│ ├── __init__.py
│ ├── base.py # 工具基类
│ ├── file_tools.py # 文件操作
│ ├── web_tools.py # 网络工具
│ └── system_tools.py # 系统工具
├── config/
│ └── settings.py # 配置管理
├── main.py # 入口文件
└── requirements.txt
4.2 工具基类
所有工具都继承自同一个基类,确保统一的接口:
# tools/base.py
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field
class ToolSchema(BaseModel):
"""工具的描述 Schema,用于 LLM 理解工具功能"""
name: str
description: str
parameters: Dict[str, Any]
class BaseTool(ABC):
"""所有工具的基类"""
@property
@abstractmethod
def name(self) -> str:
"""工具名称"""
pass
@property
@abstractmethod
def description(self) -> str:
"""工具描述"""
pass
@property
@abstractmethod
def schema(self) -> ToolSchema:
"""工具的 JSON Schema"""
pass
@abstractmethod
async def execute(self, **kwargs) -> Any:
"""执行工具"""
pass
async def __call__(self, **kwargs) -> Any:
"""允许像函数一样调用"""
return await self.execute(**kwargs)
4.3 实现具体工具
以文件读取工具为例:
# tools/file_tools.py
import aiofiles
from typing import Optional
from .base import BaseTool, ToolSchema
class ReadFileTool(BaseTool):
"""读取文件内容"""
@property
def name(self) -> str:
return "read_file"
@property
def description(self) -> str:
return "读取指定路径的文件内容,支持文本文件"
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name=self.name,
description=self.description,
parameters={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "文件路径(绝对路径或相对路径)"
},
"encoding": {
"type": "string",
"description": "文件编码,默认 utf-8",
"default": "utf-8"
}
},
"required": ["path"]
}
)
async def execute(self, path: str, encoding: str = "utf-8") -> str:
try:
async with aiofiles.open(path, 'r', encoding=encoding) as f:
return await f.read()
except FileNotFoundError:
return f"错误:文件不存在 - {path}"
except Exception as e:
return f"错误:{str(e)}"
4.4 Agent 核心逻辑
# agent/core.py
import json
from typing import List, Dict, Any, Optional
from openai import AsyncOpenAI
from .planner import TaskPlanner
from .reflector import TaskReflector
from tools.base import BaseTool
class Agent:
"""AI Agent 核心类"""
def __init__(
self,
api_key: str,
model: str = "gpt-4o",
tools: Optional[List[BaseTool]] = None,
system_prompt: Optional[str] = None
):
self.client = AsyncOpenAI(api_key=api_key)
self.model = model
self.tools = tools or []
self.tool_map = {tool.name: tool for tool in self.tools}
self.messages = []
self.planner = TaskPlanner()
self.reflector = TaskReflector()
# 默认系统提示
self.system_prompt = system_prompt or self._default_system_prompt()
self.messages.append({
"role": "system",
"content": self.system_prompt
})
def _default_system_prompt(self) -> str:
"""生成包含工具信息的系统提示"""
tool_descriptions = "\n".join([
f"- {tool.name}: {tool.description}"
for tool in self.tools
])
return f"""你是一个智能助手,可以调用工具来完成任务。
可用工具:
{tool_descriptions}
调用工具时,请返回 JSON 格式:
{{
"action": "tool_name",
"parameters": {{...}}
}}
如果不需要调用工具,直接返回文本回复。"""
async def chat(self, user_message: str) -> str:
"""处理用户消息并返回回复"""
# 添加用户消息
self.messages.append({
"role": "user",
"content": user_message
})
# 调用 LLM
response = await self.client.chat.completions.create(
model=self.model,
messages=self.messages,
temperature=0.7
)
assistant_message = response.choices[0].message.content
# 检查是否需要调用工具
tool_call = self._parse_tool_call(assistant_message)
if tool_call:
# 执行工具
result = await self._execute_tool(tool_call)
# 添加工具调用记录
self.messages.append({
"role": "assistant",
"content": f"[调用工具:{tool_call['action']}]"
})
self.messages.append({
"role": "system",
"content": f"工具执行结果:{result}"
})
# 再次调用 LLM 处理结果
response = await self.client.chat.completions.create(
model=self.model,
messages=self.messages,
temperature=0.7
)
assistant_message = response.choices[0].message.content
# 添加助手回复
self.messages.append({
"role": "assistant",
"content": assistant_message
})
return assistant_message
def _parse_tool_call(self, text: str) -> Optional[Dict]:
"""解析工具调用请求"""
try:
# 尝试提取 JSON
start = text.find('{')
end = text.rfind('}') + 1
if start >= 0 and end > start:
return json.loads(text[start:end])
except:
pass
return None
async def _execute_tool(self, tool_call: Dict) -> Any:
"""执行工具调用"""
tool_name = tool_call.get('action')
params = tool_call.get('parameters', {})
if tool_name not in self.tool_map:
return f"错误:未知工具 - {tool_name}"
tool = self.tool_map[tool_name]
return await tool(**params)
def clear_history(self):
"""清空对话历史"""
self.messages = [{
"role": "system",
"content": self.system_prompt
}]
4.5 入口文件
# main.py
import asyncio
from agent.core import Agent
from tools.file_tools import ReadFileTool
from tools.web_tools import FetchUrlTool
from tools.system_tools import ShellTool
async def main():
# 初始化 Agent
agent = Agent(
api_key="your-api-key",
model="gpt-4o",
tools=[
ReadFileTool(),
FetchUrlTool(),
ShellTool()
]
)
# 交互式对话
print("🤖 Agent 已就绪,输入 q 退出")
while True:
user_input = input("\n你:").strip()
if user_input.lower() == 'q':
break
response = await agent.chat(user_input)
print(f"\nAgent: {response}")
if __name__ == "__main__":
asyncio.run(main())
五、进阶功能
5.1 任务规划
复杂任务需要分解。规划器可以将"分析项目代码并生成报告"分解为:
- 读取项目目录结构
- 读取主要源代码文件
- 分析代码结构和依赖
- 生成分析报告
5.2 记忆管理
长期记忆让 Agent 更智能:
- 短期记忆:当前对话历史
- 长期记忆:向量数据库存储重要信息
- 工作记忆:当前任务的中间状态
5.3 错误处理与重试
工具调用可能失败,需要:
- 捕获异常并记录
- 根据错误类型决定是否重试
- 超过重试次数后向用户报告
六、部署与监控
6.1 Docker 部署
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "main.py"]
6.2 日志与监控
- 记录所有工具调用
- 跟踪任务执行时间
- 监控 LLM API 调用成本
- 设置异常告警
七、总结与建议
关键收获
- 从简单开始:先实现核心功能,再逐步扩展
- 工具设计要统一:统一的接口让扩展更容易
- 错误处理很重要:生产环境必须考虑各种异常情况
- 监控不可少:了解 Agent 的行为才能优化它
下一步
- 尝试添加更多工具(数据库、邮件、日历等)
- 实现多 Agent 协作
- 探索 RAG(检索增强生成)
- 学习使用成熟的 Agent 框架
关于作者:本文作者是一名 AI 工程师,专注于 Agent 架构设计与落地实践。欢迎在评论区交流讨论。
参考资料:
- OpenAI Function Calling 文档
- LangChain 官方教程
- AutoGen 论文