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

2026年5月14日 · 入门 · 含代码

30秒结论

解决什么问题：把 ReAct 理论变成可运行的代码——模型能搜索网页、执行 Python、自主完成任务。
核心方法：不到 50 行 Python，包含工具定义（JSON Schema）、ReAct 主循环、消息管理和轮次上限安全边界。
关键结论：这 50 行代码包含了 LangChain、AutoGPT 等所有生产级框架的核心逻辑。
读完能做什么：复制代码即可运行你自己的 AI Agent，理解如何添加新工具、切换模型。

上一篇文章我们讲了 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 会：

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

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

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

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

这 50 行代码的本质

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

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

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

三个扩展方向

从这里出发，你可以：

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

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

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

下一步阅读

常见问题

Q: 这段代码需要什么 API？可以用国产模型吗？

A: 代码使用 OpenAI 兼容接口，任何支持 Function Calling 的模型都可以。DeepSeek、Qwen、GLM 等只要兼容接口格式即可替换。

Q: run_python 工具在服务器上执行代码安全吗？

A: 本文代码为了演示简化。生产环境必须在 Docker 容器或虚拟机沙箱中执行，限制网络、文件系统和 CPU 资源。

Q: Agent 最多执行几轮？会不会无限循环？

A: max_turns 默认 10，Agent 在限制内未完成任务会主动终止。实际使用中大多数任务 3-5 轮即可完成。

Q: 如何给这个 Agent 添加更多工具？

A: 三步：① 实现工具函数，② 添加 JSON Schema 定义，③ 在循环分支中添加执行逻辑。模型会自动根据描述选择使用新工具。