手写第一个 AI Agent — 50 行代码让模型自己搜索和计算

上一篇文章我们讲了 AI Agent 的概念和 ReAct 循环。这次直接动手——用不到 50 行 Python,写一个能自己搜索网页、执行代码的 AI Agent

你需要:Python 3.10+,一个支持 Function Calling 的 API。我们使用 OpenAI 兼容接口(任何支持 /v1/chat/completions 的模型都可以,包括本地部署的)。

Agent 的两个工具

一个能搜索、能计算——这两个工具已经覆盖了大多数实际需求。

工具 1:网页搜索

def search_web(query: str) -> str:
    """搜索网页,返回前 5 条结果的标题和链接。"""
    import urllib.request, urllib.parse, json
    # 使用 DuckDuckGo Instant Answer API(免费,无需 API Key)
    url = "https://api.duckduckgo.com/?" + urllib.parse.urlencode({
        "q": query, "format": "json", "no_html": 1, "skip_disambig": 1
    })
    with urllib.request.urlopen(url, timeout=10) as resp:
        data = json.loads(resp.read())
    results = []
    if data.get("AbstractText"):
        results.append(f"摘要: {data['AbstractText']}")
    for item in data.get("RelatedTopics", [])[:5]:
        if isinstance(item, dict) and item.get("Text"):
            results.append(f"- {item['Text']}")
    return "\n".join(results) if results else "未找到结果"

工具 2:Python 代码执行

def run_python(code: str) -> str:
    """执行 Python 代码,返回 stdout 输出。"""
    import subprocess
    try:
        result = subprocess.run(
            ["python3", "-c", code],
            capture_output=True, text=True, timeout=30
        )
        if result.returncode == 0:
            return result.stdout or "(无输出)"
        return f"错误: {result.stderr}"
    except subprocess.TimeoutExpired:
        return "错误: 代码执行超时"
⚡ 安全提示:生产环境中,run_python 必须在沙箱(Docker/虚拟机)中执行,避免恶意代码破坏系统。这里为演示简化了。

Agent 主循环

核心就是 ReAct 循环,加上工具定义和消息管理:

import json
from openai import OpenAI

# 初始化客户端(替换为你的 API 地址和密钥)
client = OpenAI(
    base_url="https://api.openai.com/v1",  # 或其他兼容接口
    api_key="your-api-key"
)

# 工具定义(JSON Schema,模型需要这个来理解工具)
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "搜索网页获取最新信息。当需要实时数据或超出训练集的知识时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "run_python",
            "description": "执行 Python 代码进行计算或数据处理。",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {"type": "string", "description": "要执行的 Python 代码"}
                },
                "required": ["code"]
            }
        }
    }
]

def run_agent(user_input: str, max_turns: int = 10) -> str:
    """主 Agent 循环。"""
    messages = [
        {"role": "system", "content": "你是一个有用的 AI 助手。你可以搜索网页获取最新信息,也可以执行 Python 代码进行计算。优先使用工具获取准确信息,不要凭记忆猜测。"},
        {"role": "user", "content": user_input}
    ]

    for turn in range(max_turns):
        # 调用模型
        response = client.chat.completions.create(
            model="gpt-4o",  # 或 deepseek-chat 等
            messages=messages,
            tools=TOOLS,
            tool_choice="auto"
        )

        msg = response.choices[0].message

        # 如果模型直接回复(不再调用工具),任务结束
        if not msg.tool_calls:
            return msg.content or ""

        # 执行每个工具调用
        for tool_call in msg.tool_calls:
            fn_name = tool_call.function.name
            fn_args = json.loads(tool_call.function.arguments)

            print(f"🔧 调用工具: {fn_name}({fn_args})")

            # 执行工具
            if fn_name == "search_web":
                result = search_web(**fn_args)
            elif fn_name == "run_python":
                result = run_python(**fn_args)
            else:
                result = f"未知工具: {fn_name}"

            # 将工具结果加入消息
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

        # 将模型的工具调用消息也加入
        messages.append(msg)

    return "已达到最大轮次,任务可能未完成。"

# 运行
if __name__ == "__main__":
    answer = run_agent("北京和上海的时差是多少?用 Python 算一下 2^20 等于多少。")
    print(f"\n✅ 最终回答:\n{answer}")

运行效果

当你运行这段代码,Agent 会:

  1. 理解任务中有两个子问题
  2. 先调用 run_python("print(2**20)") 计算 2^20 = 1048576
  3. 然后调用 search_web("北京上海时差") 确认两地都在东八区
  4. 整合两个结果,给出最终回答

在实际运行中你会看到类似这样的输出:

🔧 调用工具: run_python({'code': 'print(2**20)'})
🔧 调用工具: search_web({'query': '北京上海时差'})

✅ 最终回答:
2^20 = 1,048,576。
北京和上海都在中国标准时间(UTC+8),没有时差。

这 50 行代码的本质

这不是玩具。这 50 行代码包含了所有生产级 Agent 框架的核心:

  1. 工具抽象——用 JSON Schema 描述工具,模型自动理解
  2. ReAct 循环——思考→行动→观察→再思考
  3. 消息管理——system/user/assistant/tool 四种角色精确控制上下文
  4. 安全边界——轮次上限防止无限循环

理解了这 50 行,你就理解了 LangChain、AutoGPT、CrewAI 等框架的底层逻辑——它们只是在这个循环上加了更多工具、更好的错误处理、更复杂的编排。

三个扩展方向

从这里出发,你可以:

  1. 加更多工具——读文件、发邮件、调 API……任何有接口的东西都能变成工具
  2. 加记忆系统——用向量数据库或简单文件存储,让 Agent 记住跨会话的信息
  3. 加错误恢复——工具执行失败时,让模型知道错误原因并重试

后续文章会逐一展开这些方向。

📖 下一篇:Agent 工具设计最佳实践 — 怎么写工具描述模型才不迷糊