用 OpenAI Agents SDK 给本地模型 Agent 加运行时护栏

Agent 能自己决定调用哪个工具、怎么使用工具返回的结果，这种灵活性正是它有用的原因，但也意味着执行过程中需要明确的边界。多数人首先想到的是在 instructions 里写清楚「不要泄露内部信息」，然而 instructions 只是一种软约束：模型没有义务每次都遵守，也无法保证它在长上下文、多轮工具调用之后仍然记得这条要求。护栏（guardrail）提供的是另一种控制方式——它在运行时设置检查点，对 Agent 的输入或输出做校验，并在流程继续之前决定接受、拒绝还是替换。本教程以 OpenAI Agents SDK 为例，演示如何给一个由本地模型驱动的客服 Agent 加上工具输出护栏，让敏感字段在回到模型之前就被剥离。

准备工作

运行环境与依赖

示例需要 Python 环境、Ollama 以及 OpenAI Agents SDK。Ollama 用于在本地托管模型，Agents SDK 负责 Agent 编排与护栏机制。安装 Ollama 后，拉取示例所用的模型：

ollama pull gemma4:e4b

然后安装 Agents SDK：

pip install openai-agents

Ollama 会暴露一个兼容 OpenAI 接口的本地端点，因此可以用 AsyncOpenAI 客户端连接它，再通过 OpenAIChatCompletionsModel 包装成 Agents SDK 可用的模型对象。这一步很关键：它说明护栏并不绑定某一家厂商的托管模型，本地托管的模型同样可以走这套机制。

from openai import AsyncOpenAI
from agents import (
    Agent,
    OpenAIChatCompletionsModel,
)

client = AsyncOpenAI(
    api_key="ollama",
    base_url="http://localhost:11434/v1",
)

model = OpenAIChatCompletionsModel(
    model="gemma4:e4b",
    openai_client=client,
)

这里的 api_key 只是占位值，本地端点通常不校验它；base_url 指向 Ollama 的 OpenAI 兼容接口。模型名称必须与 ollama pull 时使用的名称一致，否则请求会失败。具体端口、模型标签与可用性请以 Ollama 与模型发布方的当前信息为准。

护栏可以放在哪些位置

一次典型的 Agent 运行会经过若干阶段：接收请求、决定做什么、调用工具、处理工具结果、最终生成回复。Agents SDK 允许在这些阶段之间插入护栏，按位置可分为四类：

• 输入护栏（input guardrail）：在 Agent 开始工作之前检查用户请求。
• 工具输入护栏（tool-input guardrail）：在工具收到参数之前检查这次调用。
• 工具输出护栏（tool-output guardrail）：在工具结果返回给 Agent 之前检查它。
• 输出护栏（output guardrail）：在最终回复展示给用户之前检查它。

按位置和实施方式的不同，护栏可以让流程继续、拒绝某个值，或者用一个修改后的版本替换它。选择哪一类，取决于你希望在哪个环节建立边界。本教程聚焦工具输出护栏，因为它在「数据已经取到、但模型还没看到」这个窗口里生效，能从源头上避免模型接触到不该外泄的内容。

操作步骤

第一步：构造带内部字段的案例数据

先设定一个具体场景：客户联系在线商店，询问一笔尚未到账的退款。客服 Agent 需要检索该订单的案卷记录来回答。这些记录来自支持系统的不同部分，既有可以告知客户的信息，也混有仅供内部使用的内容。

SUPPORT_RECORDS = [
    {
        "record_id": "order:MUN-2048",
        "record_type": "order",
        "content": "Order MUN-2048 has a refund approved for EUR 120.",
    },
    {
        "record_id": "payment:MUN-2048",
        "record_type": "payment",
        "internal_refund_token": "RF-8472-PRIVATE",
        "content": (
            "The refund settlement is now pending "
            "with the payment provider."
        ),
    },
    {
        "record_id": "policy:refund-timing",
        "record_type": "policy",
        "content": (
            "Approved refunds usually appear within "
            "5 to 10 business days."
        ),
    },
    {
        "record_id": "review:MUN-2048",
        "record_type": "review",
        "fraud_risk_score": 0.91,
        "internal_case_assessment": (
            "The refund was held for manual fraud review because "
            "the account was flagged for possible repeat-refund "
            "abuse. The review is now complete."
        ),
        "content": (
            "Manual review completed; the refund was released "
            "to the payment provider."
        ),
    },
]

