从零构建你的第一个 AI 智能体（不使用框架）

如何用约 150 行 Python 构建一个生产级别的智能体——不依赖 LangChain、不依赖 CrewAI、不依赖任何你无法逐行阅读的库。

一位创业公司创始人问了我一个我经常听到的问题："我们为什么要为一个智能体框架付费？我们的功能本质上就是'调用模型、调用 API、重试'啊。"他并不是在哭穷。他看过一个演示，框架"什么都能做"，然后花了两周时间与它的抽象概念搏斗，因为他的工作流并不符合框架对智能体应该如何工作的预设。框架没有错。只是它对你的问题做了一个猜测，而他的问题比那个猜测更具体。

所以我做了一件当框架碍事时我总是做的事：我徒手构建了这个智能体。大约 150 行 Python，除了一个 OpenAI 兼容的 HTTP 客户端外没有任何依赖。当我为他打开这个文件时，每一个 token 都是可追溯的——他可以精确地看到什么进入了上下文、模型返回了什么、以及循环在什么时候决定停止。他第二周就把它部署到了生产环境，到现在还在运行。

这篇文章就是那个构建过程，一步一步来。你最终会得到一个能够接收目标、使用工具、拥有工作记忆、遵守预算、以及在超出能力范围时升级的智能体——而且你将理解它的每一行代码。一旦你徒手构建过一个这样的智能体，每个框架就不再是魔法，而变成了一套你可以评估的观点。

为什么要徒手构建（既然框架已经存在）

让我坦诚地说这个权衡，因为确实存在权衡。像 LangChain 和 CrewAI 这样的框架将数月的模式压缩成配置，对于标准工作流——带检索的聊天、少量工具、一个编排器——它们确实可以为你节省一周时间。压缩后的版本同时也是你无法阅读的版本：当一个工具调用出现问题时，堆栈跟踪指向框架的内部，而框架的内存策略是你继承的设计决策，而不是你做出的决策。

徒手构建为你带来了三样配置无法给你的东西：

可读懂的上下文。 你可以看到进入模型的每一个 token。当智能体行为异常时，你可以复现它，因为你控制着组装过程。

真实的预算。 步数限制、成本限制和升级是你的代码，而不是框架文档深处某个你可能永远找不到的标志。

可调试的失败。 运行日志是你的。你精确地知道智能体尝试了什么、为什么尝试、以及在哪里放弃了。

代价是你要自己写循环——大约五十行。这整个构建的其余部分是工具、内存和护栏，这些东西你在框架内部也还是要写的。

我们要构建的架构

在代码之前，先看清这个东西的形态。我们的智能体运行一个包含四个组件的循环：

        ┌─────────────────────────────────────────────┐
        │                  AGENT LOOP                 │
        │                                             │
 User goal ──▶ assemble context ──▶ model decides ──┐ │
        │                             │             │ │
        │                  answer? ──▶ return       │ │
        │                  tool call ─▶ execute ────┼─┘
        │                             │             │
        └─────────────────────────────┼─────────────┘
                                      ▼
                            budget guards & escalation

Tools（工具）是由模型调用、由我们的运行时执行的声明式函数。

Memory（记忆）是通过向量索引按模型自身的相关性检索并注入到模型决策之前的上下文。

Guardrails（护栏）决定循环何时可以继续、何时必须停止。

Escalation（升级）是一个定义好的交接动作，包含可读的摘要。

Step 1：工具层

第一个要构建的是将模型的结构化请求转换为真实函数调用的机制。我把它保持得很简单：一个工具注册表，每个工具都有名称、描述、JSON schema 和一个 Python 可调用对象。

import json
from typing import Callable, Any

class Tool:
    def __init__(self, name: str, description: str, schema: dict, fn: Callable):
        self.name = name
        self.description = description
        self.schema = schema
        self.fn = fn

def to_openai_spec(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.schema,
            },
        }

def call(self, arguments: str) -> str:
        try:
            args = json.loads(arguments)
            result = self.fn(**args)
            return json.dumps(result)
        except (json.JSONDecodeError, TypeError, KeyError) as exc:
            return json.dumps({"error": f"invalid tool call: {exc}"})

def lookup_order(order_id: str) -> dict:
    # Production: query your orders DB here, with authz and caching.
    return {"order_id": order_id, "status": "paid", "amount": 14900}

TOOLS = [
    Tool(
        name="lookup_order",
        description=(
            "Look up an order by its ID. Returns status, amount in paise, "
            "and delivery status. Raises an error if the order does not exist."
        ),
        schema={
            "type": "object",
            "properties": {"order_id": {"type": "string"}},
            "required": ["order_id"],
        },
        fn=lookup_order,
    ),
]

有两个细节很重要。description 是模型的契约，不是给人类看的注释——模型阅读它来决定何时调用这个工具，所以它必须说明前置条件。call() 方法不会让异常崩溃循环；一次糟糕的调用会变成一个可读的工具结果，模型可以从中恢复。

