agent-">纯 Python 手搓 AI Agent：可行性与工程实践

一、问题的提出：为什么有人坚持不用通用框架

在真实项目落地过程中，一个反复出现的现象是：原型阶段用通用编排框架跑得飞快，一旦进入生产环境，就开始出现各种”别扭”——想加一个自定义的重试策略，得绕过框架的抽象层；想换一个模型供应商，发现逻辑和框架的调用约定绑死了；想看清楚某一步到底发生了什么，日志里全是框架内部的嵌套调用栈。

于是相当一部分团队选择了一条看起来更”笨”的路：核心逻辑自己写，通用能力才用现成方案。检索增强生成（RAG）这类标准化流程，用成熟的编排库完全够用；但 Agent 的决策循环、工具调度、状态管理，自己从零实现。

这不是为了炫技。观察行业里那些真正跑在生产环境的智能编程工具、代码助手、自动化运维 Agent，翻它们公开的技术分享会发现一个共同点：几乎清一色是定制架构，而不是把某个通用框架直接套上去。

原因归结为三个词：高效率、个性化、鲁棒性。通用框架为了覆盖尽可能多的场景，必然要做大量抽象和妥协，而这些妥协恰恰是生产级 Agent 最不能接受的。

打个比方：大厂造轮子，是因为别人的轮子跑不了他的路。想要更强，就得自己从零起步，并且走出不一样的路径。

正确的姿势不是”全都不用框架”，而是分层：

• 核心决策循环：自己实现，掌握完全控制权；
• 通用能力（RAG、文档加载、向量检索、自动化工作流）：用成熟库，别重复造轮子。

本文带你从零搭一个 AI Agent。不是玩具，是能跑起来、能扩展、能调试的那种。

---

二、先讲清楚：Agent 的本质是什么

在动手之前，概念必须先理明白。

早期的对话式 AI，本质是”你问它答”的聊天机器人，一次交互结束就完事。而 Agent 的不同在于它拥有工具：能检索网页、能生成图片、能调用外部 API、能操作数据库。复杂度取决于你要它干什么。

最简单的 Agent 可能就是一个客服场景：一个检索工具 + 一个发邮件工具，遇到搞不定的问题转人工。听起来很简单？但它确实就是一个 Agent。

那么 Agent 的本质，可以用八个字概括：

感知 — 推理 — 行动 — 学习

• 感知（Perceive）：接收用户输入、环境状态、工具返回结果；
• 推理（Reason）：基于当前上下文决定下一步做什么；
• 行动（Act）：调用工具、生成回复、修改状态；
• 学习（Learn）：把有效经验沉淀到长期记忆，优化后续决策。

本文会完整实现前三个环节。学习环节涉及向量数据库与长期记忆管理，属于进阶话题，将在后续内容中展开。

推理与行动的实现，业界普遍采用 ReAct 模式（Reasoning + Acting）。它的精髓是八个字：显式思考，显式行动。每一步的思考过程、调用的工具、拿到的结果都有明确记录，出了问题可以逐步回溯，而不是面对一个黑盒输出干瞪眼。

---

三、架构设计：三个必须分开的部分

写代码之前先把架构想清楚。我们要搭三个东西：

1. 工具层（Tool Layer）：统一注册、描述、调用所有外部能力；
2. LLM 客户端层（LLM Client）：封装模型调用，屏蔽供应商差异；
3. Agent 核心（Agent Core）：编排推理—行动循环，维护状态与历史。

3.1 为什么要这样分

第一，工具抽象带来可扩展性。

新增一个工具时，不应该改动核心代码。想加数据库查询？注册一下就行。这是所有正经 Agent 都在遵循的可扩展性原则。工具层要做的是统一接口：每个工具提供名称、用途描述、参数结构、执行函数。核心循环只需要面向这个接口编程，完全不关心工具背后是 HTTP 请求还是本地函数。