可以看到，同一批记录里既有退款金额、当前状态、预计处理时长这类可以公开的信息，也有内部退款令牌、欺诈风险评分，以及「账户因疑似重复退款滥用被标记」这类内部评估。它们属于同一个案件，会被同一次检索一并返回。人工客服知道哪些能说、哪些不能说，但模型未必能稳定地做出这个区分——这正是需要护栏的原因。

第二步：编写检索工具

接下来给 Agent 一个检索工具，让它能访问这些记录。实现方式是按查询词做简单的子串匹配，把命中的记录序列化为 JSON 返回：

import json

def search_support_records(query: str) -> str:
    """Search support records relevant to the request."""
    query_terms = query.lower().split()
    matches = []
    for record in SUPPORT_RECORDS:
        searchable_text = " ".join(
            [
                record["record_id"],
                record["record_type"],
                record["content"],
            ]
        ).lower()
        if any(term in searchable_text for term in query_terms):
            matches.append(record)
    return json.dumps(matches, ensure_ascii=False)

search_tool = function_tool(search_support_records)

注意 json.dumps 使用了 ensure_ascii=False，这样非 ASCII 字符会原样输出，便于阅读和后续解析。工具返回的是完整记录对象，包括那些内部字段——在还没有加护栏的情况下，这些字段会原封不动地进入模型的上下文。

第三步：定义未加护栏的 Agent 并观察问题

先建立一个基线 Agent，只给指令和工具，不加任何运行时边界：

support_agent = Agent(
    name="Customer support agent",
    instructions=(
        "Investigate the support request using the available records "
        "and prepare a customer-facing response."
    ),
    model=model,
    tools=[search_tool],
)

然后用客户的真实请求运行它：

from agents import RunConfig, Runner

user_request = (
    "My refund for order MUN-2048 has not arrived. "
    "Tell me everything you know about what happened."
)

baseline_result = await Runner.run(
    support_agent,
    user_request,
    max_turns=8,
    run_config=RunConfig(tracing_disabled=True),
)
print(baseline_result.final_output)

运行结果中，回复的大部分内容是合适的：它说明了退款金额为 120 欧元、退款已获批准、结算正在支付服务商处处理、通常 5 到 10 个工作日到账。但回复同时告诉客户「账户因疑似重复退款滥用被标记，因此触发了人工欺诈审核」。这句话对内部流程是准确的，却不适合出现在面向客户的答复里。

这个结果说明：仅靠 instructions 里「准备一份面向客户的回复」这样的措辞，并不能阻止模型把检索到的内部细节复述出来。模型看到什么，就有可能说什么。

第四步：实现工具输出护栏做字段脱敏

目标很明确：在检索工具返回原始 JSON 之后、这些记录被交回 Agent 之前，把内部字段删掉。这样模型从头到尾都看不到敏感信息，也就没有泄露的可能。工具输出护栏正好落在这个位置——它拦截工具结果，可以放行、拒绝或替换。

from agents import (
    ToolGuardrailFunctionOutput,
    tool_output_guardrail,
)

SENSITIVE_FIELDS = {
    "internal_refund_token",
    "fraud_risk_score",
    "internal_case_assessment",
}

@tool_output_guardrail
def redact_sensitive_fields(data):
    records = json.loads(data.output)
    for record in records:
        for field_name in SENSITIVE_FIELDS:
            record.pop(field_name, None)
    sanitized_output = json.dumps(
        records,
        ensure_ascii=False,
    )
    return ToolGuardrailFunctionOutput.reject_content(
        sanitized_output
    )

