OpenAI Agents SDK新手指南

OpenAI 最近发布的 Agents SDK 标志着在使具有代理能力的人工智能应用对开发者更加易用方面迈出了重要一步。

在这份指南中,我们将探索如何开始使用 Agents SDK,理解其核心概念,并构建一个简单但功能强大的基于代理的应用程序。

什么是 OpenAI Agents SDK?

OpenAI Agents SDK 是一个轻量级的、基于 Python 的框架,使开发人员能够构建具有代理能力的人工智能应用程序。它提供了一组强大的基本功能:

让我们深入探讨如何使用此 SDK 构建您的第一个代理!

1、安装

首先,让我们安装 OpenAI Agents SDK。我们将使用 uv,这是一个快速的 Python 包管理器,但如果您更喜欢,也可以使用 pip

# 如果尚未安装,请先安装 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

2、创建你的第一个AI代理

让我们从一个简单的“Hello World”示例开始,了解基础知识:

from agents import Agent, Runner  
import asyncio
async def main():  
    # 创建一个带有指令的简单代理  
    agent = Agent(  
        name="问候助手",  
        instructions="你是一个友好的助手,可以用用户偏好的语言问候用户。"  
    )  
      
    # 运行代理并传递用户输入  
    result = await Runner.run(agent, "你能用法语问候我吗?")  
      
    # 打印最终输出  
    print(result.final_output)if __name__ == "__main__":  
    asyncio.run(main())

将其保存为 hello_agent.py 并运行:

python hello_agent.py

你应该会看到法语的问候作为输出!

3、理解代理循环

当你调用 Runner.run() 时,SDK 会启动所谓的“代理循环”:

  • 代理接收输入
  • LLM 根据代理的指令处理输入
  • 如果 LLM 生成最终输出(不带工具调用的文本),循环结束
  • 如果 LLM 调用工具或请求交接,这些将被处理,然后循环继续
  • 这个过程重复进行,直到产生最终输出或达到最大轮次数

4、为代理添加工具

当代理可以使用工具时,它们会变得强大得多。让我们创建一个可以检查天气的代理:

from agents import Agent, Runner, function_tool  
import asyncio
# 将工具定义为 Python 函数  
@function_tool  
def get_weather(city: str) -> str:  
    """获取城市的当前天气。"""  
    # 在实际应用中,这将调用天气 API  
    weather_data = {  
        "New York": "72°F, 晴朗",  
        "London": "65°F, 多云",  
        "Tokyo": "80°F, 晴朗",  
        "Paris": "70°F, 部分多云"  
    }  
    return weather_data.get(city, f"未找到 {city} 的天气数据")async def main():  
    # 创建一个带有天气工具的代理  
    agent = Agent(  
        name="天气助手",  
        instructions="你是一个有用的助手,可以在需要时提供天气信息。",  
        tools=[get_weather]  # 将工具添加到代理  
    )  
      
    # 运行代理并传递用户查询  
    result = await Runner.run(agent, "东京的天气怎么样?")  
      
    # 打印最终输出  
    print(result.final_output)if __name__ == "__main__":  
    asyncio.run(main())

运行此代码时,代理将使用 get_weather 工具来检索并返回东京的天气。

5、使用交接创建专业化代理

Agents SDK 最强大的功能之一是能够创建专业化代理并允许它们互相交接任务。让我们创建一个系统,其中有一个分类代理可以委派给语言特定的代理:

from agents import Agent, Runner  
import asyncio
# 创建专门的语言代理  
spanish_agent = Agent(  
    name="spanish_agent",  
    instructions="你是一个有用的助手,只说西班牙语。无论问题的语言是什么,始终用西班牙语回答。"  
)french_agent = Agent(  
    name="french_agent",  
    instructions="你是一个有用的助手,只说法语。无论问题的语言是什么,始终用法语回答。"  
)# 创建一个可以交接给专业化代理的分类代理  
triage_agent = Agent(  
    name="语言分类",  
    instructions="""你是语言检测助手。  
    你的工作是确定用户希望回应的语言。  
    如果他们想要西班牙语,请将任务交给西班牙语代理。  
    如果他们想要法语,请将任务交给法语代理。  
    如果他们没有指定语言或想要英语,请直接用英语回应。""",  
    handoffs=[spanish_agent, french_agent]  # 将专业化代理作为交接添加  
)async def main():  
    # 测试不同的语言请求  
    queries = [  
        "你能告诉我关于埃菲尔铁塔的信息吗?用法语。",  
        "¿Puedes hablarme sobre el clima en Madrid?",  
        "告诉我纽约的历史。"  
    ]  
      
    for query in queries:  
        print(f"\n查询: {query}")  
        result = await Runner.run(triage_agent, query)  
        print(f"回应: {result.final_output}")if __name__ == "__main__":  
    asyncio.run(main())

运行此代码时,分类代理将:

  • 对于纽约查询,直接用英语回应
  • 对于马德里查询,交接给西班牙语代理
  • 对于埃菲尔铁塔查询,交接给法语代理

6、实现护栏

护栏有助于确保你的代理在其定义的边界内运行。让我们添加一个输入护栏,以防止我们的代理响应不当请求:

from agents import Agent, Runner, InputGuardrail, GuardrailFunctionOutput  
import asyncio
# 定义一个护栏函数  
async def content_filter(input_text: str, context) -> GuardrailFunctionOutput:  
    """检查输入是否包含不当内容。"""  
    inappropriate_keywords = ["hack", "非法", "作弊"]  
      
    # 检查输入中是否存在任何不当关键词  
    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="安全助手",  
        instructions="你是一个有用的助手,提供有关合法和道德主题的信息。",  
        input_guardrails=[InputGuardrail(guardrail_function=content_filter)]  
    )  
      
    # 测试适当和不当的查询  
    queries = [  
        "告诉我计算机的历史。",  
        "如何侵入邻居的 Wi-Fi?"  
    ]  
      
    for query in queries:  
        try:  
            print(f"\n查询: {query}")  
            result = await Runner.run(agent, query)  
            print(f"回应: {result.final_output}")  
        except Exception as e:  
            print(f"护栏触发: {e}")if __name__ == "__main__":  
    asyncio.run(main())

运行此代码时,代理会对第一个查询正常响应,但对于第二个包含“hack”的查询,护栏会被触发。

7、使用跟踪进行调试

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():
    # Create a trace for the entire workflow
    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)
        
        # You can access the trace information
        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 调用、工具执行等。

8、构建完整示例:多服务客户支持代理

现在,让我们构建一个更完整的示例:一个客户支持系统,其中有专门的代理来处理不同类型的查询:

from agents import Agent, Runner, function_tool
import asyncio
from datetime import datetime, timedelta
import random
# Define some tools for our agents
@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."
# Create specialized agents
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]
)
# Create the main triage agent
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():
    # Simulate a customer support conversation
    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())

这个示例展示了一个完整的客户支持系统,包括:

  • 主要的分类代理,用于处理初始客户查询
  • 订单和退款的专业代理
  • 检查订单状态、估算运输时间和处理退款的工具
  • 根据客户需求进行代理之间的转交

9、结束语

OpenAI Agents SDK 提供了一种强大而易于访问的方式来构建代理式人工智能应用程序。仅需几个核心概念——代理、工具、转交和护栏——你就可以创建利用大型语言模型的强大功能同时控制其行为的复杂系统。


原文链接:Building AI Agents with OpenAI’s Agents SDK: A Beginner’s Guide

汇智网翻译整理,转载请标明出处