从零实现 Agent 内核

这个教程不调用任何 Agent 框架。我们从空目录开始，写出一个能跑、能测试、能观察的 Agent 内核。完成后你会真正理解循环、工具、预算和 trace 是怎么拼起来的。

你将得到什么

一个命令行 Agent：

bash

python -m miniagent "ACME 毛利率从 42.4% 降到 38.2%，下降了多少个百分点？"

它会：

判断需要计算，调用 calculator；
拿到结果后继续推理；
输出最终答案；
打印每一步的 trace、token 与耗时。

代码约 300 行，依赖只有 openai 和 pytest。

前置条件

Python 3.11+
一个兼容 OpenAI Chat Completions 的 API（OpenAI、DeepSeek、Qwen、本地 vLLM 都可以）
能读官方文档、会看报错

bash

mkdir miniagent && cd miniagent
python -m venv .venv
# Windows
.venv\Scripts\activate
# macOS / Linux
source .venv/bin/activate

pip install openai pytest

目录结构：

text

miniagent/
├─ __init__.py
├─ core.py          # Tool / Budget / 事件类型
├─ tools.py         # 工具实现与 Schema
├─ llm.py           # 模型适配层
├─ agent.py         # 循环
├─ __main__.py      # CLI
└─ tests/
   └─ test_agent.py

Step 1：定义核心类型

core.py：

python

from __future__ import annotations

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


class BudgetExceeded(RuntimeError):
    pass


@dataclass
class Tool:
    name: str
    description: str
    parameters: dict
    func: Callable[..., dict]
    read_only: bool = True

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


@dataclass
class LLMResult:
    content: str | None
    tool_calls: list[dict] = field(default_factory=list)
    input_tokens: int = 0
    output_tokens: int = 0


@dataclass
class Budget:
    max_steps: int = 8
    max_tokens: int = 60_000
    max_seconds: float = 120.0
    tokens: int = 0
    started: float = field(default_factory=time.monotonic)

    def check(self) -> None:
        if self.tokens > self.max_tokens:
            raise BudgetExceeded(f"token 预算耗尽：{self.tokens}/{self.max_tokens}")
        if time.monotonic() - self.started > self.max_seconds:
            raise BudgetExceeded("时间预算耗尽")

    def add(self, result: LLMResult) -> None:
        self.tokens += result.input_tokens + result.output_tokens

三个设计决定：

预算独立成对象：后面加金额、工具调用次数都方便；
LLMResult 与厂商解耦：循环不直接依赖 OpenAI SDK；
Tool 带 read_only：为后续权限分级留位置。

Step 2：写三个工具

tools.py：

python

from __future__ import annotations

import ast
import operator
from datetime import datetime, timezone

from .core import Tool

_OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.Pow: operator.pow,
    ast.USub: operator.neg,
}

_DOCS = {
    "react": "ReAct 让模型交替推理与行动，用工具结果更新上下文（Yao et al., 2022）。",
    "rag": "RAG 在生成前检索外部文档，把相关内容放进上下文，降低幻觉。",
    "mcp": "MCP 用统一协议连接模型应用与外部工具、资源、提示模板。",
}


def _eval(node: ast.AST) -> float:
    if isinstance(node, ast.Expression):
        return _eval(node.body)
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return node.value
    if isinstance(node, ast.BinOp) and type(node.op) in _OPS:
        return _OPS[type(node.op)](_eval(node.left), _eval(node.right))
    if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
        return _OPS[type(node.op)](_eval(node.operand))
    raise ValueError("只支持数字和 + - * / **")


def calculator(expression: str) -> dict:
    try:
        value = _eval(ast.parse(expression, mode="eval"))
        return {"ok": True, "expression": expression, "value": value}
    except Exception as e:
        return {
            "ok": False,
            "error": {"code": "BAD_EXPRESSION", "message": str(e), "retryable": True},
        }


def search_docs(query: str, limit: int = 3) -> dict:
    q = query.lower().strip()
    hits = [{"id": k, "text": v} for k, v in _DOCS.items() if q in k or q in v.lower()]
    if not hits:
        return {"ok": True, "summary": "没有找到相关资料。", "items": []}
    return {"ok": True, "summary": f"找到 {len(hits)} 条资料。", "items": hits[:limit]}


def now() -> dict:
    return {"ok": True, "utc": datetime.now(timezone.utc).isoformat()}


