Openai 新發布agent SDK輕量框架，方便開發通用AI整合，可做交接、搜尋、計算、串流媒體等等的分工。

python -m venv env
source env/bin/activate

pip install openai-agents

from agents import Agent, Runner

agent = Agent(name="Assistant", instructions="You are a helpful assistant")

result = Runner.run_sync(agent, "Write a haiku about recursion in programming.")
print(result.final_output)

# Code within the code,
# Functions calling themselves,
# Infinite loop's dance.

from agents import Agent, Runner
import asyncio

spanish_agent = Agent(
    name="Spanish agent",
    instructions="You only speak Spanish.",
)

english_agent = Agent(
    name="English agent",
    instructions="You only speak English",
)

triage_agent = Agent(
    name="Triage agent",
    instructions="Handoff to the appropriate agent based on the language of the request.",
    handoffs=[spanish_agent, english_agent],
)


async def main():
    result = await Runner.run(triage_agent, input="Hola, ¿cómo estás?")
    print(result.final_output)
    # ¡Hola! Estoy bien, gracias por preguntar. ¿Y tú, cómo estás?


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

import asyncio

from agents import Agent, Runner, function_tool


@function_tool
def get_weather(city: str) -> str:
    return f"The weather in {city} is sunny."


agent = Agent(
    name="Hello world",
    instructions="You are a helpful agent.",
    tools=[get_weather],
)


async def main():
    result = await Runner.run(agent, input="What's the weather in Tokyo?")
    print(result.final_output)
    # The weather in Tokyo is sunny.


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

from agents import Agent, AgentRunner

agent = Agent(
    name="basic_agent",
    instructions="You are a helpful agent."
)

result = await AgentRunner.run(agent, ["What's the capital of the USA?"])
print(result.agent_output)  # The capital of the United States is Washington, D.C.

from agents.tool import function_tool

@function_tool
def get_weather(location: str, unit: str = "C") -> str:
    """
    根據位置獲取天氣資訊。
    """
    return f"The weather in {location} is 22 degrees {unit}."

agent.tools.append(get_weather)

from agents.run_context import AgentContextWrapper
from agents.tool import function_tool

class MyContext:
  def __init__(self, user_id: str):
      self.user_id = user_id
      self.seen_messages = []

@function_tool
def greet_user(context: AgentContextWrapper[MyContext], greeting: str) -> str:
  user_id = context.agent_context.user_id
  return f"Hello {user_id}, you said: {greeting}"

from pydantic import BaseModel
from agents import Agent, AgentRunner

class WeatherResponse(BaseModel):
    location: str
    temperature_c: float
    summary: str

agent = Agent(
    name="weather_agent",
    instructions="請提供結構化的天氣資訊。",
    output_type=WeatherResponse,
)

result = await AgentRunner.run(agent, ["巴黎的天氣如何？"])
print(result.agent_output)

from agents import Agent, AgentRunner, handoff

english_agent = Agent(name="english_agent", instructions="回應使用英文。")
spanish_agent = Agent(name="spanish_agent", instructions="回應使用西班牙文。")

triage_agent = Agent(
    name="triage_agent",
    instructions="根據用戶語言選擇適當的代理。",
    handoffs=[handoff(english_agent), handoff(spanish_agent)],
)

result = await AgentRunner.run(triage_agent, ["Hola, ¿cómo estás?"])
print(result.agent_output)  # 會被交接給 spanish_agent。

from agents import Agent, AgentRunner

stream = AgentRunner.run_streamed(agent, ["請講一個故事"])

async for event in stream.stream_events():
    print(event.delta, end="", flush=True)
print("\n--- done ---")

from agents import AgentRunner, AgentRunConfig

config = AgentRunConfig(
    run_name="CustomerSupportFlow",
    tracing_disabled=False,
    trace_non_openai_generations=True,
)

result = await AgentRunner.run(my_agent, ["Hello?"], run_config=config)

from agents.guardrails import CustomGuardrail

async def is_not_swearing(msgs, context) -> bool:
    content = " ".join(m["content"] for m in msgs if "content" in m)
    return "badword" not in content.lower()

my_guardrail = CustomGuardrail(
    guardrail_function=is_not_swearing,
    tripwire_config=lambda output: not output  # 如果返回 False，則拋出錯誤
)

agent = Agent(
    name="safe_agent",
    input_guardrails=[my_guardrail]
)

out = await AgentRunner.run(agent, ["some text"])
# 若輸入含有 "badword"，則會觸發 GuardrailTripwireTriggered。