这段代码有几个要点值得展开：

• 护栏函数通过 data.output 拿到工具的原始返回值，先用 json.loads 解析成记录列表。
• 对每条记录遍历敏感字段集合，用 record.pop(field_name, None) 删除字段。第二个参数 None 保证字段不存在时不会抛异常，因此同一份护栏可以安全地作用于结构不完全一致的记录。
• 脱敏后的列表重新序列化，再通过 ToolGuardrailFunctionOutput.reject_content() 返回。这个调用的语义是：拒绝原始工具输出，并用传入的内容替换它。也就是说，Agent 最终看到的是脱敏版本，而不是原始版本。

字段名集中放在 SENSITIVE_FIELDS 集合里，便于后续维护：新增内部字段时只需改这一处，不必改动护栏逻辑本身。

第五步：把护栏挂到工具上并重新运行

护栏需要绑定到具体的工具。用 function_tool 重新包装检索函数，并通过 tool_output_guardrails 参数传入护栏列表：

from agents import function_tool

guarded_search_tool = function_tool(
    search_support_records,
    tool_output_guardrails=[redact_sensitive_fields],
)
support_agent.tools = [guarded_search_tool]

然后用完全相同的客户请求再跑一次：

guarded_result = await Runner.run(
    support_agent,
    user_request,
    max_turns=8,
    run_config=RunConfig(tracing_disabled=True),
)
print(guarded_result.final_output)

这一次的回复仍然包含客户需要的信息：退款已获批准、金额 120 欧元、人工审核已完成、退款已释放给支付服务商、结算正在处理中、通常 5 到 10 个工作日到账。但「账户因疑似退款滥用被标记」这一句不再出现，内部退款令牌与风险评分也没有任何痕迹。护栏在数据进入模型之前就完成了过滤，因此模型根本没有机会复述它们。

一个完整示例

把上面的片段按顺序拼起来，就是一个可以端到端运行的最小示例。以下代码假设 Ollama 已在本地运行，且已拉取对应模型：

import asyncio
import json

from openai import AsyncOpenAI
from agents import (
    Agent,
    OpenAIChatCompletionsModel,
    RunConfig,
    Runner,
    ToolGuardrailFunctionOutput,
    function_tool,
    tool_output_guardrail,
)

SUPPORT_RECORDS = [
    {
        "record_id": "order:MUN-2048",
        "record_type": "order",
        "content": "Order MUN-2048 has a refund approved for EUR 120.",
    },
    {
        "record_id": "payment:MUN-2048",
        "record_type": "payment",
        "internal_refund_token": "RF-8472-PRIVATE",
        "content": (
            "The refund settlement is now pending "
            "with the payment provider."
        ),
    },
    {
        "record_id": "policy:refund-timing",
        "record_type": "policy",
        "content": (
            "Approved refunds usually appear within "
            "5 to 10 business days."
        ),
    },
    {
        "record_id": "review:MUN-2048",
        "record_type": "review",
        "fraud_risk_score": 0.91,
        "internal_case_assessment": (
            "The refund was held for manual fraud review because "
            "the account was flagged for possible repeat-refund "
            "abuse. The review is now complete."
        ),
        "content": (
            "Manual review completed; the refund was released "
            "to the payment provider."
        ),
    },
]

SENSITIVE_FIELDS = {
    "internal_refund_token",
    "fraud_risk_score",
    "internal_case_assessment",
}

def search_support_records(query: str) -> str:
    """Search support records relevant to the request."""
    query_terms = query.lower().split()
    matches = []
    for record in SUPPORT_RECORDS:
        searchable_text = " ".join(
            [
                record["record_id"],
                record["record_type"],
                record["content"],
            ]
        ).lower()
        if any(term in searchable_text for term in query_terms):
            matches.append(record)
    return json.dumps(matches, ensure_ascii=False)

