AI Agent 项目新手入门：从零搭建你的第一个智能体

共计 2482 个字符，预计需要花费 7 分钟才能阅读完成。

为什么需要 AI Agent？

传统脚本和智能体的核心差异就像计算器和助理的区别：

• 静态 vs 动态：脚本按预定流程执行（如定时爬虫），而 Agent 能根据环境动态调整（如根据用户问题自动选择工具）
• 单次 vs 持续：脚本调用后立即结束，Agent 维持长期会话状态（如记住用户偏好）

新手最常踩的两个坑：

1. 异步事件卡死：同时处理用户输入和 API 调用时线程阻塞
2. 记忆丢失：多轮对话中忘记之前的聊天上下文

框架选型：LangChain 还是 AutoGPT？

建议初学者从 LangChain 开始，它的 Tool 和Memory设计更易于调试。

手把手实现基础 Agent

1. 搭建骨架结构

from typing import Dict, List, Callable
import asyncio

class BaseAgent:
    def __init__(self):
        self.message_queue = asyncio.Queue()  # 异步消息队列
        self.skills: Dict[str, Callable] = {}  # 技能注册表
        self.context = {}  # 会话上下文

    def register_skill(self, name: str, func: Callable):
        """注册新技能（如天气查询）"""
        self.skills[name] = func

2. 接入 OpenAI 对话

import openai
from collections import deque

class ChatAgent(BaseAgent):
    def __init__(self, api_key: str):
        super().__init__()
        self.history = deque(maxlen=10)  # 限制历史记录长度
        openai.api_key = api_key

    async def chat(self, query: str) -> str:
        """流式响应实现"""
        self.history.append({"role": "user", "content": query})

        try:
            response = await openai.ChatCompletion.acreate(
                model="gpt-3.5-turbo",
                messages=list(self.history),
                stream=True  # 启用流式传输
            )

            full_response = []
            async for chunk in response:
                delta = chunk.choices[0].delta
                if "content" in delta:
                    full_response.append(delta.content)
                    yield delta.content  # 实时返回片段

            self.history.append({"role": "assistant", "content": "".join(full_response)})
        except openai.error.RateLimitError:
            yield "[错误]请求过于频繁，请稍后再试"

避坑实战指南

线程安全三件套

import threading

class SafeAgent(BaseAgent):
    def __init__(self):
        super().__init__()
        self.lock = threading.Lock()  # 状态修改加锁

    def update_context(self, key: str, value):
        with self.lock:  # 防止多线程冲突
            self.context[key] = value

API 限流处理

推荐使用 tenacity 库实现自动重试：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def call_external_api(params):
    # 模拟 API 调用
    if random.random() > 0.8:  # 故意制造 20% 失败率
        raise Exception("API 超时")
    return "success"

敏感词过滤

import re

sensitive_patterns = [r"\b(密码 | 银行卡)\b",  # 中文关键词
    r"\d{4}-\d{4}-\d{4}-\d{4}"  # 信用卡号
]

def sanitize_input(text: str) -> str:
    for pattern in sensitive_patterns:
        text = re.sub(pattern, "[已过滤]", text)
    return text

性能压测方案

使用 Locust 模拟并发请求（保存为locustfile.py）：

from locust import HttpUser, task, between

class AgentUser(HttpUser):
    wait_time = between(0.5, 2)

    @task
    def chat(self):
        self.client.post("/chat", json={"query": "推荐周末旅游景点"})

运行测试：

locust -f locustfile.py --headless -u 100 -r 10 -t 1m

关键指标监控：
– 当并发 >50 时，关注 P99 延迟是否超过 2 秒
– API 成本 = (总 token 数 / 1000) * 单价

下一步行动建议

1. 增强记忆：尝试接入 Redis 存储长期对话历史
2. 扩展技能 ：用@tool 装饰器添加维基百科查询
3. 混合架构：结合 AutoGPT 的自动规划能力

推荐实验：
– 在 Gradio 中构建 Web 界面
– 接入知识图谱实现事实校验

参考资源：
– LangChain 官方 Cookbook
– OpenAI Token 计算器
–《Building AI Agents》O’Reilly 电子书