从零开发一个 Agent

读十篇 Agent 教程，不如亲手跑通一个会"干活"的 Agent。本文用最小可实现路径，从零做一个能调用工具完成任务的小助手——"帮我查天气并写进日程"——每步给出可运行代码，让你看清 Agent 的内核不过是一个循环。

**智能体（Agent）**是一种以大语言模型为决策中枢、能够调用外部工具完成任务、并能感知工具结果继续行动的软件系统。它和"普通调用 LLM"的区别只有一条：模型不只是回答问题，还决定"下一步该做什么"——查资料、执行代码、调 API、写文件，做完一步再看结果决定下一步。

为什么要亲手写？因为"Agent"最容易被营销词包装成玄学，但它的内核极其朴素：一个 while 循环。框架（LangGraph、OpenAI Agents SDK）做的只是把这个循环工程化——加状态管理、加记忆、加并行、加可观测性。你亲手实现一遍循环，之后用任何框架都能秒懂它在干什么。概念全景见Agent 核心概念与Manus 与 Agent 应用。

先看整体地图：

用户请求
   │
   ▼
┌─────────────────────┐
│ LLM 决策层（ReAct）  │ ← 循环：思考 → 决定调用哪个工具
└────────┬────────────┘
         │ 调用工具（function call）
         ▼
┌─────────────────────┐
│  工具层（Tools）     │  get_weather / add_event / ...
└────────┬────────────┘
         │ 把结果喂回 LLM
         ▼
┌─────────────────────┐
│  记忆层（Memory）    │  对话历史 / 向量记忆 / 文件
└────────┬────────────┘
         │ 结果已满足用户请求
         ▼
       最终答案

我们的实现路径分五步：定义工具 → 工具调用循环 → 加记忆 → 加规划 → 完整 Demo。

前置知识

本文假设你已了解：Agent 是什么（Agent 核心概念）、LLM 的基本能力边界（大语言模型）、怎么给模型写指令（提示词工程）。想先看理论上的 ReAct 循环怎么推理，可对照Transformer 与注意力机制理解"上下文"如何承载整个循环的记忆。

一、定义工具：function calling schema

要让 LLM"会用"工具，需要给它一份机器可读的工具说明书（schema），说明每个工具叫什么、干什么、参数有哪些。LLM 不会执行任何代码，它只是输出一段结构化的"工具调用请求"，由你的程序解析后执行。

① 工具实现（真正的逻辑）

import json
from pathlib import Path

def get_weather(city: str, date: str = "today") -> str:
    """查询某个城市的天气。演示用模拟数据，真实项目替换为天气 API。"""
    import random
    conditions = ["晴", "多云", "小雨", "大风"]
    return json.dumps({
        "city": city, "date": date,
        "weather": random.choice(conditions),
        "temp": f"{random.randint(18, 34)}℃",
    }, ensure_ascii=False)

def add_event(date: str, time: str, title: str) -> str:
    """往日历写入一条日程。演示用本地 JSON 文件，真实项目接日历 API。"""
    path = Path("events.json")
    events = json.loads(path.read_text(encoding="utf-8")) if path.exists() else []
    events.append({"date": date, "time": time, "title": title})
    path.write_text(json.dumps(events, ensure_ascii=False, indent=2), encoding="utf-8")
    return f"OK，已把「{title}」写入 {date} {time}"

② function calling schema（OpenAI 风格，各家大同小异）

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询某个城市某天的天气，返回天气现象和气温。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名，例如：北京、上海"},
                    "date": {"type": "string", "description": "日期（YYYY-MM-DD），缺省为今天"},
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "add_event",
            "description": "把一条日程写入日历。",
            "parameters": {
                "type": "object",
                "properties": {
                    "date": {"type": "string", "description": "日程日期（YYYY-MM-DD）"},
                    "time": {"type": "string", "description": "日程时间（HH:MM）"},
                    "title": {"type": "string", "description": "日程标题"},
                },
                "required": ["date", "time", "title"],
            },
        },
    },
]