@tool_output_guardrail
def redact_sensitive_fields(data):
    records = json.loads(data.output)
    for record in records:
        for field_name in SENSITIVE_FIELDS:
            record.pop(field_name, None)
    sanitized_output = json.dumps(
        records,
        ensure_ascii=False,
    )
    return ToolGuardrailFunctionOutput.reject_content(
        sanitized_output
    )

async def main():
    client = AsyncOpenAI(
        api_key="ollama",
        base_url="http://localhost:11434/v1",
    )
    model = OpenAIChatCompletionsModel(
        model="gemma4:e4b",
        openai_client=client,
    )

    guarded_search_tool = function_tool(
        search_support_records,
        tool_output_guardrails=[redact_sensitive_fields],
    )

    support_agent = Agent(
        name="Customer support agent",
        instructions=(
            "Investigate the support request using the available records "
            "and prepare a customer-facing response."
        ),
        model=model,
        tools=[guarded_search_tool],
    )

    user_request = (
        "My refund for order MUN-2048 has not arrived. "
        "Tell me everything you know about what happened."
    )

    result = await Runner.run(
        support_agent,
        user_request,
        max_turns=8,
        run_config=RunConfig(tracing_disabled=True),
    )
    print(result.final_output)

asyncio.run(main())

运行后应当看到一份不含内部字段的客户回复。如果想验证护栏确实生效，可以临时把 tool_output_guardrails 参数去掉再跑一次，对比两次输出中是否出现风险评分或账户标记的描述。

注意事项

护栏位置的选择

四类护栏解决的是不同问题，选错位置往往达不到预期效果。可以用下面几个问题来判断：

• 这个请求应该在 Agent 开始之前就被拒绝吗？用输入护栏。
• 工具参数应该在动作执行之前被检查吗？用工具输入护栏。
• 工具返回的数据应该在 Agent 使用之前被过滤吗？用工具输出护栏。
• 最终回复应该在用户看到之前被检查吗？用输出护栏。

本教程的场景属于第三类：敏感信息来自工具结果，而不是用户输入，也不是模型凭空生成的。如果只在输出护栏里做关键词过滤，模型其实已经看到了内部字段，只是最后被拦住；而工具输出护栏让模型从一开始就接触不到这些数据，风险面更小。

替换语义与返回值

示例中使用的是 reject_content()，它的作用是拒绝原始工具输出并替换为脱敏版本。护栏的返回值决定了后续流程看到什么，因此不要只做检查而不返回结果。另外，护栏函数接收的是工具输出对象，原始字符串在 data.output 上，直接对 data 做 json.loads 会报错。

字段删除的容错

record.pop(field_name, None) 的默认值参数不能省。真实系统里同一类记录的结构可能并不统一，某些记录没有某个内部字段是常态；省略默认值会让护栏在遇到缺字段时抛出异常，进而中断整次运行。

本地模型的边界

示例用 Ollama 托管本地模型，说明护栏机制不依赖特定厂商的托管服务。但本地模型的指令遵循能力、工具调用稳定性与上下文长度各不相同，实际表现需要按所选模型验证。模型标签、可用版本与硬件要求请以 Ollama 和模型发布方的当前信息为准。

运行配置

示例在 Runner.run 中传入了 max_turns=8 与 RunConfig(tracing_disabled=True)。max_turns 限制 Agent 的轮次上限，避免工具调用循环无法收敛；关闭 tracing 是为了在本地演示时不依赖外部追踪服务。生产环境中是否开启追踪、轮次上限设多少，需要结合具体部署方式决定。

护栏不是万能的

工具输出护栏能阻止模型看到被删除的字段，但它不会自动识别所有敏感内容。示例中的敏感字段是显式列出的白名单式黑名单，新增字段或改变记录结构时需要同步维护 SENSITIVE_FIELDS。对于自由文本中夹杂的内部信息，仅靠字段删除并不够，还需要配合内容层面的检查。