裸 API 手写 Agent

我们从空目录开始，每次只写一个小部件，运行一下，再继续下一步。你可以把代码片段复制到自己的练习目录中，也可以故意改错，观察模型和程序分别会发生什么。

贯穿案例是一个企业 IT 服务台助手：员工问“VPN 怎么连”时查内部知识库；员工问“我有几个未关闭工单”时查工单系统；员工要求“替我关闭工单”时，助手只能先说明需要人工确认。知识库和工单服务先用本地 Mock，等流程跑通后再替换成真实 API。

先画出边界

裸 API Agent 不是“让模型自己执行代码”。它的职责分界很简单：

用户问题
   ↓
程序把消息和工具说明发给模型
   ↓
模型选择：直接回答，或请求某个工具
   ↓
程序校验参数并执行工具
   ↓
程序把工具结果放回消息，再请求模型
   ↓
模型给出最终答复

模型只能提出“请调用 search_kb”，真正读文件、查数据库、关闭工单的代码必须由你控制。后面的每一步都围绕这个边界展开。

第 1 步：准备一个空练习目录

这一小步要完成什么

先把练习和仓库其他代码隔离，建立三个文件：配置、工具数据、主程序。文件名可以按自己的习惯改，但后续片段会沿用下面的名字。

raw-agent-demo/
├── .env
├── config.py
├── tools.py
└── main.py

在目录中安装 SDK：

python -m venv .venv
source .venv/bin/activate       # Windows 使用 .venv\\Scripts\\activate
pip install "openai>=1.40,<2" python-dotenv

这里锁定主版本范围，是为了避免 SDK 自动升级后字段突然变化。你也可以使用兼容 OpenAI Chat Completions 格式的其他服务商，但要先确认它支持 tools 和 tool_calls。

建立配置文件

在 .env 中先写占位配置：

MODEL_PROVIDER=real
MODEL_NAME=gpt-4o-mini
OPENAI_API_KEY=

这里的 real 表示使用真实模型服务；先把自己的 API Key 填入再运行第 2 步。不要把 .env 提交到 Git。本文中的 Mock 只用于知识库和工单工具，模型选择仍由真实 API 完成。

在 config.py 读取配置：

import os
from dotenv import load_dotenv

load_dotenv()

MODEL_PROVIDER = os.getenv("MODEL_PROVIDER", "real")
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

关键点是给每个配置提供默认值。没有 API Key 时仍可先运行 tools.py 的 Mock 函数，验证参数和权限边界；要走完整模型循环，仍需真实模型或你自己实现的 FakeResponse。

运行观察

暂时在 main.py 写三行：

from config import MODEL_PROVIDER, MODEL_NAME

print(MODEL_PROVIDER, MODEL_NAME)

执行 python main.py，应该看到 real gpt-4o-mini。如果输出为空，先检查 .env 是否和 main.py 在同一目录。

动手改一改

把 MODEL_NAME 改成你有权限使用的模型，再运行一次。注意：此时还没有发起网络请求，改错模型名也不会马上报错。

第 2 步：先让程序能直接回答

这一小步要完成什么

在接工具之前，先验证“用户消息 -> 模型 -> 文本回答”这条最短路径。将 main.py 改成下面的增量代码：

from openai import OpenAI
from config import MODEL_NAME, OPENAI_API_KEY

client = OpenAI(api_key=OPENAI_API_KEY)

messages = [
    {"role": "system", "content": "你是公司 IT 服务台助手，回答要短，不能编造公司制度。"},
    {"role": "user", "content": "请用一句话说明你能做什么。"},
]

response = client.chat.completions.create(
    model=MODEL_NAME,
    messages=messages,
    temperature=0,
)

print(response.choices[0].message.content)

messages 是模型看到的对话历史；temperature=0 不是“绝对不会变化”，只是让练习结果更稳定。先不要加入 tools，这样可以单独确认密钥、网络和模型名都没问题。

运行观察

真实模式下执行 python main.py，应该得到一段说明性文字。记录两项信息：

print("finish_reason:", response.choices[0].finish_reason)
print("usage:", response.usage)