工具描述决定成败

schema 里最关键的不是 name 而是 description 和参数 description——LLM 靠描述决定"什么时候该用、参数怎么填"。描述含糊（如"处理天气"）会导致模型该用时不用、参数乱填；描述要写清触发条件、参数含义、取值范围。这本质上是提示词工程，方法论见提示词工程。

工具实现与 schema 之间的派发器（dispatcher）——把模型请求的名字和 JSON 参数映射到真实函数：

def dispatch(name: str, args_json: str) -> str:
    """把 LLM 的工具调用请求派发到真实实现。"""
    args = json.loads(args_json)
    if name == "get_weather":
        return get_weather(args.get("city"), args.get("date", "today"))
    if name == "add_event":
        return add_event(args["date"], args["time"], args["title"])
    return f"错误：未知工具 {name}"

二、LLM 工具调用循环：ReAct

工具定义好后，核心就是那个循环。经典范式叫 ReAct（Reasoning + Acting）：模型先"推理"（reason）下一步该做什么，再"行动"（act）——输出工具调用请求；你的程序执行工具，把结果作为"观察"（observation）喂回；模型基于观察继续推理。如此循环，直到模型认为任务完成、直接输出答案。这个概念与提示词工程的关系详见提示词工程。

循环结构（伪代码）：
while 步数 < max_steps:
    reply = LLM(messages, tools)          # 1. 让模型决定下一步
    if reply 没有工具调用: return reply   # 2a. 完成 → 输出最终答案
    for 每个工具调用:
        result = dispatch(工具名, 参数)    # 2b. 执行工具
        messages.append(tool 消息)         # 3. 把观察结果喂回对话

完整实现（OpenAI 新版接口）：

from openai import OpenAI

client = OpenAI()   # 需要环境变量 OPENAI_API_KEY；本地 Ollama 见文末

def run_agent(user_query: str, max_steps: int = 6, model: str = "gpt-4o-mini") -> str:
    messages = [{"role": "user", "content": user_query}]
    for step in range(max_steps):
        resp = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=TOOLS,          # 把工具 schema 给模型
        )
        msg = resp.choices[0].message
        messages.append(msg)      # 模型的回复（含工具调用请求）进入对话

        if not msg.tool_calls:    # 模型决定直接回答 → 任务完成
            return msg.content

        # 逐个执行工具调用，结果以 role="tool" 的消息喂回
        for call in msg.tool_calls:
            result = dispatch(call.function.name, call.function.arguments)
            print(f"[step {step}] 调用 {call.function.name}({call.function.arguments}) -> {result}")
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            })
    return "已达到最大步数限制，任务未完成。"

跑一次：

print(run_agent("北京今天适合跑步吗？查一下天气再给建议。"))

典型输出轨迹（print 的那几行）会是这样：

[step 0] 调用 get_weather({"city":"北京","date":"today"}) -> {"city":"北京","date":"today","weather":"晴","temp":"27℃"}
北京今天晴、27℃，非常适合跑步。建议傍晚出门，注意补水。

三个容易忽略的细节

1. 消息历史不能丢：模型回复 msg 必须原样 append 进 messages（包括其中的 tool_calls 字段），否则工具结果和调用对不上，模型会"失忆"。
2. 一个回复可能调用多个工具：msg.tool_calls 是列表，要逐个执行、逐个回填。
3. tool_call_id 必须回传：工具结果消息里的 tool_call_id 与调用一一对应，这是多工具场景不出错的关键。

本地 Ollama 版：只需换 client 与模型名，其余代码不变，实现完全离线的 Agent：

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
model = "qwen2.5:7b"     # 需支持 function calling；先 ollama pull qwen2.5

三、加记忆

循环里所有信息都在 messages 里流动，这本身就是会话内记忆——多轮对话、上一步的工具结果都靠它。但有两类记忆需要显式设计：