第二，LLM 与 Agent 分离，这是最关键的解耦。

LLM 是”大脑”，Agent 是指挥官，但你必须能换大脑——今天用这家模型，明天换那家，后天可能因为成本或合规要求再换一家。如果代码耦合在一起，换一次供应商就得重写一遍逻辑，纯粹是给自己找罪受。

分开之后的好处非常直接：

• 换供应商不改 Agent 逻辑；
• 一个服务不可用能切到另一个（故障回退）；
• 可以针对不同任务选用不同规格的模型，比如简单分类用轻量模型，复杂规划用高能力模型。

第三，状态与执行分离，便于调试。

Agent 的每一步都应该产生结构化记录：当前轮次、模型输出、选择的工具、工具入参、工具返回、耗时。这些记录既是调试依据，也是后续做可观测性建设的基础。

---

四、代码实现：从零构建

4.1 工具抽象层

from abc import ABC, abstractmethod
from typing import Any
class Tool(ABC):
    name: str
    description: str
    @abstractmethod
    def run(self, **kwargs) -> Any:
        ...
class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, Tool] = {}
    def register(self, tool: Tool) -> None:
        self._tools[tool.name] = tool
    def get(self, name: str) -> Tool:
        if name not in self._tools:
            raise KeyError(f"未注册的工具: {name}")
        return self._tools[name]
    def describe(self) -> str:
        return "\n".join(
            f"- {t.name}: {t.description}" for t in self._tools.values()
        )

这里的关键点在于：Tool 只规定三件事——名称、描述、执行入口。描述会被注入到提示词里，供模型判断何时调用。注册表负责查找与描述生成，核心循环只依赖注册表，不依赖任何具体工具。

举个例子，注册一个天气查询工具：

class WeatherTool(Tool):
    name = "get_weather"
    description = "查询指定城市的当前天气，参数为 city"
    def run(self, city: str) -> str:
        # 实际项目中此处调用外部 API
        return f"{city} 当前晴，26 摄氏度"
registry = ToolRegistry()
registry.register(WeatherTool())

之后想加数据库查询、发邮件、调用内部服务，都是同样的模式，核心代码一行不用动。

4.2 LLM 客户端层

这一层的目标只有一个：让上层拿到统一格式的输出，不关心底层是哪家服务。

class LLMClient(ABC):
    @abstractmethod
    def chat(self, messages: list[dict], **kwargs) -> str:
        ...
class ProviderA(LLMClient):
    def chat(self, messages, **kwargs) -> str:
        # 调用某供应商的接口，返回纯文本
        ...
class ProviderB(LLMClient):
    def chat(self, messages, **kwargs) -> str:
        # 调用另一供应商的接口，返回纯文本
        ...

再包一层带故障回退的调度器：

class ResilientLLM:
    def __init__(self, providers: list[LLMClient]):
        self.providers = providers
    def chat(self, messages, **kwargs) -> str:
        last_err = None
        for p in self.providers:
            try:
                return p.chat(messages, **kwargs)
            except Exception as e:
                last_err = e
                continue
        raise RuntimeError(f"所有模型服务均不可用: {last_err}")

这就是生产环境和玩具的分水岭之一。单一模型服务总会有抖动、限流、超时的时候，有了这层封装，Agent 不会因为一次调用失败就整体崩溃。

4.3 Agent 核心循环

采用 ReAct 思路，让模型在每一轮显式输出”思考”和”行动”。