正常直接回答时，finish_reason 通常是 stop。如果是 length，说明输出被长度限制截断；如果提示认证失败，先回到 .env 检查密钥，不要急着改 Prompt。

Mock 与真实 API 的边界

本教程的 Mock 是工具层：tools.py 里的知识库和工单数据不访问公司系统，适合练习参数校验、超时和错误分支；它不能证明模型真的会正确选择工具。模型选择、参数生成和流式分片必须用真实 API 验证。没有 API Key 时，可以先单独运行工具函数，或自行写一个返回固定 tool_calls 的 FakeResponse，再接入同一个循环。

第 3 步：给模型两份“接口文档”

这一小步要完成什么

服务台助手需要两个只读工具：查知识库、查当前用户的工单。先只描述接口，不急着实现函数。

在 main.py 的 messages 前加入：

tools = [
    {
        "type": "function",
        "function": {
            "name": "search_kb",
            "description": "搜索公司 IT 知识库。只用于回答内部制度、软件配置和故障排查问题。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "要搜索的关键词或问题"},
                    "top_k": {"type": "integer", "description": "最多返回几条，范围 1 到 5"},
                },
                "required": ["query"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "list_my_tickets",
            "description": "查询当前登录员工的未关闭工单。不能查询其他员工。",
            "parameters": {
                "type": "object",
                "properties": {},
                "additionalProperties": False,
            },
        },
    },
]

工具描述要像接口文档，而不是“万能助手”。不能查询其他员工 这类边界应该写在描述中，但不能只依赖描述来做权限控制，真正的身份校验仍要在 Python 工具函数里完成。

把请求改成：

response = client.chat.completions.create(
    model=MODEL_NAME,
    messages=messages,
    tools=tools,
    tool_choice="auto",
    temperature=0,
)

运行观察

把用户问题换成“VPN 连不上，知识库里有排查步骤吗？”。模型可能返回 message.tool_calls，此时 message.content 可能是 None，这是正常的。把它打印出来：

message = response.choices[0].message
print("content:", message.content)
print("tool_calls:", message.tool_calls)

动手改一改

分别测试三个问题：普通寒暄、VPN 排查、查询我的工单。观察什么时候有 tool_calls。如果三个问题都调用同一个工具，先收窄 description，不要立刻修改循环。

第 4 步：先用 Mock 工具数据跑通执行边界

这一小步要完成什么

在 tools.py 写两个本地函数。它们故意返回短数据，避免把大段文档塞进上下文：

KB = {
    "vpn": "公司 VPN：先打开 SecureLink，选择‘企业入口’，再使用工号登录；连续失败 3 次请联系服务台。",
    "邮箱": "邮箱客户端服务器地址为 mail.example.com，端口 993；密码修改后需重新登录。",
}

def search_kb(query: str, top_k: int = 3) -> dict:
    query_lower = query.lower()
    hits = [text for key, text in KB.items() if key in query_lower]
    return {"hits": hits[:max(1, min(top_k, 5))], "source": "mock-kb"}


def list_my_tickets(user_id: str) -> dict:
    if not user_id:
        raise ValueError("missing user identity")
    return {
        "user_id": user_id,
        "tickets": [{"id": "IT-1042", "status": "open", "title": "VPN 登录失败"}],
    }

这里的 user_id 不由模型提供，而是由服务端会话注入。即使工具 schema 没有这个参数，执行函数也可以要求它。这样能避免模型伪造身份。

运行观察

在 main.py 临时加几行，确认工具自身能工作：

from tools import search_kb, list_my_tickets

print(search_kb("VPN", 2))
print(list_my_tickets("u-1001"))

先执行一次，再删掉这两行。工具测试通过后，模型循环出错时才容易判断是“模型没选对”还是“工具本身坏了”。

第 5 步：识别一次 tool call

这一小步要完成什么

模型返回的工具请求不是 Python 函数，而是一段带 ID 的结构。先只读取它，不执行：

import json

message = response.choices[0].message
if message.tool_calls:
    for call in message.tool_calls:
        name = call.function.name
        arguments_text = call.function.arguments
        arguments = json.loads(arguments_text)
        print("call_id:", call.id)
        print("name:", name)
        print("arguments:", arguments)