Step 2：作为检索上下文的记忆

接下来是记忆。我要把它保持最小化但符合生产形态：一个存储过去事实的内存存储，通过向量索引按模型自身的相关性检索。对于第一个智能体，你不需要数据库服务器；你需要这一层，而且你需要它是以后可以替换的东西。

class SimpleMemory:
    def __init__(self, embed: Callable[[str], list[float]]):
        self.embed = embed
        self.items: list[tuple[str, list[float]]] = []

def remember(self, text: str) -> None:
        self.items.append((text, self.embed(text)))

def recall(self, query: str, top_k: int = 3) -> str:
        if not self.items:
            return "(no memory yet)"
        q = self.embed(query)
        scored = sorted(
            self.items,
            key=lambda item: _cosine(item[1], q),
            reverse=True,
        )
        return "\n---\n".join(text for text, _ in scored[:top_k])

对于 embedding 函数，使用任何 OpenAI 兼容端点，搭配 text-embedding-3-small 或本地模型。在生产环境中这会变成 pgvector 或 Qdrant；而接口——remember() 和 recall()——是跨这次替换保持不变的东西。这才是徒手构建这一层的真正价值：你拥有这个接缝。

Step 3：带护栏的循环

现在是核心部分。循环组装上下文、调用模型、解释响应。护栏在这里不是事后添加的；它们被写入循环本身，这样就不会被跳过。

from openai import OpenAI

class Agent:
    def __init__(self, model: str, tools: list[Tool], memory: SimpleMemory,
                 client: OpenAI):
        self.model = model
        self.tools = {t.name: t for t in tools}
        self.memory = memory
        self.client = client

def run(self, goal: str, max_steps: int = 6,
            max_cost_cents: float = 10.0) -> dict:
        system = (
            "You are an assistant that completes a goal using available "
            "tools. Rules: only call a tool when you need data; never invent "
            "tool results; if you cannot finish, escalate with a summary of "
            "what you tried and what is missing."
        )
        history = self.memory.recall(goal)
        messages = [
            {"role": "system", "content": system},
            {"role": "user", "content": f"RELEVANT PAST CONTEXT:\n{history}\n\nGOAL: {goal}"},
        ]
        steps = 0
        cost = 0.0
        for steps in range(1, max_steps + 1):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=[t.to_openai_spec() for t in self.tools.values()],
            )
            cost += _estimate_cost(response)  # tokens * rate
            if cost > max_cost_cents:
                return {"outcome": "escalated",
                        "summary": "cost budget exceeded",
                        "steps": steps, "cost_cents": cost}

msg = response.choices[0].message
            if not msg.tool_calls:
                self.memory.remember(f"goal: {goal} -> answer: {msg.content}")
                return {"outcome": "done", "answer": msg.content,
                        "steps": steps, "cost_cents": cost}

messages.append(msg)
            for call in msg.tool_calls:
                tool = self.tools.get(call.function.name)
                result = (
                    tool.call(call.function.arguments)
                    if tool else json.dumps({"error": "unknown tool"})
                )
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": result,
                })

return {"outcome": "escalated",
                "summary": "step budget exhausted, tried: " +
                           repr([m.get("content", "")[:80] for m in messages[-4:]]),
                "steps": steps, "cost_cents": cost}

读一读这些护栏，因为它们才是产品：步数预算使得循环不会永久运行，成本预算做到优雅失败，未知工具处理使得幻觉出来的工具名不会导致运行崩溃，以及内存写回使得下一个目标能从本次运行学到的东西继续。升级返回一个人类可读的摘要——那是人类收到的内容，而不是错误堆栈。

Step 4: 将各部分连接起来

def embed(text: str) -> list[float]:
    r = client.embeddings.create(model="text-embedding-3-small", input=text)
    return r.data[0].embedding

client = OpenAI()  # any OpenAI-compatible endpoint

memory = SimpleMemory(embed=embed)
agent = Agent(model="your-model", tools=TOOLS, memory=memory, client=client)

result = agent.run("Where is order ORD-9911 and has it been paid?")
print(result)

运行它，智能体会调用 lookup_order，获取状态，然后回答——或者在无法完成时升级。这就是整个骨架。它很小，因为循环很小。你以后要添加的一切——向量数据库、重试机制、多智能体、权限层——都附加在你刚刚构建的某个接缝上。

生产环境现实：我第一次构建时踩过的坑

我构建过足够多的这类系统，知道naive版本具体会怎么失败，你也会遇到它们。以下是它们找上你的顺序：

1. 工具描述太敷衍。我的第一个 lookup_order 写的是"获取订单信息"，结果模型为用户自己编造的订单 ID 也调用了它。把描述重写成写明前置条件和错误行为之后，大多数误用不通过任何代码修改就解决了。把描述当作文档来写，因为模型读的就是它。

