目录

• 1. 前言：从「调用 API」到「交付结果」
• 2. 什么是 AI Agent
• 3. 项目结构与工具系统
• 4. 实现 LLM 客户端
• 5. 编写 Agent 主循环
• 6. 实战：注册工具并跑通一个任务
• 7. 进阶方向
• 8. 总结

1. 前言：从「调用 API」到「交付结果」

很多人接触大模型的第一件事，就是写一段调用 OpenAI / DeepSeek / 通义千问 API 的代码：

import openai

resp = openai.ChatCompletion.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "帮我写一首诗"}],
)
print(resp["choices"][0]["message"]["content"])

这段代码确实「调用了大模型」，但它离一个真正能帮你干活的 AI Agent 还很远。区别在于：单次 API 调用只能给出一段文本，而 Agent 能自己规划、调用工具、观察结果、修正计划，最终交付一个任务结果。

举例来说：

• 「调用 API」：你问「今天天气怎么样」，模型凭训练数据猜一个答案。
• 「AI Agent」：你问「帮我把 docs 目录下所有 Markdown 文件里过期的链接找出来并替换成新域名」，Agent 会先列出文件、读取内容、用正则匹配、逐个替换、最后汇报改了多少个文件。

这篇文章就用纯 Python 从零实现一个最小但完整的 ReAct 风格 AI Agent，包含工具系统、LLM 客户端和主循环，代码可以直接复制运行。

2. 什么是 AI Agent

AI Agent 可以抽象为这样一个循环：

问题 → 思考(Thought) → 选择动作(Action) → 执行工具 → 观察结果(Observation) → 再思考 → … → 最终答案(Final Answer)

每一轮，Agent 都面临一个问题：「我现在知道什么？下一步该做什么？」它可以选择调用某个工具获取信息，也可以选择直接给出最终答案。这个「思考-行动-观察」的循环，就是 ReAct（Reasoning + Acting） 范式，也是本文学实现的核心。

一个最小 Agent 需要三块拼图：

1. LLM 客户端：负责「思考」，把当前状态转成下一步行动。
2. 工具系统：负责「行动」，提供一组可被调用的函数（读文件、发请求、执行命令等）。
3. 主循环：把前两者串起来，管理上下文直到产出最终结果。

下面从零实现这三块。

3. 项目结构与工具系统

先在目录里建一个 agent_core.py，我们按以下结构组织代码：

agent_core.py
├── Tool            # 工具的数据结构（名称、描述、参数、函数）
├── ToolRegistry    # 工具注册与查询
├── LLMClient       # 调用 OpenAI 兼容接口
└── ReActAgent      # 主循环

工具系统的核心是让 LLM 知道「我有哪些工具、每个工具怎么用」。我们用函数签名 + 文档字符串来生成工具描述：

from dataclasses import dataclass, field
from typing import Any, Callable
import inspect


@dataclass
class Tool:
    name: str
    description: str
    func: Callable
    # 参数说明，格式如 {"file_path": "要读取的文件路径"}
    parameters: dict[str, str] = field(default_factory=dict)

    def render(self) -> str:
        """生成给 LLM 看的工具说明。"""
        params = ", ".join(
            f"{k}: {v}" for k, v in self.parameters.items()
        )
        return f"- {self.name}({params}): {self.description}"


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}

    def register(
        self,
        func: Callable,
        name: str | None = None,
        description: str | None = None,
    ) -> Callable:
        """将函数注册为工具，name/description 默认取自函数签名与 docstring。"""
        tool_name = name or func.__name__
        desc = description or (inspect.getdoc(func) or "").strip()
        params = {
            k: v.annotation if isinstance(v.annotation, str) else str(v.annotation)
            for k, v in inspect.signature(func).parameters.items()
        }
        self._tools[tool_name] = Tool(
            name=tool_name,
            description=desc,
            func=func,
            parameters={k: v for k, v in params.items() if v != "inspect._empty"},
        )
        return func

    def list_tools(self) -> str:
        return "\n".join(t.render() for t in self._tools.values())

    def execute(self, name: str, args: dict[str, Any]) -> Any:
        tool = self._tools.get(name)
        if tool is None:
            return f"错误：未找到工具 {name}"
        try:
            return tool.func(**args)
        except Exception as exc:  # noqa: BLE001
            return f"错误：执行 {name} 失败 - {exc}"

这里有三个关键设计：