arguments 在 SDK 中是 JSON 字符串，必须 json.loads 后再做字段校验。call.id 稍后要原样放进 tool_call_id，否则模型无法把工具结果和请求对应起来。

运行观察

输入“我有几个未关闭工单？”时，理想输出类似：

call_id: call_...
name: list_my_tickets
arguments: {}

模型可能把 top_k 传成字符串，也可能给出额外字段。不要假设它永远遵守 schema，执行前要在代码侧拒绝不合法参数。

动手改一改

故意把 arguments_text 改成 "{bad json}"，确认 json.loads 会抛错。下一步我们会把这个异常变成模型能理解的结构化工具结果。

第 6 步：用白名单分派工具

这一小步要完成什么

把工具名映射到函数，禁止模型调用任意 Python 名称。在 main.py 加入：

from tools import search_kb, list_my_tickets

TOOL_HANDLERS = {
    "search_kb": search_kb,
    "list_my_tickets": list_my_tickets,
}


def dispatch_tool(name: str, arguments: dict, user_id: str) -> dict:
    handler = TOOL_HANDLERS.get(name)
    if handler is None:
        return {"ok": False, "error": "unknown_tool"}
    if name == "search_kb":
        query = arguments.get("query")
        top_k = arguments.get("top_k", 3)
        if not isinstance(query, str) or not query.strip():
            return {"ok": False, "error": "invalid_query"}
        if not isinstance(top_k, int) or not 1 <= top_k <= 5:
            return {"ok": False, "error": "invalid_top_k"}
        return {"ok": True, "data": handler(query=query, top_k=top_k)}
    if name == "list_my_tickets":
        if arguments:
            return {"ok": False, "error": "unexpected_arguments"}
        return {"ok": True, "data": handler(user_id)}
    return {"ok": False, "error": "unreachable"}

TOOL_HANDLERS 是第一道白名单；list_my_tickets 的身份参数由程序补上，是第二道边界。以后加入“关闭工单”时，要在这里额外检查权限和人工批准，不能因为模型说“用户已经同意”就直接执行。

运行观察

手动调用：

print(dispatch_tool("search_kb", {"query": "VPN"}, "u-1001"))
print(dispatch_tool("delete_all", {}, "u-1001"))

第二行应该返回 unknown_tool，而不是尝试执行某个同名函数。

第 7 步：把工具结果放回 messages

这一小步要完成什么

工具执行完后，必须追加两类消息：模型刚才的 assistant 消息，以及每个调用对应的 tool 消息。assistant 消息只追加一次，即使这一轮有多个工具调用。先写追加单个工具结果的小函数：

def append_tool_result(messages, call, result):
    messages.append({
        "role": "tool",
        "tool_call_id": call.id,
        "content": json.dumps(result, ensure_ascii=False),
    })

不要只追加 assistant_message.content。工具调用信息在 tool_calls 字段里，丢掉它，下一轮模型就不知道自己刚才请求过什么。

现在把请求封装成函数，便于下一步循环复用：

def ask_model(messages):
    return client.chat.completions.create(
        model=MODEL_NAME,
        messages=messages,
        tools=tools,
        tool_choice="auto",
        temperature=0,
    )

运行观察

在还没有写循环前，手动执行一次：收到 tool call 后调用 dispatch_tool，再调用 append_tool_result，打印 messages。你应该能看到消息顺序是 system -> user -> assistant(tool_calls) -> tool。

第 8 步：写出最小 tool loop

这一小步要完成什么

现在把前面的部件串起来。先写一个“最多三轮”的版本，代码保持短小：

def run_agent(user_text: str, user_id: str, max_turns: int = 3) -> str:
    messages = [
        {"role": "system", "content": "你是 IT 服务台助手，只依据工具返回的事实回答。"},
        {"role": "user", "content": user_text},
    ]

    for turn in range(1, max_turns + 1):
        response = ask_model(messages)
        message = response.choices[0].message
        print(f"turn={turn}, finish_reason={response.choices[0].finish_reason}")

        if not message.tool_calls:
            return message.content or "模型没有返回文字。"

        messages.append(message.model_dump(exclude_none=True))
        for call in message.tool_calls:
            try:
                arguments = json.loads(call.function.arguments)
                result = dispatch_tool(call.function.name, arguments, user_id)
            except (json.JSONDecodeError, TypeError, ValueError) as exc:
                result = {"ok": False, "error": "invalid_arguments", "message": str(exc)}
            append_tool_result(messages, call, result)

    return "这次请求步骤太多，已停止继续调用工具。"