2. 格式错误的 JSON 会导致运行崩溃。模型偶尔会发出截断的 JSON 参数，而我的第一个版本直接抛出异常，杀死了循环。Tool.call() 的错误路径——返回一个可读的错误而不是抛出异常——是救了我一命的设计。能收到工具错误讯息的模型可以恢复；崩溃的循环不能。

3. 成本护栏是我第一个删掉的东西，后来后悔了。在测试中，一次病态的运行在我注意到之前已经命中了数百次工具调用。从第一天就把成本预算放进去，别等到收到账单之后。

4. 内存写回污染了后续运行。我当时记住了每一条原始目标，几个会话之后召回的上下文就被噪音主导了。修复方法：记住蒸馏过的事实（"订单 ORD-9911：已付款，已送达"），而不是原始用户消息，并且对注入的 token 数量设上限。

5. 检索没有评估。当我换了 embedding 模型之后，召回质量悄然下降，我一周都没注意到。每次更换模型或工具时，用一组固定的测试目标和期望召回的事实来跑一遍评估。

Scale Up: 第二个工具、重试机制和权限层

骨架跑起来之后，下一步自然要问 scale。这里是从 150 行到客户愿意付钱的程度的一条诚实路径，以及每个新增功能如何附加到你已经构建的接缝上。

添加第二个工具是注册，不是重新设计。定义函数，写一份合同级别的描述，加到 TOOLS 列表里，循环自动处理剩下的。模型会开始根据描述来选择工具——这正是描述是你写过的最高杠杆文档的原因。当我给已经有 lookup_order 的智能体添加 send_refund 时，模型学会了先调用 lookup_order 确认资格，再调用 send_refund——两个从未被演示过的步骤，完全由描述驱动。

重试必须显式声明，否则循环会自己发明。一个 naive 循环遇到限速的工具时，下一步会再次调用，白白消耗预算和延迟。在两个地方之一添加重试行为：在工具层，遇到暂时性失败（429、503）时返回模型可以响应的特定错误；或者在循环层，加上固定尝试次数上限。不要让循环通过改写调用"决定"重试——那是贪婪循环的失败模式，花着真钱却什么都发现不了。

权限层改变的是循环，不是工具。对于变更操作——发邮件、退款、删除记录——智能体不应该直接执行。模式：在工具元数据中标记 requires_approval: true，当循环到达它时，返回一个 pending 结果并给人类一个带是否确认门的摘要。工具函数本身保持不变；运行时决定是执行还是等待。我交付过的智能体中，每个写操作都流经这道门，它是"智能体行动了"和"智能体提议了，人类批准了"之间的区别。

多智能体是最后一步，不是第一步。抵制把工作正常的单个智能体变成团队的冲动。每个智能体边界都是一次交接，每次交接都会丢失上下文和消耗 token。只在真正的约束要求拆分时才拆分——比如只读研究者与有写权限的操作者拥有不同权限——绝不要为了美观而拆分。你能追踪的单个循环，胜过你无法调试的五智能体团队。

何时不该手写

诚实的对比例子：当你的问题是标准流水线且框架的意见与之完全匹配时，不要手写智能体——检索聊天、固定工具表面、单编排器。你花在做胶水代码上的时间是一样的，而框架的生态系统（内存集成、追踪、部署）能节省真正的努力。当工作流不寻常、当你需要追踪每一个 token、或者当框架的抽象层正在消耗比它提供的更多的成本时，手写。

决策规则：如果你能用一段话描述你的循环，手写。如果不能，你也没准备好把它委托给框架。

发货前的检查清单

[ ] 工具描述写明了前置条件和错误行为

[ ] 格式错误的工具参数返回可读的错误，绝不导致循环崩溃

[ ] 未知的工具名被优雅处理（升级，不崩溃）

[ ] 步数预算和成本预算在循环内部强制执行

[ ] 升级返回一个人类可读的所有尝试摘要

[ ] 内存有一个可替换向量数据库的召回接缝

[ ] 内存写入蒸馏事实，而非原始日志

[ ] 工具结果被视为模型的不受信任输入

[ ] 一套评估集在每次模型/工具变更时测量召回质量

[ ] 完整运行被记录：步骤、工具调用、成本、延迟、结果

150 行代码改变了他的想法

那位询问为什么要为一个框架付钱的人，在文件打开的那一刻就得到了答案：循环只有六行，工具是一个注册表，内存是一个接缝，护栏是显式的。两周后，同样的论证又为另一个客户省下了一次他们不需要的迁移。循环不是困难的部分——它从来都不是。困难的部分是决定什么放进上下文、循环何时可以停止、以及什么样的失败算是优雅失败。那些都是决策，而它们属于你能阅读的代码。

用手写你的第一个智能体。读每一个 token。故意打破它。当你能信任地理解这个循环时，你就会准确知道框架何时在帮助你——以及何时它只是在猜测。