【AIエージェント】swarmを使ったマルチエージェントの実例

pip install git+https://github.com/openai/swarm.git

from swarm import Swarm, Agent

os.environ['OPENAI_API_KEY'] = 'xxxxxxxxx'

client = Swarm()


def transfer_to_agent_b():
    return agent_b


agent_a = Agent(
    name="Agent A",
    instructions="あなたはAgent Aです。",
    functions=[transfer_to_agent_b],
)

agent_b = Agent(
    name="Agent B", instructions="あなたはAgent Bです。"
)

response = client.run(
    agent=agent_a,
    messages=[{"role": "user", "content": "Agent Bと話せますか？"}],
)

print(response.messages[-1]["content"])

## 各エージェントが使用する関数を定義
def search_perplexity(query: str) -> str:
    """2024年の時点の最新の情報をウェブ検索する関数"""
    messages = [
        {
            "role": "system",
            "content": (
                "You are an artificial intelligence assistant and you need to "
                "engage in a helpful, detailed, polite conversation with a user."
            ),
        },
        {
            "role": "user",
            "content": (
                query
            ),
        },
    ]
    client = OpenAI(api_key="xxxxx", base_url="https://api.perplexity.ai")

    response = client.chat.completions.create(
        model="llama-3.1-sonar-large-128k-online",
        messages=messages,
    )
    result = response.choices[0].message.content

    return result

def database_search_employee (query: str) -> str:
    """社員情報検索を実行する関数"""
    print(f"社員情報検索を実行します")

    # 本来は検索機能を実装するべきだか、ここではダミーデータを返す
    result = """Aさん
    - 生産管理システムの最適化
        - 生産ラインの稼働状況をリアルタイムで監視するダッシュボードの開発
        - AIを活用した生産計画の自動化と効率化
        - 現場スタッフ向けの使いやすいモバイルアプリの設計・導入
    
    Bさん
    - 製品検査プロセスのデジタル化
        - 画像認識技術を活用した自動検査システムの構築
        - データ解析による不良品の原因分析と予防策の提案
        - 検査結果の可視化ツールを用いた品質管理レポートの作成

    Cさん
    - IoTを活用した工場のスマート化推進
        - センサーデータを活用した設備稼働状況の予測分析
        - クラウドプラットフォームによる生産データの一元管理
        - 設備保全のためのAI予測保全モデルの導入"""
    
    return result

def database_search_design (query: str) -> str:
    """設計情報検索を実行する関数"""

    print(f"設計情報検索を実行します")
    
    # 本来は検索機能を実装するべきだか、ここではダミーデータを返す
    result ="""当社では、3D CADを活用した製品設計を中心に行っています。
    初期段階でのコンセプト設計から、詳細設計、試作品の製造までを一貫してサポート。
    CAE解析を用いて構造強度や熱伝導性のシミュレーションを実施し、製品性能を最適化しています。
    また、設計データはPDMシステムで一元管理し、設計変更やバージョン管理の効率化を図っています。
    これにより、品質向上と設計リードタイムの短縮を実現。顧客のニーズに合わせた高精度な設計を提供します。
    """

    return result

# 各エージェントの定義
def transfer_to_router_agent():
    return triage_agent

def transfer_to_web_search_agent():
    return Agent(
        name="Web Search Agent",
        instructions="あなたはWeb Search Agentです。最新の情報を検索して回答を生成します。",
        functions=[search_perplexity, transfer_to_router_agent],
    )


def transfer_to_employee_search_agent():
    return Agent(
        name="Employee Database Search Agent",
        instructions="あなたはEmployee Database Search Agentです。該当の技術スタックに精通する社員を検索し、その情報を提供してください。",
        functions=[database_search_employee, transfer_to_router_agent],
    )


def transfer_to_design_search_agent():
    return Agent(
        name="Design Database Search Agent",
        instructions="あなたはDesign Database Search Agentです。技術情報を検索し、その情報を提供してください。",
        functions=[database_search_design, transfer_to_router_agent],
    )

# トリアージエージェントの定義
triage_agent = Agent(
    name="Triage Agent",
    instructions="""あなたはTriage Agentです。ユーザーの質問に対して、適切なエージェントに転送します。

    ウェブ検索を行う場合は、transfer_to_web_search_agentを呼び出してください。
    社員情報に関する情報を聞く場合は、transfer_to_employee_search_agentを呼び出してください。
    設計情報を聞く場合は、transfer_to_design_search_agentを呼び出してください。
    それ以外の場合はあなたが回答してください。必要な情報があればユーザーに質問してください。
    """,
    functions=[
        transfer_to_web_search_agent,
        transfer_to_employee_search_agent,
        transfer_to_design_search_agent,
    ],
)

def invoke_router_agent(user_input: str):
    messages = [{"role": "user", "content": user_input}]
    response = client.run(
        agent=triage_agent,
        messages=messages,
        debug=True,
    )
    print(response.messages[-1]["content"])

invoke_router_agent("現在の日本の総理大臣は誰ですか？")

invoke_router_agent("社員Aの情報について教えてください？")

invoke_router_agent("当社で行っている設計情報について教えてください？")