• 用 inspect 自动提取函数签名，减少手动维护成本；
• render() 生成 LLM 可见的工具清单，上下文里就靠它让模型知道该调什么；
• execute() 统一处理异常，工具出错时把错误信息返回给 LLM，而不是让程序崩溃，这样 Agent 才有机会根据错误调整下一步。

4. 实现 LLM 客户端

为了不绑定特定厂商，我们直接通过 HTTP 调用 OpenAI 兼容的 Chat Completions 接口。这样 base_url 换成 DeepSeek、通义千问兼容模式或本地 Ollama 都可以：

import json

import requests


class LLMClient:
    def __init__(
        self,
        api_key: str,
        model: str,
        base_url: str = "https://api.openai.com/v1",
        timeout: int = 60,
    ) -> None:
        self.api_key = api_key
        self.model = model
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

    def chat(self, messages: list[dict[str, str]], temperature: float = 0.0) -> str:
        url = f"{self.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature,
        }
        resp = requests.post(url, headers=headers, json=payload, timeout=self.timeout)
        resp.raise_for_status()
        data = resp.json()
        return data["choices"][0]["message"]["content"]

依赖只有一个 requests。如果你用官方的 openai 包也行，但自己发 HTTP 请求的好处是：你能看到完整的消息结构，对 Agent 的上下文拼接有完全的控制权。

5. 编写 Agent 主循环

主循环是整个 Agent 的大脑。它的工作流程是：

1. 组装系统提示词，告知 LLM 它有哪些工具、输出格式要遵循什么约定；
2. 调用 LLM，拿到它返回的文本；
3. 解析文本，判断下一步是「调用工具」还是「给出最终答案」；
4. 如果调用工具，就执行并把结果追加进消息上下文，回到第 2 步；
5. 如果给出最终答案，结束循环并返回答案。

输出协议约定得越明确，解析越稳定。下面是完整实现：

import re


class ReActAgent:
    SYSTEM_PROMPT = """你是一个能调用工具完成任务的人工智能助手。

请严格按照以下格式输出，每次只能输出一个 Thought 和一个 Action 或 Final Answer：

Thought: 你当前的思考和计划
Action: 要调用的工具名
Action Input: 工具入参（JSON 对象）

或者当你已有足够信息给出最终答案时：

Thought: 你当前的思考
Final Answer: 面向用户的最终答案

可用工具：
{tools}
"""

    def __init__(self, llm: LLMClient, registry: ToolRegistry, max_steps: int = 8) -> None:
        self.llm = llm
        self.registry = registry
        self.max_steps = max_steps

    def run(self, question: str) -> str:
        tools_desc = self.registry.list_tools()
        system = self.SYSTEM_PROMPT.format(tools=tools_desc)
        messages: list[dict[str, str]] = [
            {"role": "system", "content": system},
            {"role": "user", "content": f"任务：{question}"},
        ]

        for step in range(self.max_steps):
            raw = self.llm.chat(messages)
            thought = self._extract("Thought", raw)
            final = self._extract("Final Answer", raw)
            action = self._extract("Action", raw)
            action_input = self._extract("Action Input", raw)

            if final:
                messages.append({"role": "assistant", "content": raw})
                return final

            if not action:
                # 模型没有按协议输出，把原始返回当观察结果塞回去逼它继续
                messages.append({"role": "assistant", "content": raw})
                messages.append(
                    {
                        "role": "user",
                        "content": "你上一条输出不符合约定格式，请按 Thought/Action/Action Input 格式继续。",
                    }
                )
                continue

            try:
                args = json.loads(action_input or "{}")
            except json.JSONDecodeError as exc:
                messages.append({"role": "assistant", "content": raw})
                messages.append(
                    {
                        "role": "user",
                        "content": f"Action Input 必须是合法 JSON，解析失败：{exc}",
                    }
                )
                continue

            observation = self.registry.execute(action, args)
            messages.append({"role": "assistant", "content": raw})
            messages.append({"role": "user", "content": f"Observation: {observation}"})

        return "达到最大步数仍未完成任务，请检查工具是否齐全或提示词是否需要优化。"

    @staticmethod
    def _extract(field: str, text: str) -> str | None:
        pattern = rf"^\s*{field}\s*[:：]\s*(.+)$"
        for line in text.splitlines():
            match = re.match(pattern, line)
            if match:
                return match.group(1).strip()
        return None

几个值得注意的细节：

