OpenAI Agents SDK的发布，标志着开发者可以更便捷地构建基于代理（Agents）的AI应用。本文将深入探讨如何使用Agents SDK入门，理解其核心概念，并构建一个简单但强大的基于代理的应用，同时也会涉及到工具 (Tools)、移交 (Handoffs)和防护栏 (Guardrails) 的使用。
OpenAI Agents SDK简介：赋能智能代理
OpenAI Agents SDK是一个轻量级的Python框架，旨在赋能开发者构建具有代理能力的AI应用。它提供了一组强大的原语，包括：
- 代理（Agents）：具备指令和工具的大型语言模型（LLMs）。
- 移交（Handoffs）：代理将任务委派给其他专业代理的机制。
- 防护栏（Guardrails）：验证输入和输出的系统。
- 追踪（Tracing）：内置的可视化和调试代理工作流程的能力。
这些原语共同构成了Agents SDK的核心，使开发者能够更轻松地构建复杂的、多步骤的智能应用。
安装与配置：快速上手Agents SDK
在使用Agents SDK之前，需要进行安装和配置。以下步骤演示了如何使用uv（一个快速的Python包安装器，当然你也可以使用pip）安装和配置Agents SDK：
-
安装uv (如果尚未安装):
curl -sSf https://install.python-poetry.org | python3 - -
创建并激活虚拟环境:
python -m venv venv source venv/bin/activate # 在Windows上，使用: venv\Scripts\activate -
安装OpenAI Agents SDK:
uv pip install openai-agents-sdk -
设置OpenAI API密钥:
export OPENAI_API_KEY=your-api-key-here在Windows上，使用：
set OPENAI_API_KEY=your-api-key-here
安装完成后，就可以开始使用Agents SDK构建你的第一个代理应用了。
构建你的第一个代理：Hello World示例
首先，让我们创建一个简单的“Hello World”示例，以理解基本概念：
from agents import Agent, Runner
import asyncio
async def main():
# 创建一个简单的代理，带有指令
agent = Agent(
name="Greeting Assistant",
instructions="You are a friendly assistant that greets users in their preferred language."
)
# 运行代理，使用用户输入
result = await Runner.run(agent, "Can you greet me in French?")
# 打印最终输出
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
将此代码保存为hello_agent.py并运行：
python hello_agent.py
你应该看到以法语问候语作为输出！这个简单的示例展示了如何创建一个代理，定义其指令，并使用Runner运行它。Runner 负责处理与LLM的交互，并管理代理的生命周期。
理解代理循环：核心工作流程
当你调用Runner.run()时，Agents SDK会启动所谓的“代理循环”：
- 代理接收输入：用户提供的初始查询或指令。
- LLM处理输入：LLM基于代理的指令处理输入。
- 判断是否产生最终输出：如果LLM产生最终输出（没有工具调用的文本），则循环结束。
- 处理工具调用或移交请求：如果LLM调用工具或请求移交，则会处理这些请求，并且循环继续。
这个循环会重复执行，直到产生最终输出或达到最大迭代次数。 理解这个循环对于调试和优化你的代理至关重要。 例如，如果你的代理没有产生预期的输出，你可以检查LLM是否正确地处理了输入，或者是否需要添加或修改工具。
添加工具：扩展代理的能力
代理可以通过使用工具变得更加强大。工具是代理可以使用的外部函数或API。 让我们创建一个可以检查天气的代理：
from agents import Agent, Runner, function_tool
import asyncio
# 定义一个工具作为Python函数
@function_tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
# 在实际应用中，这会调用一个天气API
weather_data = {
"New York": "72°F, Sunny",
"London": "65°F, Cloudy",
"Tokyo": "80°F, Clear",
"Paris": "70°F, Partly Cloudy"
}
return weather_data.get(city, f"Weather data not available for {city}")
async def main():
# 创建一个具有天气工具的代理
agent = Agent(
name="Weather Assistant",
instructions="You are a helpful assistant that provides weather information when asked.",
tools=[get_weather] # 将工具添加到代理
)
# 运行代理，使用用户查询
result = await Runner.run(agent, "What's the weather like in Tokyo?")
# 打印最终输出
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
运行此代码时，代理将使用get_weather工具来获取并返回东京的天气。@function_tool装饰器将Python函数转换为代理可以使用的工具。这使得将外部功能集成到代理中变得非常容易。
创建专业代理：使用移交进行任务委派
Agents SDK最强大的功能之一是能够创建专业的代理，并允许它们相互移交任务。 让我们创建一个系统，其中包含一个可以委派给特定语言代理的分类代理：
from agents import Agent, Runner
import asyncio
# 创建专业的语言代理
spanish_agent = Agent(
name="spanish_agent",
instructions="You are a helpful assistant that only speaks Spanish. Always respond in Spanish, regardless of the language of the question."
)
french_agent = Agent(
name="french_agent",
instructions="You are a helpful assistant that only speaks French. Always respond in French, regardless of the language of the question."
)
# 创建一个可以移交给专业代理的分类代理
triage_agent = Agent(
name="language_triage",
instructions="""You are a language detection assistant.
Your job is to determine what language the user wants a response in.
If they want Spanish, hand off to the spanish_agent.
If they want French, hand off to the french_agent.
If they don't specify a language or want English, respond directly in English.""",
handoffs=[spanish_agent, french_agent] # 将专业代理添加为移交
)
async def main():
# 使用不同的语言请求进行测试
queries = [
"Can you tell me about the Eiffel Tower in French?",
"¿Puedes hablarme sobre el clima en Madrid?",
"Tell me about the history of New York."
]
for query in queries:
print(f"\nQuery: {query}")
result = await Runner.run(triage_agent, query)
print(f"Response: {result.final_output}")
if __name__ == "__main__":
asyncio.run(main())
运行此代码时，分类代理将：
- 直接以英语回复有关纽约的查询。
- 移交给西班牙语代理，处理有关马德里的查询。
- 移交给法语代理，处理有关埃菲尔铁塔的查询。
移交是构建复杂的多步骤工作流程的关键。通过将任务分解为更小的、更专业的代理，你可以构建更有效和可维护的系统。
实施防护栏：确保代理安全和负责任
防护栏有助于确保你的代理在定义的边界内运行。 让我们添加一个输入防护栏，以防止我们的代理回复不适当的请求：
from agents import Agent, Runner, InputGuardrail, GuardrailFunctionOutput
import asyncio
# 定义一个防护栏函数
async def content_filter(input_text: str, context) -> GuardrailFunctionOutput:
"""Check if the input contains inappropriate content."""
inappropriate_keywords = ["hack", "illegal", "cheat"]
# 检查输入中是否存在不适当的关键字
contains_inappropriate = any(keyword in input_text.lower() for keyword in inappropriate_keywords)
return GuardrailFunctionOutput(
output_info={"contains_inappropriate": contains_inappropriate},
tripwire_triggered=contains_inappropriate
)
async def main():
# 创建一个带有防护栏的代理
agent = Agent(
name="Safe Assistant",
instructions="You are a helpful assistant that provides information on legal and ethical topics.",
input_guardrails=[InputGuardrail(guardrail_function=content_filter)]
)
# 使用适当的和不适当的查询进行测试
queries = [
"Tell me about the history of computers",
"How can I hack into my neighbor's WiFi?"
]
for query in queries:
try:
print(f"\nQuery: {query}")
result = await Runner.run(agent, query)
print(f"Response: {result.final_output}")
except Exception as e:
print(f"Guardrail triggered: {e}")
if __name__ == "__main__":
asyncio.run(main())
运行此代码时，代理将正常回复第一个查询，但会触发包含“hack”的第二个查询的防护栏。 防护栏对于确保你的代理安全、负责任且符合伦理道德至关重要。 它们可以用于过滤不适当的内容、防止有害行为和强制执行业务规则。
使用追踪进行调试：深入了解代理行为
Agents SDK包含内置的追踪功能，可帮助你了解代理执行期间发生的事情。 让我们将追踪添加到我们的天气代理：
from agents import Agent, Runner, function_tool
from agents.tracing import trace
import asyncio
@function_tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
weather_data = {
"New York": "72°F, Sunny",
"London": "65°F, Cloudy",
"Tokyo": "80°F, Clear",
"Paris": "70°F, Partly Cloudy"
}
return weather_data.get(city, f"Weather data not available for {city}")
async def main():
# 为整个工作流程创建一个跟踪
with trace(workflow_name="weather_inquiry"):
agent = Agent(
name="Weather Assistant",
instructions="You are a helpful assistant that provides weather information when asked.",
tools=[get_weather]
)
result = await Runner.run(
agent,
"What's the weather like in Tokyo and Paris?",
run_config=RunConfig(
workflow_name="weather_inquiry",
trace_include_sensitive_data=True
)
)
print(result.final_output)
# 你可以访问跟踪信息
current_trace = get_current_trace()
if current_trace:
print(f"Trace ID: {current_trace.trace_id}")
if __name__ == "__main__":
from agents import RunConfig
from agents.tracing import get_current_trace
asyncio.run(main())
此代码将追踪添加到我们的天气代理，使我们能够查看整个工作流程，包括LLM调用、工具执行等。 追踪对于调试和优化你的代理至关重要。 它可以帮助你识别性能瓶颈、了解LLM的行为并诊断错误。
构建完整的示例：多服务客户支持代理
现在，让我们构建一个更完整的示例：一个具有针对不同类型查询的专业代理的客户支持系统：
from agents import Agent, Runner, function_tool
import asyncio
from datetime import datetime, timedelta
import random
# 定义我们代理的一些工具
@function_tool
def check_order_status(order_id: str) -> str:
"""Check the status of an order by ID."""
# Simulate order database
statuses = ["Processing", "Shipped", "Delivered", "Cancelled"]
return f"Order {order_id} is currently {random.choice(statuses)}."
@function_tool
def get_shipping_estimate(product_id: str, destination: str) -> str:
"""Get shipping estimate for a product to a destination."""
# Simulate shipping calculation
days = random.randint(2, 10)
estimated_date = (datetime.now() + timedelta(days=days)).strftime("%B %d, %Y")
return f"Shipping to {destination} takes approximately {days} days. Estimated delivery: {estimated_date}"
@function_tool
def process_refund(order_id: str, reason: str) -> str:
"""Process a refund request."""
return f"Refund for order {order_id} has been initiated. Reason: {reason}. It will be processed within 3-5 business days."
# 创建专业的代理
order_agent = Agent(
name="order_specialist",
instructions="""You are an order specialist.
Help customers check their order status and provide shipping estimates.
Be friendly and professional.""",
tools=[check_order_status, get_shipping_estimate]
)
refund_agent = Agent(
name="refund_specialist",
instructions="""You are a refund specialist.
Help customers process refunds for their orders.
Be empathetic and helpful.""",
tools=[process_refund]
)
# 创建主分类代理
support_agent = Agent(
name="customer_support",
instructions="""You are a customer support agent for an e-commerce store.
If customers have questions about order status or shipping, hand off to the order specialist.
If customers want to request a refund, hand off to the refund specialist.
For general questions, answer directly.
Always be polite and helpful.""",
handoffs=[order_agent, refund_agent]
)
async def main():
# 模拟客户支持对话
queries = [
"What's the status of my order #12345?",
"I want to return my purchase because it's damaged. Order #54321.",
"Do you offer gift wrapping services?",
"How long will shipping take for product XYZ-789 to Seattle?"
]
for i, query in enumerate(queries):
print(f"\n--- Customer Query {i+1} ---")
print(f"Customer: {query}")
result = await Runner.run(support_agent, query)
print(f"Agent: {result.final_output}")
if __name__ == "__main__":
asyncio.run(main())
此示例演示了一个完整的客户支持系统，其中包含：
- 处理初始客户查询的主分类代理。
- 用于订单和退款的专业代理。
- 用于检查订单状态、估算运费和处理退款的工具。
- 基于客户需求的代理之间的移交。
这个示例展示了Agents SDK的强大功能，它可以构建复杂的多代理系统，这些系统可以自动执行各种任务。 例如，可以使用此框架构建虚拟助手、聊天机器人、自动化营销系统等。
结论：Agent SDK的未来展望
OpenAI Agents SDK提供了一种强大但易于访问的方式来构建基于代理的AI应用。 凭借几个核心概念——代理、工具、移交和防护栏——你可以创建复杂的系统，这些系统可以利用大型语言模型的力量，同时保持对其行为的控制。
随着你继续探索SDK，你可能想要尝试：
- 连接到真实API的更复杂的工具实现。
- 用于结构化数据的自定义输入和输出类型。
- 用于安全和质量控制的高级防护栏。
- 用于生产应用程序的追踪和监控。
OpenAI Agents SDK的出现，为构建智能化、自动化和可扩展的AI应用打开了新的可能性。随着LLM技术的不断发展，Agents SDK有望成为开发者构建下一代AI应用的关键工具。