理解使用OpenAI Agents SDK进行AI代理开发的概况及使用方法

pip install openai-agents

from agents import Agent

agent = Agent(
    name='simple agent', # 必须
    instructions='请将输入的内容原样返回',
    model='gpt-4o-mini' # 当前未指定时默认为`gpt-4o`
)

import asyncio
from agents import Runner

async def main():
    result = await Runner.run(agent, '早上好!!')
    print(result)

if __name__ == '__main__':
    asyncio.run(main())

import asyncio
from agents import Agent, Runner, FileSearchTool
agent = Agent(
    name='file search agent',
    instructions='你是一位技术博客运营者。请使用file_search工具搜索博客文件。',
    model='gpt-4o-mini',
    tools=[FileSearchTool(
        # Vector Store对象的ID
        vector_store_ids=['vs_xxxxxxxxxxxxxxx'],
        max_num_results=3, # 最大的搜索结果数量（可选）
        # 如果需要包含搜索结果的详情（分数、文本），请启用以下选项
        # include_search_results=True
        # 如果需要指定元数据过滤，请在下面写入
        # filters={...}
    )]
)

async def main():
    result = await Runner.run(
        agent,
        '请简要介绍一下与AI相关的文章'
    )
    print(result)
    
if __name__ == '__main__':
    asyncio.run(main())

from dataclasses import dataclass, asdict
from agents import Agent, RunContextWrapper, Runner, function_tool
import asyncio

@dataclass
class LoginUser:
    id: str

@dataclass
class Customer:
    id: str
    location: str
    name: str

@function_tool
def fetch_customer(ctx: RunContextWrapper[LoginUser], customer_id: str) -> str:
    """
    获取客户信息。

    Args:
        customer_id (str): 要获取的客户ID。
    Returns:
        str: 顧客情報(JSON)。
    """
    print(f'[audit]fetched by {ctx.context.id}')
    user = Customer(
        id=customer_id,
        location='東京都新宿区西新宿二丁目1番1号 新宿三井ビルディング34階',
        name='株式会社豆蔵'
    )
    return asdict(user)

{
  "name": "fetch_customer",
  "parameters": {
    "properties": {
      "customer_id": {
        "description": "要获取的客户ID。",
        "title": "Customer Id",
        "type": "string"
      }
    },
    "required": [
      "customer_id"
    ],
    "title": "fetch_customer_args",
    "type": "object",
    "additionalProperties": false
  },
  "strict": true,
  "type": "function",
  "description": "获取客户信息。"
}

agent = Agent(
    name='customer agent',
    instructions='请使用fetch_customer工具获取客户信息',
    # 添加自定义工具
    tools=[fetch_customer],
    model='gpt-4o-mini'
)

async def main():
    result = await Runner.run(
        agent,
        '请告诉我客户的地址。\nID:CUSTOMER_0001',
        # 在工作流程中共享的执行上下文
        context=LoginUser('MZ0001'))
    print(result)

if __name__ == '__main__':
    asyncio.run(main())

import asyncio
from pydantic import BaseModel
from agents import Agent, Runner

class TripPlan(BaseModel):
    """代理的输出schema"""
    departure_location: str
    destination: str
    days: int
    schedule: str
    budget: int

agent = Agent(
    name='trip planner',
    instructions='请根据指定的出发地、目的地和天数，以日语提出旅行计划',
    output_type=TripPlan,
)


async def main():
    result = await Runner.run(agent, '出発地:东京\n目的地:冲绳\n天数:3')
    # 使用final_output_as将输出转换为TripPlan
    print(result.final_output_as(TripPlan).model_dump_json(indent=2))

if __name__ == '__main__':
    asyncio.run(main())

from agents import Agent, RunContextWrapper, input_guardrail, GuardrailFunctionOutput

@input_guardrail
def validate_trip_input(context: RunContextWrapper[None], agent: Agent, user_input: str):
    """输入护栏"""
    days_match = re.search(r'天数[:：]\s*(\d+)', user_input)

    if not days_match:
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info='请务必正确指定天数！（例如：天数:3）'
        )
    return GuardrailFunctionOutput(
        tripwire_triggered=False,
        output_info='OK'
    )

agent = Agent(
    name='trip planner',
    instructions='请根据指定的出发地、目的地和天数，以日语提出旅行计划',
    output_type=TripPlan,
    # 指定输入护栏
    input_guardrails=[validate_trip_input]

from agents import InputGuardrailTripwireTriggered, Runner

async def main():
    try:
        result = await Runner.run(agent, '出発地:东京\n目的地:冲绳')
        print(result.final_output_as(TripPlan).model_dump_json(indent=2))
    except InputGuardrailTripwireTriggered as e:
        # 输入护栏违规
        print(e.guardrail_result.output)

from agents import Agent, InputGuardrailTripwireTriggered, OutputGuardrailTripwireTriggered, RunContextWrapper, Runner, input_guardrail, output_guardrail, GuardrailFunctionOutput

# 前略

@output_guardrail
def validate_trip_output(context: RunContextWrapper[None], agent: Agent, agent_output: TripPlan):
    """输出护栏"""
    if agent_output.budget > 200000:
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info=f'超出预算了！请再推荐一个更实惠的计划！预算:{agent_output.budget}'
        )
    return GuardrailFunctionOutput(
        tripwire_triggered=False,
        output_info='OK'
    )