import json
class Agent:
    def __init__(self, llm: ResilientLLM, tools: ToolRegistry, max_steps: int = 8):
        self.llm = llm
        self.tools = tools
        self.max_steps = max_steps
        self.trace = []
    def _build_prompt(self, task: str) -> list[dict]:
        system = (
            "你是一个可以调用工具的智能体。\n"
            "可用工具：\n" + self.tools.describe() + "\n\n"
            "每一步请严格输出 JSON：\n"
            '{"thought": "你的思考", "action": "工具名或 final", '
            '"action_input": {...} 或 "最终回答"}\n'
            "信息足够时，action 使用 final。"
        )
        return [
            {"role": "system", "content": system},
            {"role": "user", "content": task},
        ]
    def run(self, task: str) -> str:
        messages = self._build_prompt(task)
        for step in range(self.max_steps):
            raw = self.llm.chat(messages)
            decision = self._parse(raw)
            self.trace.append({"step": step, "decision": decision})
            if decision["action"] == "final":
                return decision["action_input"]
            tool = self.tools.get(decision["action"])
            try:
                result = tool.run(**decision.get("action_input", {}))
            except Exception as e:
                result = f"工具执行失败: {e}"
            self.trace.append({"step": step, "observation": result})
            messages.append({"role": "assistant", "content": raw})
            messages.append({
                "role": "user",
                "content": f"工具返回结果：{result}\n请继续。"
            })
        return "已达到最大步数限制，任务未完成。"
    @staticmethod
    def _parse(raw: str) -> dict:
        try:
            return json.loads(raw)
        except json.JSONDecodeError:
            return {"thought": raw, "action": "final", "action_input": raw}

这段代码不到百行，却已经具备了生产级 Agent 的骨架特征：

• 有明确的步数上限，防止无限循环；
• 每一步都有结构化记录，出问题可以回放；
• 工具异常被捕获，不会中断整个流程；
• 模型输出格式错误时有兜底解析。

4.4 一个完整的运行示例

llm = ResilientLLM([ProviderA(), ProviderB()])
registry = ToolRegistry()
registry.register(WeatherTool())
agent = Agent(llm, registry)
answer = agent.run("北京今天适合穿短袖吗？")
print(answer)
for item in agent.trace:
    print(item)

执行过程大致是：模型先思考”需要知道北京天气”，决定调用天气工具；工具返回结果后，模型结合温度信息判断是否适合穿短袖，输出最终回答。整个过程透明可查。

---

五、从能跑到好用：三个必须补齐的工程能力

5.1 工具描述的质量决定调用准确率

工具描述不是注释，而是提示词的一部分。描述里应该写清楚：这个工具做什么、参数含义、适用场景、不适用场景。描述越精确，模型选错工具的概率越低。这是一个被严重低估的优化点——很多时候 Agent 表现差，不是模型不行，而是工具描述太含糊。

5.2 错误处理要分级

• 可重试错误（超时、限流）：退避重试；
• 可降级错误（某工具不可用）：尝试替代工具或跳过；
• 不可恢复错误（参数非法、权限不足）：把错误信息返回给模型，让它重新规划。

把错误信息喂回模型，往往能让它自己纠正路径，这比直接抛出异常终止任务要优雅得多。

5.3 可观测性不能省

trace 只是最小实现。生产环境需要把它接入统一的日志与监控体系，记录每步耗时、模型 token 消耗、工具成功率。没有这些数据，优化就无从谈起。

---

六、结论：核心自研，通用借用

回到最初的问题：纯 Python 手搓 AI Agent，可行吗？

完全可行，而且在生产场景下往往是更优解。

框架的价值在于快速验证想法，在原型开发、标准化 RAG 流程、常规自动化工作流上，用成熟方案确实快。但一旦你对效率、个性化、鲁棒性有要求，核心决策逻辑就必须掌握在自己手里。

合理的分工是：

• 核心循环、工具抽象、状态管理：自研，掌握控制权；
• 检索、文档处理、向量存储、工作流编排：使用成熟库，不重复造轮子。

本文实现的代码骨架不到两百行，但每一行你都能说清楚它在干什么。这就是自研的最大价值——不是更少的代码，而是完全的可见性与可修改性。当你需要排查一个诡异的行为、需要针对业务做定向优化时，你会庆幸这套逻辑是你自己写的。