记忆类型	存什么	实现	何时有用
会话内记忆	本次任务的全部消息	就是 messages 列表	多步任务、多轮对话
长期记忆	用户偏好、事实、历史结论	文件 / 数据库 / 向量库	跨会话复用、个性化
工作记忆	中间计算结果	变量、临时文件	复杂计算任务
向量记忆	"语义可检索"的历史信息	向量库 + 相似度检索	历史太多时按需召回

会话内记忆无需额外代码（上文循环天然支持）。向量记忆的典型实现：任务中产生的关键结论写入向量库，之后按需检索，避免把全部历史塞进上下文（上下文越长越贵，见推理优化与量化）。原理与实现见向量数据库与语义检索：

import numpy as np
import faiss
from sentence_transformers import SentenceTransformer

mem_model = SentenceTransformer("BAAI/bge-small-zh-v1.5")
mem_index = faiss.IndexFlatIP(384)   # 语义记忆库（进程内）
memory_texts: list[str] = []

def remember(text: str) -> None:
    v = mem_model.encode([text], normalize_embeddings=True)
    mem_index.add(v.astype("float32"))
    memory_texts.append(text)

def recall(query: str, k: int = 3) -> list[str]:
    if not memory_texts:
        return []
    qv = mem_model.encode([query], normalize_embeddings=True)
    _, ids = mem_index.search(qv.astype("float32"), k)
    return [memory_texts[i] for i in ids[0] if i >= 0]

# 例子：把"用户每周二晨跑"写进记忆，下次自动提到
remember("用户的习惯：每周二早上 7 点晨跑")
print(recall("这个用户有什么运动习惯？"))   # ['用户的习惯：每周二早上 7 点晨跑']

记忆的容量管理

所有记忆最终都要变成 token 进上下文。会话长了要截断或压缩（保留最近 N 轮 + 摘要）；向量记忆召回 k 要小。没有容量管理的 Agent，会在第 20 轮对话后把上下文撑爆、又贵又慢。

四、加规划：任务分解

复杂任务（"帮我安排周末"）一次工具调用往往做不完。加一层规划器：先让 LLM 把大任务拆成有序子任务，再逐个执行、合并结果。这对应 Agent 的 planning 能力：

def plan(task: str) -> list[str]:
    """把任务拆成 2~4 个有序子任务，每行一个。"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system",
             "content": "把用户任务拆成 2~4 个有序子任务，每行一个，只输出子任务本身。"},
            {"role": "user", "content": task},
        ],
    )
    return [ln.strip("- ").strip() for ln in resp.choices[0].message.content.splitlines() if ln.strip()]

然后让执行循环带规划跑：先 plan，再对每个子任务调用 run_agent，最后汇总：

def run_with_plan(task: str) -> str:
    steps = plan(task)
    print("规划：", steps)
    results = [run_agent(f"子任务：{s}") for s in steps]
    # 汇总子任务结果，交给 LLM 生成最终答复
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": "根据下面的子任务执行结果，给用户一个完整、友好的最终答复。"},
                  {"role": "user", "content": "\n\n".join(f"### {s}\n{r}" for s, r in zip(steps, results))}],
    )
    return resp.choices[0].message.content

规划的两层皮

规划有两种做法：显式规划（上面的先拆后做，可控、可审计）与隐式规划（ReAct 循环里模型自己边做边想，灵活但难预测）。入门阶段先做显式：拆步、执行、汇总三步走，出了问题能定位到具体子任务。复杂任务的规划边界与"任务分解"的深入讨论见Agent 核心概念。

五、完整可运行 Demo：查天气并写进日程

把以上全部拼成 agent_demo.py。目标任务："帮我查一下北京和上海明天的天气，把天气更好的那个城市安排一次明天下午 3 点的户外跑步。"

"""
agent_demo.py —— 最小可运行的 Agent：查天气 + 写日程