TOOLS = [
    Tool(
        name="calculator",
        description="计算数学表达式。涉及精确算术时使用，不要心算。",
        parameters={
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "例如 (42.4-38.2)"}
            },
            "required": ["expression"],
            "additionalProperties": False,
        },
        func=calculator,
    ),
    Tool(
        name="search_docs",
        description="检索 AI Agent 主题的资料。问题涉及概念、定义、出处时使用。",
        parameters={
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "检索关键词，尽量简短"},
                "limit": {"type": "integer", "minimum": 1, "maximum": 5},
            },
            "required": ["query"],
            "additionalProperties": False,
        },
        func=search_docs,
    ),
    Tool(
        name="now",
        description="获取当前 UTC 时间。涉及「今天」「现在」时使用。",
        parameters={"type": "object", "properties": {}, "additionalProperties": False},
        func=now,
    ),
]

注意 calculator 没有用 eval，而是用 AST 只允许数字和四则运算。让模型生成代码再执行时，这是最低限度的隔离。

Step 3：模型适配层

llm.py：

python

from __future__ import annotations

import os
import time

from openai import APITimeoutError, OpenAI, RateLimitError

from .core import LLMResult


class LLM:
    def __init__(self, model: str | None = None):
        self.client = OpenAI(
            base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
            api_key=os.getenv("OPENAI_API_KEY"),
        )
        self.model = model or os.getenv("MODEL", "gpt-4.1-mini")

    def complete(self, messages: list[dict], tools: list[dict]) -> LLMResult:
        last_error: Exception | None = None
        for attempt in range(3):
            try:
                resp = self.client.chat.completions.create(
                    model=self.model,
                    messages=messages,
                    tools=tools or None,
                    tool_choice="auto" if tools else None,
                    temperature=0,
                    max_tokens=800,
                    timeout=30,
                )
                msg = resp.choices[0].message
                calls = []
                for tc in msg.tool_calls or []:
                    calls.append(
                        {
                            "id": tc.id,
                            "type": "function",
                            "function": {
                                "name": tc.function.name,
                                "arguments": tc.function.arguments,
                            },
                        }
                    )
                return LLMResult(
                    content=msg.content,
                    tool_calls=calls,
                    input_tokens=resp.usage.prompt_tokens if resp.usage else 0,
                    output_tokens=resp.usage.completion_tokens if resp.usage else 0,
                )
            except (APITimeoutError, RateLimitError) as e:
                last_error = e
                time.sleep(0.5 * (2**attempt))
        raise RuntimeError(f"模型调用失败：{last_error}")

只对超时和限流重试，其他错误直接抛出。生产代码还应区分 5xx、内容拒绝和鉴权失败。

Step 4：循环

agent.py：

python

from __future__ import annotations

import json
import time
from typing import Callable

from .core import Budget, BudgetExceeded, Tool
from .llm import LLM

SYSTEM = """你是一个严谨的助手。
规则：
1. 需要外部事实时调用 search_docs；需要精确计算时调用 calculator；涉及当前时间调用 now。
2. 不要编造资料中没有的内容；信息不足时明确说明。
3. 工具报错时根据错误信息修正一次；仍失败就说明原因。
4. 信息足够后直接回答，用中文，先结论后依据，不超过 300 字。"""


