Agent SDK:从一个能跑的 Agent 开始
先把每一步的代码复制到自己的练习文件里运行,再回头改参数、换问题、故意制造错误。示例沿用前两篇的服务台助手:用户会询问公司制度,也可能查询自己的工单。工具只读。
文中以 OpenAI Agents SDK(Python) 为主。它是 OpenAI 的编排层,不是另一种模型 API:底层仍然要调用模型,工具仍然由你的 Python 代码执行。Anthropic 的 Claude Agent SDK 是另一套产品,类名、运行器和事件模型不能直接照抄;如果切换到 Claude,应以 Anthropic 当前 SDK 文档为准。
这次要写到哪里
你会按这个顺序完成小目标:
- 安装 SDK,让一个 Agent 回答一句话。
- 注册
search_policy只读工具,查询本地制度资料。 - 观察事件,分清模型输出、工具调用和工具结果。
- 用 session 保存两轮对话。
- 把工单问题交给专门的 Agent。
- 在输入和工具边界加 guardrail。
- 开启流式事件和 tracing,记录过程但不记录隐藏推理。
前一步没跑通时,先不要把后面的概念一起粘上去。
第 1 步:准备练习环境
先建目录和虚拟环境。练习代码放在文档仓库之外,方便你随意试错:
mkdir agent-sdk-lab
cd agent-sdk-lab
python -m venv .venv
source .venv/bin/activate # Windows PowerShell:.venv\Scripts\Activate.ps1
python -m pip install -U pip
python -m pip install "openai-agents"
python -m pip show openai-agents准备环境变量:
OPENAI_API_KEY=替换成你的密钥
OPENAI_MODEL=你有权限使用的模型 IDSDK 默认读取 OPENAI_API_KEY。模型名建议显式传入,避免账号默认值变化。若使用代理或兼容 OpenAI 协议的服务,base URL 和模型能力按该服务文档配置,不能把兼容服务的行为当成 OpenAI 官方行为。
运行观察:pip show 找不到包通常是虚拟环境没有激活;缺少 API key 时,检查当前终端是否真的能读到变量。若依赖 .env,请明确使用 python-dotenv 加载,不要假设所有启动方式都会自动加载。
第 2 步:先让 Agent 说一句话
新建 hello_agent.py:
import asyncio
import os
from agents import Agent, Runner
agent = Agent(
name="ServiceDesk",
instructions="你是公司 IT 服务台助手。回答简短、具体;没有资料时明确说不知道。",
model=os.environ.get("OPENAI_MODEL"),
)
async def main() -> None:
result = await Runner.run(agent, "你好,请告诉我你能帮我做什么?")
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())运行:
python hello_agent.pyAgent 描述助手的身份、指令和模型,Runner 负责实际运行。 Runner.run 是异步方法,所以入口使用 asyncio.run。在同步脚本里可试 Runner.run_sync,但不要在已经运行事件循环的 Web 服务里嵌套 asyncio.run。
应该看到:一段自我介绍,此时没有工具调用和 handoff,final_output 只是最终文本。
动手改一改:把 instructions 改成“只回答天气”,再问一个工单问题。模型可能仍会泛泛作答,这说明 instructions 是行为约束,不是权限系统;权限要放在代码和工具边界里。
第 3 步:注册一个只读工具
先写一个从字典中查询制度资料的函数:
from agents import function_tool
POLICIES = {
"远程办公": "每周最多申请三天远程办公,需提前一个工作日提交。",
"密码": "公司密码至少 12 位,不能与最近三次密码相同。",
}
@function_tool
def search_policy(keyword: str) -> str:
"""按关键词查找公司制度。只读,不会修改制度。"""
keyword = keyword.strip()
for title, content in POLICIES.items():
if keyword in title or keyword in content:
return f"制度:{title}\n内容:{content}"
return "没有找到匹配的制度资料。"把工具放进 Agent:
agent = Agent(
name="ServiceDesk",
instructions=(
"你是公司 IT 服务台助手。涉及公司制度时先调用 search_policy;"
"资料没有命中就直说,不要凭空编制度。"
),
tools=[search_policy],
model=os.environ.get("OPENAI_MODEL"),
)@function_tool 会根据函数签名和 docstring 生成工具 schema。参数类型、描述和不适用场景都应写清楚,但 schema 不是安全边界,代码仍要校验长度、枚举和权限。
运行:
result = await Runner.run(agent, "公司密码有什么要求?")
print(result.final_output)运行观察:模型先提出调用 search_policy,SDK 执行 Python 函数,再把结果交回模型,最后生成回答。只看 final_output 时,中间过程被 Runner 隐藏了。
常见坑:
- 返回值必须能序列化,不要返回连接、文件句柄等运行时对象。
- “查询所有信息”这种工具说明太宽,容易导致误调用。
- 工具参数来自模型,不能因为 schema 校验过就跳过业务校验。
给工具增加 department: str 参数,只允许 IT 和 HR,再观察缺少部门时模型是否追问。
第 4 步:看见运行结果和工具调用
先查看结果对象里适合业务使用的字段:
result = Runner.run_sync(agent, "我想知道远程办公制度。")
print("最终回答:", result.final_output)
print("最后一个 Agent:", result.last_agent.name)SDK 版本升级时,结果辅助属性可能变化。调试可以先查看公开属性:
print(type(result))
print([name for name in dir(result) if not name.startswith("_")])真正观察过程应使用流式事件。你要关心的是:
- 模型响应:文本增量或一次完整响应。
- 工具调用:工具名、调用 ID、参数片段。
- 工具结果:函数成功、业务错误或异常。
- handoff、guardrail、异常和结束事件。
不要把模型未公开的隐藏推理当日志。记录“调用了哪个工具、参数是否通过校验、耗时和结果摘要”就够了,不要试图打印完整 Thought。
第 5 步:用 session 保留两轮对话
不传 session 时,两次运行互相独立。用本地 SQLite session 试试:
from agents import SQLiteSession
session = SQLiteSession("service-desk-demo")
async def main() -> None:
first = await Runner.run(
agent, "我想查一下远程办公制度。", session=session
)
print(first.final_output)
second = await Runner.run(
agent, "那每周最多几天?", session=session
)
print(second.final_output)第二轮能理解“那”指什么,是因为 session 保存了上下文。它不是永久记忆:仍要决定何时清理、摘要和隔离会话,不能把密钥或完整个人档案塞进历史。
常见坑:不要在并发 Web 请求中让所有用户共享一个全局 session,否则会话会串线。session 的键至少要包含租户和用户边界;涉及权限的资料也不能只依赖历史消息。
第 6 步:增加工单 Agent 和 handoff
先写一个只读工单工具:
@function_tool
def get_ticket(ticket_id: str) -> str:
"""查询工单状态。只接受形如 INC-1001 的工单号,不会修改工单。"""
tickets = {
"INC-1001": "状态:处理中;负责人:IT 支持组;预计今天 18:00 更新。",
}
return tickets.get(ticket_id.upper(), "没有找到这个工单,请确认工单号。")定义专门 Agent,再让分流 Agent 交接:
ticket_agent = Agent(
name="TicketAgent",
instructions=(
"你只负责查询工单。需要工单号时先询问,"
"调用 get_ticket 后只根据返回内容回答。"
),
tools=[get_ticket],
model=os.environ.get("OPENAI_MODEL"),
)
triage_agent = Agent(
name="TriageAgent",
instructions=(
"你是服务台分流助手。制度问题由你处理;"
"工单状态、负责人或进度问题交给 TicketAgent。"
),
tools=[search_policy],
handoffs=[ticket_agent],
model=os.environ.get("OPENAI_MODEL"),
)运行:
result = await Runner.run(triage_agent, "帮我查一下 INC-1001 到哪一步了?")
print(result.final_output)handoffs=[ticket_agent] 让下游 Agent 接管后续对话,它不是普通函数工具。每个 Agent 只拿完成职责所需的最小工具集;生产代码还要设置最大交接深度和总预算。
观察:轨迹中会出现 TriageAgent 到 TicketAgent 的交接,然后由 TicketAgent 调用 get_ticket。
练习:分别问制度问题、缺少工单号的问题和完全无关的问题,观察谁回答、谁追问。再移除 handoff,感受“模型建议”和“代码授予能力”的差别。
第 7 步:给输入加一个最小 guardrail
Guardrail 是可执行拦截逻辑,不是 prompt 里的一句“请安全回答”。例如服务台不应处理导出员工隐私数据的请求:
from agents import GuardrailFunctionOutput, RunContextWrapper, input_guardrail
@input_guardrail
async def block_sensitive_request(
ctx: RunContextWrapper[None],
agent: Agent,
input: str,
) -> GuardrailFunctionOutput:
blocked_words = ("工资表", "全员联系方式", "导出所有员工")
blocked = any(word in input for word in blocked_words)
return GuardrailFunctionOutput(
output_info={"blocked_words": blocked_words if blocked else []},
tripwire_triggered=blocked,
)挂到分流 Agent:
triage_agent = Agent(
name="TriageAgent",
instructions="你是服务台分流助手,只处理制度和工单查询。",
tools=[search_policy],
handoffs=[ticket_agent],
input_guardrails=[block_sensitive_request],
model=os.environ.get("OPENAI_MODEL"),
)捕获护栏异常:
from agents import InputGuardrailTripwireTriggered
try:
result = await Runner.run(triage_agent, "请导出所有员工的工资表")
except InputGuardrailTripwireTriggered:
print("这个请求超出服务台助手的处理范围。")应该看到:请求在工具阶段前被拦截。可以在 search_policy 开头临时打印,确认它没有执行。
输入护栏不能替代工具内部的身份、租户、资源范围和审批校验。副作用工具必须在真正执行前再做一次代码级检查;批准记录、操作者身份、幂等键和审计落库属于业务代码。
第 8 步:副作用工具先做确认
本教程只注册只读工具。若要增加“创建工单”,先把动作拆成准备和提交两段:
@function_tool
def prepare_ticket(title: str, detail: str) -> str:
"""生成待确认的工单草稿,不会提交。"""
return f"待提交:{title}\n详情:{detail}"用户明确确认后,业务代码再调用真正的提交函数。不要只在 prompt 里写“请确认”,也不要让模型自行判断用户是否确认。
第 9 步:用流式事件看运行过程
事件类型可能随 SDK 版本演进,先打印类型和数据类型:
async def stream_once() -> None:
result = Runner.run_streamed(
triage_agent,
"查询 INC-1001 的状态",
)
async for event in result.stream_events():
print("EVENT", event.type, type(event.data).__name__)如果要把文字增量显示到 UI,应针对当前版本文档中的文本增量事件做判断;不要把所有事件的 repr 原样发给前端。工具参数可能分成多个片段,必须累积并通过 JSON/schema 校验后才能执行。
运行观察:一次工单查询通常会出现模型响应、handoff、工具调用、工具结果和后续模型响应。不同模型和 SDK 版本的事件数量可能不同,不要把“必须正好出现 N 个事件”写成测试断言。
第 10 步:开启 tracing,不记录隐藏推理
Agents SDK 自带 tracing。先给运行包一个业务上可读的名称:
from agents import trace
async def traced_run() -> None:
with trace("service-desk-query"):
result = await Runner.run(
triage_agent,
"公司密码需要设置多长?",
)
print(result.final_output)配置正确时,可在对应控制台查看 Agent、模型、工具、handoff、guardrail、异常和耗时。trace 用来回答“哪一步慢、哪一步失败、走了哪条路径”,不是窥探模型隐藏推理的窗口。
建议记录:run_id、业务请求 ID、Agent 名称、工具名、脱敏参数、开始和结束时间、token usage、错误类型和重试次数。工单号、员工信息和 prompt 按规则脱敏,密钥永远不要进 trace。若接入自有观测平台,最好写适配层,不要把 SDK 内部对象当成长期稳定格式。
第 11 步:在运行层设置错误和预算
SDK 会帮你跑循环,但边界仍要由业务设置:
from agents import MaxTurnsExceeded
try:
result = await Runner.run(
triage_agent,
user_input,
max_turns=8,
)
except MaxTurnsExceeded:
print("本次任务步骤过多,已停止,请换一种问法。")
except Exception as exc:
# 生产服务记录 request_id 和内部错误码,不把堆栈返回给用户。
print(f"服务暂时不可用:{type(exc).__name__}")max_turns 是防止模型和工具来回循环的硬闸门,不是质量保证。工具函数仍要有超时、重试和幂等策略;只读查询可以有限重试,创建工单、发邮件等副作用动作不能因网络重试而重复执行。把 token、工具调用次数、总耗时和失败出口写入日志或评测结果,才能比较改动是否真的变好。
Agent SDK、Responses API、LangGraph 的边界
可以这样记:
模型与工具协议
└─ Responses API(请求、输入输出 item、工具调用、多轮续接)
└─ Agents SDK(Agent、Runner、handoff、guardrail、session、tracing)
└─ 你的业务代码(权限、审批、数据、部署和审计)Responses API 更接近底层接口。你自己决定 item 怎么组织、工具结果何时回传、循环如何结束;需要完全掌控请求格式或使用 SDK 尚未封装的新能力时,直接用它。
Agents SDK 在其上提供 Agent、Runner、工具装饰器、handoff、guardrail、session 和 tracing。它适合快速写一个或几个 Agent,但不会替你设计复杂业务状态,也不会自动授予数据库写权限。
LangGraph 关注显式状态图:节点、条件边、循环、checkpoint 和人工暂停。它适合“检索 → 起草 → 审核,不通过就回检索”这类流程。可以在图节点里调用 Agents SDK,但不要把两者当成同一层:SDK 管 Agent 内部运行,图管跨步骤流程。
一个简单判断:只有一个模型、两三个只读工具时,先用 Responses API 或 Agents SDK;出现多个分支、恢复点、审批和长流程状态时,再考虑 LangGraph。无论选哪种方式,都先把 raw API 的“模型决定 → 代码执行工具 → 回传结果”边界讲清楚。
Claude Agent SDK 不是别名
Anthropic 的 Claude Agent SDK 面向 Claude 生态,通常围绕 Claude 客户端、工具、MCP 和自身运行事件提供接口。它与 OpenAI Agents SDK 的 Agent、Runner、function_tool 不兼容。迁移时要重新核对:
- 模型客户端和认证方式。
- 工具 schema 与工具结果格式。
- 流式事件名称和增量数据结构。
- 会话、权限确认、子 Agent 和 tracing 的实现。
- 费用、上下文窗口和默认行为。
更稳的迁移方式是固定同一组问题、工具返回和验收标准,再分别实现两套适配层,而不是直接替换 import。
练习:在别处继续写
- 给
search_policy增加来源和更新时间,要求最终回答带上这两个字段。 - 让分流 Agent 在缺少工单号时先追问,而不是立即 handoff。
- 给
get_ticket增加可恢复的超时错误,观察重试并设置硬上限。 - 将只读查询和创建工单分成两个 Agent,给创建路径加人工确认和幂等键。
- 用 5 个固定问题比较 Responses API、Agents SDK 和 LangGraph 版本的工具调用次数、耗时、错误出口和日志可读性。
每次只改一个变量再运行。例如先改工具描述,再改路由指令,这样才能知道结果变化来自哪里。
小结
SDK 收拢了 Agent 定义、工具注册、运行循环、handoff、guardrail、session 和 tracing;权限、数据质量、审批、超时、幂等、脱敏和评测仍是你的责任。遇到行为不符合预期时,先回到三个问题:模型收到了什么,代码真正执行了什么,运行层在哪里决定结束。