这里最重要的不是代码行数，而是循环顺序：模型决定、程序执行、结果回传、再次请求。max_turns 是硬限制，不能省略；否则模型或工具出错时可能无限调用。

在文件末尾加入入口：

if __name__ == "__main__":
    question = input("你想咨询什么？ ")
    print(run_agent(question, user_id="u-1001"))

运行观察

至少测试三条输入：

你好                         -> 通常一轮直接回答
VPN 连不上怎么办？           -> 先 search_kb，再回答
我有几个未关闭工单？          -> 先 list_my_tickets，再回答

如果模型返回多个 tool_calls，内层 for 会全部执行；不要只取 message.tool_calls[0]。

动手改一改

把 max_turns 改成 1，再次询问 VPN。你会看到程序执行了工具却没有机会让模型总结结果。这个实验能直观看到“最大轮数太小”和“无限循环”之间需要平衡。

第 9 步：给工具加超时和错误回传

这一小步要完成什么

Mock 函数几乎不会超时，真实知识库和工单 API 会。先把工具调用包在一个安全函数里：

import time

def safe_dispatch(name: str, arguments: dict, user_id: str, timeout_s: float = 3.0) -> dict:
    started = time.monotonic()
    try:
        result = dispatch_tool(name, arguments, user_id)
        elapsed = time.monotonic() - started
        if elapsed > timeout_s:
            return {"ok": False, "error": "tool_timeout", "message": "工具响应超过 3 秒"}
        return result
    except PermissionError:
        return {"ok": False, "error": "permission_denied", "message": "当前身份没有执行权限"}
    except Exception:
        return {"ok": False, "error": "tool_failed", "message": "工具暂时不可用，请稍后重试"}

示例使用计时判断超时，真实项目应使用 HTTP 客户端自身的连接和读取超时；不要让请求一直阻塞。异常信息返回给模型时要使用稳定的错误码，不能把数据库连接串、堆栈和内部路径直接暴露给用户。

把第 8 步中的 dispatch_tool(...) 换成：

result = safe_dispatch(call.function.name, arguments, user_id)

运行观察

在 search_kb 中临时加入 time.sleep(4)，再运行 VPN 问题。你应看到 tool_timeout 被回传，模型可能会向用户说明暂时无法查询。测试完成后删除这行延迟代码。

工具错误回传并不等于“让模型绕过权限”。权限、金额、资源范围等检查必须在工具边界完成，模型只能根据错误决定是否换一个只读路径或向用户求助。

第 10 步：记录可审计但不泄密的日志

这一小步要完成什么

调试时要知道哪一轮慢、调用了哪个工具，但不要打印完整 Prompt、令牌和个人数据。先写一个极简日志函数：

def log_event(run_id: str, event: str, **fields):
    safe_fields = {"run_id": run_id, "event": event, **fields}
    print(json.dumps(safe_fields, ensure_ascii=False))

在 run_agent 开始时生成 run_id，每次模型调用和工具调用分别记录：

import uuid

run_id = "run-" + uuid.uuid4().hex[:8]
log_event(run_id, "model_call", turn=turn)
log_event(run_id, "tool_call", turn=turn, tool=call.function.name, call_id=call.id)

生产日志还应加入耗时、finish_reason、输入输出 token 和错误码，但工具参数只记录经过脱敏的摘要。像邮箱、手机号、工号、Authorization header 都不能原样写入日志。

运行观察

用一次 VPN 查询和一次工单查询，对照日志回答三个问题：调用了几轮？哪一个工具最慢？有没有把用户身份直接打印出来？如果回答不了，说明日志字段还不够；如果日志包含完整个人信息，说明脱敏边界还没设好。

第 11 步：加入流式输出，但不要急着执行分片工具参数

这一小步要完成什么