class Agent:
    def __init__(
        self,
        llm: LLM,
        tools: list[Tool],
        system: str = SYSTEM,
        budget: Budget | None = None,
        on_event: Callable[[dict], None] | None = None,
    ):
        self.llm = llm
        self.system = system
        self.budget = budget or Budget()
        self.on_event = on_event
        self.events: list[dict] = []
        self.tools: dict[str, Tool] = {}

        for t in tools:
            if t.name in self.tools:
                raise ValueError(f"重复工具名：{t.name}")
            self.tools[t.name] = t

    def emit(self, kind: str, **data) -> None:
        event = {"type": kind, "ts": time.time(), **data}
        self.events.append(event)
        if self.on_event:
            self.on_event(event)

    def run(self, goal: str) -> str:
        messages = [
            {"role": "system", "content": self.system},
            {"role": "user", "content": goal},
        ]
        seen: set[str] = set()

        try:
            for step in range(1, self.budget.max_steps + 1):
                self.budget.check()
                self.emit("step_start", step=step)

                result = self.llm.complete(messages, [t.schema() for t in self.tools.values()])
                self.budget.add(result)
                self.emit(
                    "llm",
                    step=step,
                    tokens={
                        "input": result.input_tokens,
                        "output": result.output_tokens,
                    },
                    has_tools=bool(result.tool_calls),
                )

                assistant = {"role": "assistant", "content": result.content}
                if result.tool_calls:
                    assistant["tool_calls"] = result.tool_calls
                messages.append(assistant)

                if not result.tool_calls:
                    answer = result.content or ""
                    self.emit("final", step=step, answer=answer)
                    return answer

                for call in result.tool_calls:
                    name = call["function"]["name"]
                    raw_args = call["function"]["arguments"] or "{}"
                    args: dict = {}
                    try:
                        args = json.loads(raw_args)
                        if not isinstance(args, dict):
                            raise ValueError("参数必须是 JSON 对象")
                    except Exception as e:
                        payload = {
                            "ok": False,
                            "error": {
                                "code": "BAD_ARGUMENTS",
                                "message": f"参数不是合法 JSON：{e}",
                                "retryable": True,
                            },
                        }
                    else:
                        fingerprint = name + json.dumps(args, sort_keys=True, ensure_ascii=False)
                        if fingerprint in seen:
                            payload = {
                                "ok": False,
                                "error": {
                                    "code": "REPEATED_CALL",
                                    "message": "相同工具和参数已经调用过，结果不会变化。请换方法或停止。",
                                    "retryable": False,
                                },
                            }
                        else:
                            seen.add(fingerprint)
                            payload = self.execute(name, args)

                    self.emit("tool", step=step, name=name, args=args, result=payload)
                    messages.append(
                        {
                            "role": "tool",
                            "tool_call_id": call["id"],
                            "content": json.dumps(payload, ensure_ascii=False)[:4000],
                        }
                    )
        except BudgetExceeded as e:
            self.emit("budget_exceeded", reason=str(e))
            return f"任务未完成：{e}"

        self.emit("max_steps", steps=self.budget.max_steps)
        return "达到最大步数，任务未完成。请缩小问题范围或转人工。"

    def execute(self, name: str, args: dict) -> dict:
        tool = self.tools.get(name)
        if tool is None:
            return {
                "ok": False,
                "error": {
                    "code": "UNKNOWN_TOOL",
                    "message": f"未知工具：{name}",
                    "retryable": False,
                },
            }
        try:
            result = tool.func(**args)
            if not isinstance(result, dict):
                return {"ok": True, "result": result}
            return result
        except TypeError as e:
            return {
                "ok": False,
                "error": {
                    "code": "BAD_PARAMETERS",
                    "message": f"参数不匹配：{e}",
                    "retryable": True,
                },
            }
        except Exception as e:
            return {
                "ok": False,
                "error": {
                    "code": "TOOL_ERROR",
                    "message": str(e),
                    "retryable": False,
                },
            }

这里的几个关键点：

重复调用检测：相同工具 + 参数只允许一次；
错误结构化：模型能根据 retryable 和 message 决定下一步；
结果截断：[:4000] 防止工具返回撑爆上下文；
预算在每一步前检查：不是最后才算账。

Step 5：命令行入口

__main__.py：

python

from __future__ import annotations

import argparse
import json

from .agent import Agent
from .llm import LLM
from .tools import TOOLS


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("goal")
    parser.add_argument("--trace", action="store_true")
    args = parser.parse_args()

    def on_event(event: dict) -> None:
        if args.trace:
            print(json.dumps(event, ensure_ascii=False))

    agent = Agent(LLM(), TOOLS, on_event=on_event)
    answer = agent.run(args.goal)
    print("\n" + answer)


if __name__ == "__main__":
    main()

运行：

bash

export OPENAI_API_KEY=sk-...
export MODEL=gpt-4.1-mini
python -m miniagent "ACME 毛利率从 42.4% 降到 38.2%，下降了多少个百分点？" --trace

预期 trace：

text

{"type":"step_start","step":1}
{"type":"llm","step":1,"tokens":{"input":812,"output":24},"has_tools":true}
{"type":"tool","step":1,"name":"calculator","args":{"expression":"42.4-38.2"},"result":{"ok":true,"value":4.2}}
{"type":"step_start","step":2}
{"type":"llm","step":2,"tokens":{"input":901,"output":38},"has_tools":false}
{"type":"final","step":2,"answer":"毛利率下降了 4.2 个百分点。"}

Step 6：写测试（不用真实模型）

tests/test_agent.py：

python

