OpenAI Swarm入門 - 軽量マルチエージェントフレームワークの実践

from swarm import Swarm, Agent

client = Swarm()

# シンプルなエージェント
agent = Agent(
    name="Assistant",
    instructions="あなたは親切なアシスタントです。"
)

response = client.run(
    agent=agent,
    messages=[{"role": "user", "content": "こんにちは"}]
)
print(response.messages[-1]["content"])

def transfer_to_sales():
    """営業チームにハンドオフ"""
    return sales_agent

# トリアージエージェント
triage_agent = Agent(
    name="Triage",
    instructions="顧客の質問を分類し、適切な担当にエスカレーション",
    functions=[transfer_to_sales]
)

# 営業エージェント
sales_agent = Agent(
    name="Sales",
    instructions="製品の販売とプライシングを担当"
)

response = client.run(
    agent=triage_agent,
    messages=[{"role": "user", "content": "製品の価格を知りたい"}]
)

from swarm import Agent

def get_weather(location: str, context_variables: dict) -> str:
    """天気情報を取得"""
    user_id = context_variables.get("user_id")
    # APIから天気を取得
    return f"{location}の天気は晴れです"

agent = Agent(
    name="WeatherBot",
    functions=[get_weather]
)

response = client.run(
    agent=agent,
    messages=[{"role": "user", "content": "Tokyoの天気は？"}],
    context_variables={"user_id": "user_123"}
)

from swarm import Swarm, Agent

client = Swarm()

# サポートチームエージェント
def escalate_to_human():
    print("人間のオペレーターにエスカレーション")
    return human_agent

def transfer_to_technical():
    return technical_agent

general_agent = Agent(
    name="GeneralSupport",
    instructions="""
    一般的な質問に回答します。
    - 技術的な問題 → transfer_to_technical()
    - 解決できない → escalate_to_human()
    """,
    functions=[transfer_to_technical, escalate_to_human]
)

technical_agent = Agent(
    name="TechnicalSupport",
    instructions="技術的な問題を解決します",
    functions=[escalate_to_human]
)

human_agent = Agent(
    name="HumanOperator",
    instructions="複雑な問題を人間が対応"
)

# 実行
response = client.run(
    agent=general_agent,
    messages=[
        {"role": "user", "content": "ログインできません"}
    ]
)

# ステップ1: リサーチ
researcher = Agent(
    name="Researcher",
    instructions="与えられたトピックについて情報を収集",
    functions=[web_search]
)

# ステップ2: ライティング
def hand_off_to_writer(research_data: str):
    return writer_agent

researcher.functions.append(hand_off_to_writer)

writer_agent = Agent(
    name="Writer",
    instructions="リサーチ結果を元に記事を作成"
)

# 実行
response = client.run(
    agent=researcher,
    messages=[{"role": "user", "content": "AIエージェントについて記事を書いて"}]
)

def route_to_specialist(query: str, context_variables: dict):
    """クエリの内容に応じて専門家を選択"""
    if "価格" in query or "料金" in query:
        return pricing_agent
    elif "技術" in query or "実装" in query:
        return technical_agent
    else:
        return general_agent

router = Agent(
    name="Router",
    instructions="質問を分析し、適切な専門家にルーティング",
    functions=[route_to_specialist]
)

pricing_agent = Agent(name="PricingExpert", instructions="料金とプランについて回答")
technical_agent = Agent(name="TechExpert", instructions="技術的な質問に回答")
general_agent = Agent(name="GeneralExpert", instructions="一般的な質問に回答")

response = client.run(
    agent=router,
    messages=[{"role": "user", "content": "エンタープライズプランの価格は？"}]
)

from swarm import Swarm, Agent

def transfer_to_flight_modification():
    return flight_modification

def transfer_to_flight_cancel():
    return flight_cancel

def transfer_to_triage():
    """トリアージに戻る"""
    return triage_agent

# トリアージエージェント
triage_agent = Agent(
    name="TriageAgent",
    instructions="""
    顧客の問い合わせを分類し、適切な担当にエスカレーション:
    - フライト変更 → transfer_to_flight_modification()
    - キャンセル → transfer_to_flight_cancel()
    """,
    functions=[transfer_to_flight_modification, transfer_to_flight_cancel]
)

# フライト変更エージェント
flight_modification = Agent(
    name="FlightModification",
    instructions="フライトの日時変更をサポート",
    functions=[transfer_to_triage]
)

# キャンセルエージェント
flight_cancel = Agent(
    name="FlightCancellation",
    instructions="フライトのキャンセル手続きをサポート",
    functions=[transfer_to_triage]
)

client = Swarm()

# 会話シミュレーション
messages = []
agent = triage_agent

user_queries = [
    "フライトを明日に変更したい",
    "変更手数料はいくらですか？",
    "ありがとうございます"
]

for query in user_queries:
    messages.append({"role": "user", "content": query})
    
    response = client.run(agent=agent, messages=messages)
    
    messages.extend(response.messages)
    agent = response.agent
    
    print(f"User: {query}")
    print(f"Agent ({agent.name}): {response.messages[-1]['content']}\n")