嘿,大家好!这里是一个专注于前沿AI和智能体的频道~
昨天,OpenAI 悄悄开源了一个完整的客服 Agent Demo,一个基于官方 openai-agents SDK 构建、包含前后端的完整项目。
整个项目还有点意思,算是一个新手教科书级别的多智能体(Multi-Agent)协作范例。我扒了扒它的源码,里面藏着一套关于智能体协作、切换和安全护栏的最佳实践。
核心架构
这个 Demo 模拟了一个航空公司的客服场景。它的架构非常经典,采用了跟上周分享的很多大厂的产品一样,都是 Orchestrator-Worker 架构(讲道理,这个架构名字可能很多,这个词是我经常用的~):
- 1个 Triage Agent (Lead智能体): 它的唯一职责就是理解用户意图,然后把任务“派发”给最合适的专家。
- N个 Specialist Agents (专家智能体): 每个专家只负责一个垂直领域,比如:Seat Booking Agent: 负责预订座位。Flight Status Agent: 负责查询航班状态。Cancellation Agent: 负责取消航班。FAQ Agent: 负责回答常见问题。
这种架构的好处是显而易见的:高内聚、低耦合。每个专家 Agent 的提示词(Prompt)、工具(Tools)和知识都可以独立维护,扩展起来非常方便,不会互相干扰。
模式一:带状态的智能体切换 (Stateful Handoff)
多智能体协作的核心难题,就是如何优雅地切换,并传递上下文信息。
OpenAI 在这个 Demo 里,用的是 handoff 方案。
Triage Agent 的核心配置如下,它通过 handoffs 列表来定义它可以将任务移交给哪些专家。
triage_agent = Agent[AirlineAgentContext](
name="Triage Agent",
model="gpt-4.1",
handoff_description="A triage agent that can delegate a customer's request to the appropriate agent.",
instructions=(
f"{RECOMMENDED_PROMPT_PREFIX} "
"You are a helpful triaging agent. You can use your tools to delegate questions to other appropriate agents."
),
handoffs=[
flight_status_agent,
handoff(agent=cancellation_agent, on_handoff=on_cancellation_handoff),
faq_agent,
handoff(agent=seat_booking_agent, on_handoff=on_seat_booking_handoff),
],
input_guardrails=[relevance_guardrail, jailbreak_guardrail],
)
值得注意的是,上边 handoff() 函数里的on_handoff 回调。
比如,当用户想改座位时,Triage Agent 会把任务交给 Seat Booking Agent。在交接的瞬间,会触发 on_seat_booking_handoff 这个函数:
async def on_seat_booking_handoff(context: RunContextWrapper[AirlineAgentContext]) -> None:
"""Set a random flight number when handed off to the seat booking agent."""
context.context.flight_number = f"FLT-{random.randint(100, 999)}"
context.context.confirmation_number = "".join(random.choices(string.ascii_uppercase + string.digits, k=6))
这个回调函数会在“交接前”更新共享的上下文(Context),比如自动生成一个航班号和确认码。这样,当 Seat Booking Agent 接手时,它需要的初始信息就已经准备好了,可以直接开始工作,而不需要再向用户询问。
这个“带状态的切换”模式,可以很好提升了用户体验和系统效率。
模式二:低成本的安全护栏 (Guardrail)
Agent 安全是另一个让人头疼的问题。如果用户问一些无关的问题(“写一首关于草莓的诗”),或者试图进行提示词注入攻击(“忽略你之前的指令……”),该怎么办?
OpenAI 的方案是设置“安全护栏”(Guardrail),而且用的是一种非常经济高效的方式。
他们创建了两个专门的护栏函数:
1.relevance_guardrail: 判断用户输入是否与航空业务相关。
2.jailbreak_guardrail: 检测用户是否在尝试越狱攻击。
这两个护栏的背后,都是一个独立的、更小、更便宜的模型(gpt-4.1-mini)。
它的逻辑是:
1在主 Agent(如 Seat Booking Agent)处理用户输入之前。
2.先用这个小模型快速检查一下输入的“安全性”和“相关性”。
3.如果小模型检测到问题(比如 is_relevant: False),就会直接触发“Tripwire”(绊线),拒绝回答,并阻止请求进入更贵、更强大的主模型。
@input_guardrail(name="Relevance Guardrail")
async def relevance_guardrail(
context: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
"""Guardrail to check if input is relevant to airline topics."""
result = await Runner.run(guardrail_agent, input, context=context.context)
final = result.final_output_as(RelevanceOutput)
return GuardrailFunctionOutput(output_info=final, tripwire_triggered=not final.is_relevant)
这种“前置检查”的设计,将安全逻辑与业务逻辑解耦,可以保证了系统的鲁棒性,同时节约成本。
模式三:后端驱动的前端交互
还有一个有趣的设计。当 Seat Booking Agent 需要让用户在座位图上选座时,它会调用一个名为 display_seat_map 的工具。
这个工具的实现非常简单,它只返回一个特殊的字符串:DISPLAY_SEAT_MAP。
@function_tool(
name_override="display_seat_map",
description_override="Display an interactive seat map to the customer so they can choose a new seat."
)
async def display_seat_map(
context: RunContextWrapper[AirlineAgentContext]
) -> str:
"""Trigger the UI to show an interactive seat map to the customer."""
# The returned string will be interpreted by the UI to open the seat selector.
return "DISPLAY_SEAT_MAP"
前端 UI 监听到这个特殊的字符串后,就会触发渲染“座位图”组件的逻辑。这是一个非常巧妙的“后端驱动UI”模式,让 Agent 能够控制前端的复杂交互,而不仅仅是返回文本。
如何上手?
这个项目上手非常简单:
1.Clone 仓库:git clone https://github.com/openai/openai-cs-agents-demo.git
2.设置 API Key:在 python-backend 目录下创建一个 .env 文件,写入你的 OPENAI_API_KEY。
3.安装依赖:分别进入 python-backend 和 ui 目录,安装依赖。
4.运行:在 ui 目录下运行 npm run dev,它会自动同时启动前端和后端。
文章来自于微信公众号“探索AGI”,作者是“猕猴桃”。