• 解析失败不中止：LLM 输出不合规是常态（尤其本地小模型），把错误信息作为 user 消息喂回去，给它一次自我纠正的机会；
• Observation 用自然语言拼在 user 消息里，而不是单独设计一个消息角色，这样对任何 OpenAI 兼容接口都通用；
• max_steps 是安全阀，防止 Agent 无限循环烧 token。

6. 实战：注册工具并跑通一个任务

现在注册几个真实可用的工具，让 Agent 真正「干活」。以文件处理任务为例：

import os
import time
from pathlib import Path


def read_file(path: str) -> str:
    """读取文本文件内容。"""
    return Path(path).read_text(encoding="utf-8")


def list_dir(path: str = ".") -> str:
    """列出目录下的所有文件，path 为目录路径。"""
    return "\n".join(sorted(os.listdir(path)))


def current_time() -> str:
    """获取当前系统时间。"""
    return time.strftime("%Y-%m-%d %H:%M:%S")

注意：工具函数的 docstring 会被自动提取为描述，因此描述要写清楚用途和参数含义，这直接决定 LLM 调用的准确率。

然后组装并运行完整 Agent：

import os

import agent_core

if __name__ == "__main__":
    registry = agent_core.ToolRegistry()
    registry.register(read_file)
    registry.register(list_dir)
    registry.register(current_time)

    llm = agent_core.LLMClient(
        api_key=os.environ["OPENAI_API_KEY"],
        model="gpt-4o-mini",
        base_url=os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    )

    agent = agent_core.ReActAgent(llm=llm, registry=registry)

    answer = agent.run(
        "列出当前目录下有哪些文件，然后读取 agent_core.py 的前 5 行内容。"
    )
    print(f"最终答案：\n{answer}")

运行前安装依赖并设置环境变量：

pip install requests
export OPENAI_API_KEY="你的密钥"
# 使用其他 OpenAI 兼容服务时：
# export OPENAI_BASE_URL="https://api.deepseek.com/v1"
python main.py

一次典型运行的中间过程长这样（日志可自行在 run 里打印）：

Thought: 我需要先列出当前目录的文件，找到 agent_core.py 再读取。
Action: list_dir
Action Input: {"path": "."}

Observation: agent_core.py
main.py

Thought: agent_core.py 存在，现在读取它的前几行。
Action: read_file
Action Input: {"path": "agent_core.py"}

Observation: import json
...

最终 Agent 会用 Final Answer 交付结果，而不是停留在「我建议你执行某某命令」——这就是「能干活的 Agent」和「聊天机器人」的本质区别。

7. 进阶方向

这个最小实现已经能跑通完整闭环，但要应对真实任务还需要几项增强：

① 结构化输出 / Function Calling

不少模型原生支持 function calling，返回的是结构化 JSON，解析更稳定。但手写解析的好处是不依赖特定厂商的专用协议，换任何 OpenAI 兼容模型都能跑。两者可以共存：优先用 function calling，失败时退回本文的文本协议。

② 流式输出与追问

上面的 chat 是同步等待完整返回。生产环境中建议接入流式输出，让 Agent 的思考过程实时可见；同时工具参数缺失时可以主动向用户追问，而不是硬猜。

③ 记忆与多轮会话

当前实现每次 run 都是独立上下文。加入长期记忆后，Agent 可以记住用户偏好和之前的工具调用结果，例如「上次替换的链接域名是 new.example.com」。

④ 更强的工具

可以扩展出 grep、http_get、run_shell（需谨慎加白名单）、sql_query 等工具。工具能力边界直接决定 Agent 的「能干」程度。

8. 总结

这篇文章从一个「只会调 API」的代码出发，实现了：

• Tool / ToolRegistry：用函数签名自动构建工具描述；
• LLMClient：与 OpenAI 兼容接口互通，不绑定特定厂商；
• ReActAgent：Thought-Action-Observation 循环，带错误自纠与步数上限；
• 三个真实工具：读文件、列目录、查时间，跑通了一个完整任务。

整个核心代码不到 150 行，无重型框架依赖。当你把这个循环跑起来，看着 Agent 自己决定调什么工具、根据观察修正计划、最后交付结果时，才算真正跨过了「调用大模型 API」的门槛，开始构建能替你干活的 AI Agent。

下一步，你可以把工具换成自己业务里的真实接口——查订单、发通知、改配置——Agent 能做的事会立刻丰富起来。