agent = Agent(
    name='trip planner',
    instructions='请根据指定的出发地、目的地和天数，以日语提出旅行计划',
    output_type=TripPlan,
    # 指定输入护栏与输出护栏
    input_guardrails=[validate_trip_input],
    output_guardrails=[validate_trip_output]
)

async def main():
    try:
        # 故意使预算超标以触发输出护栏
        result = await Runner.run(agent, '出発地:东京\n目的地:纽约\n天数:14')
        print(result.final_output_as(TripPlan).model_dump_json(indent=2))
    except InputGuardrailTripwireTriggered as e:
        # 输入护栏违规
        print(e.guardrail_result.output)
    except OutputGuardrailTripwireTriggered as e:
        # 输出护栏违规
        print(e.guardrail_result.output)

if __name__ == '__main__':
    asyncio.run(main())

from pydantic import BaseModel

class Customer(BaseModel):
    """虚拟的客户信息（执行上下文）"""
    id: str
    name: str

from agents import Agent, RunContextWrapper, Runner, function_tool
from agents.extensions.handoff_prompt import prompt_with_handoff_instructions

@function_tool
def make_payment(ctx: RunContextWrapper[Customer], payment_info: str):
    print('[Payment Processing agent]: make_payment')
    return {
        'payment': 'ok',
        'details': payment_info,
        'name': ctx.context.name
    }

def payment_instructions(ctx: RunContextWrapper[Customer], agent: Agent) -> str:
    return prompt_with_handoff_instructions((
        '请根据预订信息完成支付。\n'
        '支付完成后，为了向客户提供信息，请输出预订信息和支付信息。\n'
        '此时请注意安全，屏蔽卡号。\n'
        f'客户ID/客户名称: {ctx.context.id} / {ctx.context.name}'
    ))

payment_agent = Agent(
    name='Payment Processing',
    # 使用执行上下文动态生成instructions（也可以指定函数）
    instructions=payment_instructions,
    tools=[make_payment]
)

@function_tool
def make_booking(ctx: RunContextWrapper[Customer], booking_info: str):
    print('[Booking Processing agent]: make_booking')
    return {
        'booking': 'ok',
        'hotel': booking_info,
        'customer_id': ctx.context.id
    }

booking_agent = Agent(
    name='Booking Processing',
    instructions=prompt_with_handoff_instructions(
        '请根据提供的用户信息进行酒店预订。'),
    tools=[make_booking],
    # 转交给支付处理代理
    handoffs=[payment_agent]
)

@function_tool
def check_availability(ctx: RunContextWrapper[Customer], user_input: str):
    print('[Availability Check agent]: check_availability')
    return {
        'availability': 'ok',
        'details': user_input
    }

availability_agent = Agent(
    name='Availability Check',
    instructions=prompt_with_handoff_instructions(
        '请根据用户的请求确认酒店的空房情况。'),
    tools=[check_availability],
    # 转交给预订处理代理
    handoffs=[booking_agent]
)

  {
    "name": "transfer_to_booking_processing_agent",
    "parameters": {
      "additionalProperties": false,
      "type": "object",
      "properties": {},
      "required": []
    },
    "strict": true,
    "type": "function",
    "description": "Handoff to the Booking Processing agent to handle the request. "
  }

async def main():
    user_input = (
        '我想在2025年6月1日预订东京的酒店。\n'
        '房型为单人间，要求无烟。\n'
        '请使用以下信息进行支付。\n'
        '卡号:4500-0000-0000-0000\n'
        '有效期(MM-YY):07-28\n'
        '安全码:123'
    )
    result = await Runner.run(
        availability_agent,
        user_input,
        context=Customer(id='MZ0001', name='豆蔵太郎') # 执行上下文（客户信息）
    )
    print(result)

if __name__ == '__main__':
    asyncio.run(main())

from agents import trace, Runner

async def main():
    customers = [
        Customer(id='CUSTOMER_0001', name='豆蔵太郎'),
        Customer(id='CUSTOMER_0002', name='豆蔵花子')
    ]
    with trace('豆蔵预订工作流程'): # 跟踪范围
        results = []
        for customer in customers:
            user_input = (
                '我想在2025年6月1日预订东京的酒店。\n'
                '房型为单人间，要求无烟。\n'
                '请使用以下信息进行支付。\n'
                '卡号:4500-0000-0000-0000\n'
                '有效期(MM-YY):07-28\n'
                '安全码:123'
            )

            result = await Runner.run(
                availability_agent,
                user_input,
                context=customer
            )
            results.append((customer, result))