用法：python agent_demo.py
依赖：pip install openai  （Ollama 本地版：见代码末尾注释）
"""
import json
import random
from pathlib import Path

from openai import OpenAI

# ── 工具实现 ─────────────────────────────────────────────
def get_weather(city: str, date: str = "today") -> str:
    conditions = ["晴", "多云", "小雨", "大风"]
    return json.dumps({"city": city, "date": date,
                       "weather": random.choice(conditions),
                       "temp": f"{random.randint(18, 34)}℃"},
                      ensure_ascii=False)

def add_event(date: str, time: str, title: str) -> str:
    path = Path("events.json")
    events = json.loads(path.read_text(encoding="utf-8")) if path.exists() else []
    events.append({"date": date, "time": time, "title": title})
    path.write_text(json.dumps(events, ensure_ascii=False, indent=2), encoding="utf-8")
    return f"OK，已把「{title}」写入 {date} {time}"

# ── function calling schema ─────────────────────────────
TOOLS = [
    {"type": "function", "function": {
        "name": "get_weather",
        "description": "查询某个城市某天的天气，返回天气现象和气温。",
        "parameters": {"type": "object",
                       "properties": {"city": {"type": "string", "description": "城市名"},
                                      "date": {"type": "string", "description": "日期，缺省 today"}},
                       "required": ["city"]}}},
    {"type": "function", "function": {
        "name": "add_event",
        "description": "把一条日程写入日历。",
        "parameters": {"type": "object",
                       "properties": {"date": {"type": "string", "description": "日期 YYYY-MM-DD"},
                                      "time": {"type": "string", "description": "时间 HH:MM"},
                                      "title": {"type": "string", "description": "标题"}},
                       "required": ["date", "time", "title"]}}},
]

def dispatch(name: str, args_json: str) -> str:
    args = json.loads(args_json)
    if name == "get_weather":
        return get_weather(args.get("city"), args.get("date", "today"))
    if name == "add_event":
        return add_event(args["date"], args["time"], args["title"])
    return f"错误：未知工具 {name}"

# ── ReAct 循环 ───────────────────────────────────────────
client = OpenAI()                       # 需 OPENAI_API_KEY
# 本地版：client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
#         model = "qwen2.5:7b"

def run_agent(user_query: str, max_steps: int = 8, model: str = "gpt-4o-mini") -> str:
    messages = [{"role": "user", "content": user_query}]
    for step in range(max_steps):
        resp = client.chat.completions.create(model=model, messages=messages, tools=TOOLS)
        msg = resp.choices[0].message
        messages.append(msg)
        if not msg.tool_calls:
            return msg.content
        for call in msg.tool_calls:
            result = dispatch(call.function.name, call.function.arguments)
            print(f"[step {step}] {call.function.name}({call.function.arguments}) -> {result}")
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
    return "已达到最大步数限制，任务未完成。"

if __name__ == "__main__":
    q = "帮我查一下北京和上海明天的天气，把天气更好的那个城市安排一次明天下午 3 点的户外跑步。"
    print(run_agent(q))
    print("—— 日程文件 events.json 内容 ——")
    print(Path("events.json").read_text(encoding="utf-8"))

运行后你会看到模型自主完成"查两个城市 → 比较 → 调 add_event"的完整链条，并给出带依据的最终答复。这个 80 行脚本就是 Agent 的完整骨架——所有框架做的事情，都在这个循环之上加工程化。

六、框架选择：什么时候别自研

自己写的循环适合学习和极简场景；真实项目里，工具数量、状态复杂度、并发、可观测性会迅速超出手写循环的舒适区。常见框架对比：

框架	核心抽象	优点	代价	何时用
LangGraph	图 + 状态机	流程可控、支持分支/循环/并行、可检查点	概念多、上手曲线陡	复杂多步流程、需要精确控制与断点续跑
OpenAI Agents SDK	Agent + Handoff（交接）	官方维护、函数调用开箱即用、代码量少	绑定 OpenAI 生态	快速交付、单 Agent 为主
Claude Agent SDK	Agent + 工具 + computer use	Anthropic 官方、工具/沙箱体验好	绑定 Claude 生态	深度用 Claude、需要浏览器/电脑操作
自研（本文）	while 循环	零依赖、完全透明、可控	缺工程能力（状态/重试/观测）	学习、工具 <5 个、流程固定

选择判断一句话：工具少于 5 个、流程固定 → 自研够用；要做产品、要并行与审计 → 上框架。无论用哪个框架，"工具 schema + 循环 + 记忆"的心智模型是通用的，框架只是换个实现。模型选型与成本参考见模型与榜单速查。

七、工程要点：从 demo 到能上线

手写循环跑通后，下面这些问题会在真实场景立刻冒出来：

1. 工具返回截断

工具可能返回超长文本（查数据库、读文件）。先把工具返回截断再喂回，否则上下文被撑爆、token 成本飙升：

def safe_tool_result(result: str, max_len: int = 4000) -> str:
    if len(result) > max_len:
        return result[:max_len] + f"\n……（已截断，原文 {len(result)} 字符）"
    return result

2. 错误重试与容错

工具可能抛异常（API 超时、参数非法）。把异常转成文本喂回模型，让它自己修正，而不是让整个循环崩溃：

try:
    result = dispatch(call.function.name, call.function.arguments)
except Exception as e:          # 用 isinstance 细分异常类型更严谨
    result = f"工具执行失败：{type(e).__name__}: {e}"
messages.append({"role": "tool", "tool_call_id": call.id, "content": safe_tool_result(result)})

3. 最大步数与死循环检测

max_steps 必须有，且要检测重复动作（同一工具 + 同一参数连续 N 次），直接判定失败。死循环是 Agent 最常见的"事故"形态。

4. 成本控制

每一步循环都是一次 LLM 调用。控制手段：用便宜的小模型做规划/简单工具调用、加语义缓存（相似问题命中缓存）、压缩历史消息。这些优化见推理优化与量化和部署与推理优化实战。

5. 权限与安全

工具是 Agent 对外部世界施加影响的唯一通道，也是最危险的地方：

风险	场景	对策
提示注入	工具返回的内容（如网页抓取结果）里藏指令，诱导 Agent 执行恶意操作	工具结果按"数据"而非"指令"处理；对敏感操作做二次确认；上下文隔离
越权工具调用	模型调用了一个不该调的工具（如删除操作）	最小权限原则：每个工具单独授权、敏感操作需用户确认
沙箱执行	让 Agent 执行生成代码	在沙箱/容器内运行，禁止访问敏感路径
参数校验	工具参数来自模型，可能非法	dispatch 层做类型与取值范围校验

安全边界与治理框架详见AI 安全与治理。

两条红线

1. 永远不要把"删除""转账"这类不可逆工具直接交给 Agent 自动调用，必须加人工确认环节。
2. 工具输出里混入的外部文本（网页、邮件、文档）默认不可信——它们可能携带攻击性指令。提示注入是对 Agent 最现实的攻击面。

八、评估 Agent 与常见坑

评估

Agent 是"过程 + 结果"都要评。最实用的四个指标：

指标	度量什么	怎么测
任务成功率	端到端有没有完成任务	一批真实任务，人工判定成功/失败
工具调用正确率	该调的工具调对没、参数对不对	对比"预期工具调用序列"与实际的差异
步数效率	用了多少步完成	统计平均步数；异常值往往对应"绕弯"
成本	每次任务多少 token	记录 tokens 用量，关联任务类型

建议：先建 20~50 条任务的评估集，再改循环逻辑，每次改动跑一遍，别靠"感觉变好了"。评估管道的搭建见搭建一套 LLM 评估，指标理论见LLM 评估与基准。

常见坑

坑	症状	根因	对策
循环失控	无限调用工具、token 烧穿	没有步数上限、模型反复试错	max_steps + 重复动作检测 + 超时
幻觉工具参数	调用了不存在的城市/参数类型错	schema 描述不清、模型自由发挥	加强参数描述、dispatch 层校验、必填参数缺失时重试
上下文膨胀	越到后面越慢越贵、答非所问	历史消息全量堆叠	截断/压缩/向量记忆按需召回
工具结果"污染"	行为突然异常	工具返回里混入指令性文本	结果截断、标注为数据、隔离上下文
过度调用工具	一个简单的"你好"也调工具	schema 触发条件写得太宽	description 写清"何时不用调"
静默失败	任务"看起来完成"实则没执行	工具返回了错误但被当作成功	工具返回结构化状态（ok/error），循环里显式检查

更完整的 Agent 反模式清单见常见陷阱与反模式。

九、进阶：多 Agent 协作

单 Agent 有容量和职责边界，复杂任务让多个各司其职的 Agent 协作。三种主流编排模式：

模式	结构	适用	例子
编排者-执行者（orchestrator-worker）	一个主 Agent 拆任务、分派给专业子 Agent	任务可分解、子任务类型固定	写报告 = 检索 Agent + 写作 Agent + 排版 Agent
流水线（pipeline）	前一个 Agent 的输出是后一个的输入	任务天然有先后顺序	调研 → 分析 → 总结
辩论/评审（debate）	多个 Agent 各自给答案再互相评审	高正确性要求	代码评审、事实核查

工程上的配套：子 Agent 之间只通过"消息/结果"交互（不要共享可变状态）、给每个子 Agent 独立上下文隔离、主 Agent 负责汇总与冲突裁决。多 Agent 产品的完整解剖见Manus 与 Agent 应用；决定协作结构的依据，其实回到了Agent 核心概念里对"任务复杂度与 Agent 形态"的讨论。

收尾判断

一套"合格"的 Agent：能稳定完成任务、有步数与成本护栏、工具不可逆操作有人工确认、并且有一份评估集量化"没变坏"。做到这四点，它已经是一个可以谈工程化的系统，而不是一个 demo。术语不懂查术语表，全景看什么是 AI 热门概念。

十、延伸阅读

• Agent 核心概念 —— 本文的完整理论版：定义、分类、ReAct、记忆、规划与多 Agent
• Manus 与 Agent 应用 —— 生产级 Agent 产品的完整解剖
• 提示词工程 —— 工具描述、系统提示、few-shot 的写法方法论
• 向量数据库与语义检索 —— 向量记忆的实现原理与选型
• 从零搭建 RAG 应用 —— 把"检索"变成一个 Agent 工具，即 Agentic RAG
• LLM 评估与基准 —— Agent 评估指标的理论基础
• 搭建一套 LLM 评估 —— 把第八节落地成可运行的评估管道
• 推理优化与量化 —— 循环成本控制的原理（缓存、量化、流式）
• AI 安全与治理 —— 提示注入、权限与对齐的完整框架
• 常见陷阱与反模式 —— 第八节的 25 项完整版
• 模型与榜单速查 —— 工具调用能力强弱的模型选型数据
• 总体架构解剖 —— 从本文的小 Agent 放大到生产系统

参考资料

• Yao et al., ReAct: Synergizing Reasoning and Acting in Language Models（arXiv 2210.03629） —— ReAct 范式的原始论文，Agent 循环的源头
• OpenAI 官方文档：Function calling —— 工具调用接口的官方教程与最佳实践
• OpenAI Agents SDK —— 官方轻量 Agent 框架
• LangGraph 官方文档 —— 图状态机 Agent 框架
• Anthropic：Claude Agent SDK 与 Tool use —— Claude 生态的 Agent 与工具使用指南
• Anthropic 工程博客：Building effective agents —— 何时该用/不该用 Agent 的工程判断
• Ollama 官网 —— 本地 LLM（含工具调用）的安装与模型