折腾侠
技术教程

AI Agent 开发实战:从 0 到 1 构建你的第一个智能助手

本文带你完整经历 AI Agent 的开发流程,从架构设计到代码实现,手把手教你构建一个实用的智能助手。涵盖三层架构设计、工具系统实现、任务规划与反思模块,以及部署监控最佳实践。

折腾侠
2026/03/20 发布
9约 8 分钟1299 字 / 931 词00

AI Agent 开发实战:从 0 到 1 构建你的第一个智能助手

本文带你完整经历 AI Agent 的开发流程,从架构设计到代码实现,手把手教你构建一个实用的智能助手。

一、为什么需要 AI Agent

2024 年被认为是 AI Agent 的元年。从 AutoGPT 的爆火,到各种 Agent 框架的涌现,我们看到了 AI 从"对话工具"向"自主执行者"的进化。

传统的 AI 应用往往是被动响应:用户提问,AI 回答。但真实世界的工作流程要复杂得多——你需要查邮件、更新日历、操作数据库、调用 API、甚至控制浏览器。这就是 AI Agent 的价值所在:让 AI 不仅能思考,还能行动

Agent 的核心能力

  1. 感知(Perception):理解用户意图,读取环境状态
  2. 规划(Planning):分解复杂任务,制定执行步骤
  3. 行动(Action):调用工具、执行代码、操作外部系统
  4. 反思(Reflection):评估结果,调整策略,从错误中学习

二、架构设计:三层模型

在开始编码之前,我们先设计一个清晰的架构。我推荐采用三层模型

┌─────────────────────────────────────┐
│         应用层 (Application)         │
│  - 用户接口 (CLI/Web/API)            │
│  - 会话管理                          │
│  - 上下文管理                        │
├─────────────────────────────────────┤
│         代理层 (Agent Core)          │
│  - 任务规划器                        │
│  - 工具路由器                        │
│  - 执行引擎                          │
│  - 反思模块                          │
├─────────────────────────────────────┤
│         工具层 (Tool Layer)          │
│  - 文件操作                          │
│  - 网络请求                          │
│  - 数据库访问                        │
│  - 第三方 API                        │
└─────────────────────────────────────┘

各层职责

应用层负责与用户交互,管理对话历史,处理认证和授权。

代理层是大脑,负责理解意图、规划步骤、选择工具、执行任务。

工具层是手脚,封装所有可执行的操作,提供统一的调用接口。

三、技术选型

语言选择

  • Python:生态丰富,AI 库最全,适合快速原型
  • TypeScript/Node.js:适合 Web 集成,类型安全
  • Go:高性能,适合生产环境

本文使用 Python 3.11+,因为它的 AI 生态最成熟。

核心依赖

Bash
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 工具基类

所有工具都继承自同一个基类,确保统一的接口:

Python
# 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 实现具体工具

以文件读取工具为例:

Python
# 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 核心逻辑

Python
# 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 入口文件

Python
# 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 任务规划

复杂任务需要分解。规划器可以将"分析项目代码并生成报告"分解为:

  1. 读取项目目录结构
  2. 读取主要源代码文件
  3. 分析代码结构和依赖
  4. 生成分析报告

5.2 记忆管理

长期记忆让 Agent 更智能:

  • 短期记忆:当前对话历史
  • 长期记忆:向量数据库存储重要信息
  • 工作记忆:当前任务的中间状态

5.3 错误处理与重试

工具调用可能失败,需要:

  • 捕获异常并记录
  • 根据错误类型决定是否重试
  • 超过重试次数后向用户报告

六、部署与监控

6.1 Docker 部署

Dockerfile
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 调用成本
  • 设置异常告警

七、总结与建议

关键收获

  1. 从简单开始:先实现核心功能,再逐步扩展
  2. 工具设计要统一:统一的接口让扩展更容易
  3. 错误处理很重要:生产环境必须考虑各种异常情况
  4. 监控不可少:了解 Agent 的行为才能优化它

下一步

  • 尝试添加更多工具(数据库、邮件、日历等)
  • 实现多 Agent 协作
  • 探索 RAG(检索增强生成)
  • 学习使用成熟的 Agent 框架

关于作者:本文作者是一名 AI 工程师,专注于 Agent 架构设计与落地实践。欢迎在评论区交流讨论。

参考资料

  • OpenAI Function Calling 文档
  • LangChain 官方教程
  • AutoGen 论文
分享到:

如果这篇文章对你有帮助,欢迎请作者喝杯咖啡 ☕

加载评论中...