from miniagent.agent import Agent
from miniagent.core import Budget, LLMResult, Tool
from miniagent.tools import TOOLS


class FakeLLM:
    def __init__(self, results):
        self.results = list(results)
        self.i = 0

    def complete(self, messages, tools):
        result = self.results[self.i]
        self.i += 1
        return result


def test_calls_calculator_then_answers():
    fake = FakeLLM(
        [
            LLMResult(
                content=None,
                tool_calls=[
                    {
                        "id": "c1",
                        "type": "function",
                        "function": {
                            "name": "calculator",
                            "arguments": '{"expression":"42.4-38.2"}',
                        },
                    }
                ],
                input_tokens=100,
                output_tokens=10,
            ),
            LLMResult(content="下降了 4.2 个百分点。", input_tokens=120, output_tokens=12),
        ]
    )
    agent = Agent(fake, TOOLS, budget=Budget(max_steps=4))
    assert agent.run("算一下") == "下降了 4.2 个百分点。"
    assert [e["type"] for e in agent.events] == ["step_start", "llm", "tool", "step_start", "llm", "final"]


def test_repeated_tool_call_is_blocked():
    call = {
        "id": "c1",
        "type": "function",
        "function": {"name": "now", "arguments": "{}"},
    }
    fake = FakeLLM(
        [
            LLMResult(content=None, tool_calls=[call]),
            LLMResult(content=None, tool_calls=[{**call, "id": "c2"}]),
            LLMResult(content="停。"),
        ]
    )
    agent = Agent(fake, TOOLS, budget=Budget(max_steps=4))
    agent.run("现在几点")
    repeated = [e for e in agent.events if e["type"] == "tool"]
    assert repeated[1]["result"]["error"]["code"] == "REPEATED_CALL"


def test_budget_stops_loop():
    class LoopLLM:
        def complete(self, messages, tools):
            return LLMResult(
                content=None,
                tool_calls=[
                    {
                        "id": "x",
                        "type": "function",
                        "function": {"name": "now", "arguments": "{}"},
                    }
                ],
                input_tokens=10,
                output_tokens=5,
            )

    agent = Agent(LoopLLM(), TOOLS, budget=Budget(max_steps=2))
    assert "最大步数" in agent.run("循环")

bash

pytest -q

测试不调用真实 API，跑一次不到 1 秒。模型质量用评估集测，循环正确性用单元测试测，两者分开。

验收清单

能跑通一个需要工具的任务
trace 能看到每一步的 token 与工具结果
重复调用会被阻止
超过步数/预算能停下
工具报错不会让进程崩溃
三个测试全部通过
不依赖任何 Agent 框架

常见错误

把 assistant 的 tool_calls 丢了

回填工具结果时，必须把带 tool_calls 的 assistant 消息先加入历史，否则下一次调用会缺少对应关系，API 会报错。

工具结果不截断

一次返回 100KB，下一轮上下文直接爆。工具层裁剪，循环层再兜底。

用文本协议解析 Action

正则解析 Action: xxx 很脆弱。用原生 Function Calling，模型返回结构化 tool_calls。

没有记录 token

没有 usage 就无法做预算、成本分析和模型对比。每次调用都要记录。

继续扩展

并行工具调用：同一轮多个 tool_calls 用 asyncio.gather 执行；
流式输出：把 complete 换成流式接口，发出 text_delta 事件；
持久化：把 events 和 messages 存 SQLite，支持中断恢复；
人工审批：写工具前发 approval_required 事件，等待外部确认；
评估：接下一篇文章的黄金集，统计工具选择准确率。

小结

Agent 内核 = 消息历史 + 工具注册 + 循环 + 预算 + 错误处理。
把模型调用封装成适配层，循环才能被测试。
结构化错误和重复调用检测，比换更强的模型更能提升稳定性。
trace 不是可选项，它是你调试和计费的基础。

下一篇在这个内核上加检索：从 PDF 到带引用的 RAG。

#你将得到什么

#前置条件

#Step 1：定义核心类型

#Step 2：写三个工具

#Step 3：模型适配层

#Step 4：循环

#Step 5：命令行入口

#Step 6：写测试（不用真实模型）

#验收清单

#常见错误

#继续扩展

#小结

相关阅读

你将得到什么

前置条件

Step 1：定义核心类型

Step 2：写三个工具

Step 3：模型适配层

Step 4：循环

Step 5：命令行入口

Step 6：写测试（不用真实模型）

验收清单

常见错误

继续扩展

小结