文本回答可以边生成边显示。把 ask_model 改成一个流式函数：

def stream_answer(messages):
    return client.chat.completions.create(
        model=MODEL_NAME,
        messages=messages,
        tools=tools,
        tool_choice="auto",
        temperature=0,
        stream=True,
    )

消费文本分片时：

for chunk in stream_answer(messages):
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="", flush=True)
print()

工具调用的 name 和 arguments 也可能被拆成多个 chunk。练习阶段可以先把 arguments 字符串累积完，再 json.loads；在 JSON 完整、参数校验通过之前，绝不能执行工具。

运行观察

直接问“VPN 怎么连”时，你可能先看到少量文字，随后才出现工具调用相关分片。不同模型对工具调用的分片顺序不完全相同，不能假设“第一个 chunk 就有完整 arguments”。

如果你暂时只想实现稳定的工具循环，可以先使用非流式请求完成闭环，再单独给最终文本增加流式输出。这是降低调试难度的合理取舍。

第 12 步：把 Mock 工具换成真实服务

这一小步要完成什么

流程稳定后，再替换 tools.py 中的实现，main.py 的 tool loop 不需要改。以知识库 HTTP 接口为例：

import requests

def search_kb(query: str, top_k: int = 3) -> dict:
    response = requests.get(
        "https://kb.example.com/search",
        params={"q": query, "top_k": min(max(top_k, 1), 5)},
        timeout=(2, 3),
    )
    response.raise_for_status()
    data = response.json()
    return {"hits": data.get("hits", [])[:5], "source": "company-kb"}

真实 API 的认证信息从环境变量或密钥服务读取，不要写进代码。请求超时、状态码、响应字段和返回长度都要在工具函数内处理，让上层循环只看到稳定的成功或错误结构。

工单查询同理，但要把当前用户身份放在服务端会话或签名令牌中。绝不能让模型通过参数传入任意 user_id 来越权查询。

第 13 步：认识副作用工具的安全闸门

到这里我们只有只读工具。假设要增加“关闭工单”，先只设计一个不会真正执行的预览函数：

def preview_close_ticket(ticket_id: str, user_id: str) -> dict:
    return {
        "action": "close_ticket",
        "ticket_id": ticket_id,
        "requested_by": user_id,
        "requires_confirmation": True,
    }

在模型收到这个结果后，程序应暂停并向用户展示工单号、影响范围和确认按钮（或命令行确认）。只有明确确认、再次校验当前用户权限后，才调用真正的关闭接口。

不要把“用户说了关闭”当作确认凭证，也不要让模型生成 approved=true 就跳过人工环节。副作用工具还要考虑幂等键，防止网络重试造成重复操作。

常见卡点排查

模型一直不调用工具

先检查工具描述是否具体、用户问题是否真的需要外部事实、请求是否传入了 tools。可以临时设置 tool_choice="required" 验证模型是否能生成调用，但不要把它长期用于所有请求。

工具调用后模型报消息格式错误

确认 assistant 消息原样追加，tool 消息的 tool_call_id 与 call.id 完全一致，且每个 call 都有对应结果。不要只追加文本内容。

模型传了奇怪参数

把 JSON 解析、类型检查、范围检查放在 dispatch_tool 前面。模型输出是不可信输入，尤其是路径、SQL、金额和资源 ID。

请求成本突然升高

打印每轮 usage，检查是否把长篇工具结果或完整历史重复塞进 messages。工具结果先裁剪，只保留下一步真正需要的事实、来源和时间。

你应该亲手验证的四条路径

完成所有步骤后，用自己的代码分别验证：

1. 普通寒暄不调用工具，直接结束。
2. VPN 问题调用知识库，工具结果回传后再回答。
3. 工单查询使用服务端注入的用户身份，模型不能指定别人的身份。
4. 工具超时或参数错误时，程序不崩溃、不无限重试，并向用户给出可理解的说明。

当这四条路径都能解释清楚时，你已经真正掌握了裸 API Agent 的核心。LangGraph 或 Agent SDK 后续会替你组织节点、状态和追踪，但底层仍然是这条“模型决定 -> 程序执行 -> 结果回传”的循环。