サイト内検索

Pythonで並列エージェントワークフローを構築する方法｜ステップバイステップチュートリアル

はじめに

大規模言語モデルは、精度の欠如、一貫性のなさ、遅延などの問題があるため、実運用環境で使用するのは難しい場合があります。モデルが大きく、より高性能になるにつれて、より多くのコンテキストや複雑な指示を各プロンプトに含めたくなるかもしれませんが、単一のシステムプロンプトで比較的大きなモデルを使用すると、遅延は3～15秒程度になることがあります。もし電話会話ボットのような時間に敏感なエージェントを作ろうとしている場合、これではエージェントが一時停止するには長すぎます。

レイテンシを低くする方法の一つは、タスクを小さなタスクに分割し、LLMのワークフローでそれぞれを並行して処理させることです。これにより、小さなモデルでも正確に処理しやすくなります。これにより、レイテンシを大幅に改善でき、多くの場合、コストも削減できます。

エージェント的なワークフローを並行して実行する方法はいくつかありますが、多くのエージェント的なワークフローでは、主なボトルネックはLLMへの呼び出しです。このボトルネックは、Pythonのasyncioやaiohttpライブラリを使用して、LLM APIへの複数の同時呼び出しを行うことで軽減できます。

このチュートリアルでは、Pythonを使って複数のLLM呼び出しを非同期で行うエージェント的なワークフローを構築します。これにより、DropletやサーバーレスインスタンスにAPIエンドポイントとしてデプロイできる、より高速で精度の高いLLMアプリケーションを構築することが可能になります。

あなたは、不動産会社の電話システムのために、物件に関する質問に答えたり、予約をスケジュールしたり、通話者を担当者につなげたりするカスタマーサービスのワークフローを構築します。これは、複数の非同期呼び出しを異なるLLMエンドポイントに行うことで、LLMの遅延を減らし、各タスクに対してより柔軟にLLMを選択できるようにすることで実現されます。

重要ポイント

  • 小さくて高速なモデルに非同期呼び出しを使用することで、時間に敏感なアプリケーション向けに効果的で低遅延のエージェントワークフローを作成するのに役立ちます
  • マルチモデルのワークフローでは、各プロンプトを異なるモデル向けにカスタマイズできるため、全体の計算リソースを削減し、自己ホスティングのLLMモデルやサードパーティAPIの使用コストを低減できます

前提条件

このチュートリアルを完了するには、以下が必要です：

  • Python 3用のローカル開発環境。
  • LLM APIへのアクセス。このチュートリアルでは、vLLMを使用してGPU Droplet上にデプロイされたMistral-Small-3.2-24Bモデルを使用します。vLLMを使用してモデルを提供するGPU Dropletをセットアップするには、「How To Run GPT-OSS and vLLM on an AMD GPU Droplet with ROCm」に従ってください。
  • Pythonのasync/await構文に慣れていること（任意ですが役立ちます）。

ステップ1 — 環境の設定

まず、環境をセットアップし、必要な依存関係をインストールしてください。このチュートリアルでは、H200 GPU Droplet にデプロイされた Mistral-Small-3.2-24B モデルに呼び出しを行いますが、実際には自分でデプロイしたモデルに呼び出しを行うか、サードパーティの LLM API を使用する必要があります。このチュートリアルでは 1 つの LLM にしか呼び出しを行いませんが、希望すれば各プロンプトごとに異なるプロバイダーの異なるモデルを使用することも可能です。

仮想環境を作成し、必要な依存関係をインストールします:

python3 -m venv venv
source venv/bin/activate
pip install aiohttp

モデル呼び出しは非同期で行うことになるため、これらの呼び出しを管理できる非同期関数を作成する必要があります。Pythonではasyncioとaiohttpライブラリを使用します。異なるLLM APIを使用する場合は、そのエンドポイントへの呼び出しを管理するためにクライアントをインストールする必要があります。

新しいファイル agentic_workflows.py を作成し、必要な依存関係をインポートして API エンドポイントを定義するための次のコードを追加します。

your_server_ip を GPU ドロップレットのデプロイメントの IP アドレスに置き換えるか、LLM API 用のクライアントを定義してください:

import asyncio
import aiohttp

# Your droplet's vLLM server URL
VLLM_SERVER_URL = "http://your_server_ip:8000/v1/chat/completions"

ステップ2 — 非同期呼び出しロジックの作成

非同期呼び出しで、各辞書にモデルIDと送信したいプロンプトが含まれているリストを受け取り、各呼び出しの個別の応答を待たずにすべて送信できるようにしたいのです。プロンプトには、その特定の呼び出しに対する会話履歴とシステムプロンプトが含まれます。同期的に呼び出せる call_models 関数を追加し、それをメインスクリプトでインポートして呼び出せるようにします。

agentic_workflows.py に次の非同期関数を追加してください:

async def _call_single_model(call_spec):
    """Make an async call to the Digital Ocean GPU droplet with a given model and messages"""
    model_id = call_spec["model_id"]
    messages = call_spec["messages"]
    
    payload = {
        "model": model_id,
        "messages": messages,
        "max_tokens": 100, 
        "temperature": 0.1 
    }
    
    async with aiohttp.ClientSession() as session:
        async with session.post(VLLM_SERVER_URL, json=payload) as response:
            result = await response.json()
            
            message = result["choices"][0]["message"]["content"]
            return message

async def _call_models_async(call_list):
    """Call multiple models asynchronously and return responses in the same order"""
    # Create tasks for all model calls
    tasks = [_call_single_model(call_spec) for call_spec in call_list]
    
    # Run all tasks concurrently and return responses in order
    responses = await asyncio.gather(*tasks)
    return responses

def call_models(call_list):
    return asyncio.run(_call_models_async(call_list))

aiohttp.ClientSession() メソッドは、APIエンドポイントにPOSTリクエストを送信し、HTTPレスポンスを非同期に処理する非同期クライアントセッションを作成します。これにより、複数のLLM APIコールを順次ではなく同時に行うことが可能になります。

_call_models_async() 関数は、私たちが提供する複数のリクエストのすべての詳細を収集し、各レスポンスを待たずにすべてのAPI呼び出しを行います。そして、レスポンスが戻ってくると、それらを送信された順序と同じ順序でリストに追加します。call_models() 関数は同期関数であり、非同期呼び出しをより簡単に実行できるようにします。

ステップ3 — プロンプトを書く

基本的な機能が書けてAPIエンドポイントへの非同期呼び出しができるようになったので、次はプロンプトを書かなければなりません。システムプロンプトは3種類用意します。1つは価格に関する質問を支援するもの、1つはスケジュールに関するもの、そしてもう1つはその他の物件に関する質問を特定し、人間の担当者に案内するものです。さらに追加することも可能で、より自律的な意思決定ができるものも作れますが、まずはこの3つで十分です。

SYSTEM_PROMPT_CONFIGURATIONS = {
    "pricing_prompt": {
        "model_id": "mistralai/Mistral-Small-3.2-24B-Instruct-2506", 
        "prompt": "You are a customer service agent. Determine if the user's most recent request is asking about the price of a listing. If they are asking about the price of a listing AND if they have included the listing_id, return only the listing_id of the item they are asking about in the following format: 'listing_id: XXXXXX'. \nIf they are asking about the pricing of a listing AND did NOT mention the specific listing_id number, ask them for the listing id number. If they are requesting something other than the price of a listing: return only the word 'false'."
    },
    "scheduling_prompt": {
        "model_id": "mistralai/Mistral-Small-3.2-24B-Instruct-2506", 
        "prompt": "You are a customer service agent. Determine if the user's most recent request is asking to schedule a call with a real estate agent. If they are trying to schedule a call, return the date and time they would like to schedule the call in the following format: 'date: YYYY-MM-DD, time: HH:MM'. If they are trying to schedule a call but they did not mention the specific date and time they are available, ask them what day and time they are available. Do not provide any additional information, such as the agent's availability. If they are not asking to schedule a call, return only the word 'false'." 
    }, 
    "listing_prompt": {
        "model_id": "mistralai/Mistral-Small-3.2-24B-Instruct-2506", 
        "prompt": "You are a customer service agent. Determine if the user is asking a question about a listing. If they are asking a question about a listing, return only the word 'true'. Otherwise, return only the word 'false'."
    }
}

これらのシステムプロンプトの設定では、後で参照するためのプロンプトのタイトル、送信するプロンプトのモデルID、そして会話履歴が添付されるシステムプロンプト自体の文言を含めます。

ステップ4 — モデルを通じたユーザー入力の処理

次に、ユーザーの会話を取得し、各システムプロンプトを追加し、上で作成した call_models() 関数を使用して各プロンプトに対して正しいモデルへの非同期呼び出しを行う関数を作成する必要があります。この例では、会話の履歴がフロントエンドで当社のAPIが受け入れる形式と同じ形式で保存され、ユーザーの最新の入力と一緒に送信されることを前提としています。

agentic_workflows.py に次の関数を追加してください:

def run_agentic_workflow(conversation_history):
    model_calls_list = []
    prompt_names = []  # Keep track of prompt order for response mapping
    
    for prompt_name, config in SYSTEM_PROMPT_CONFIGURATIONS.items():
        model_id = config["model_id"]
        system_prompt = config["prompt"]
        
        # Construct the full message prompt: system prompt + conversation history
        full_messages = [{"role": "system", "content": system_prompt}] + conversation_history
        
        model_calls_list.append({
            "model_id": model_id,
            "messages": full_messages
        })
        prompt_names.append(prompt_name) 
    
    prompt_responses = call_models(model_calls_list)

この関数は、各システムプロンプトに対してモデルIDとシステムプロンプト＋会話履歴を含む辞書オブジェクトのリストを作成します。その後、非同期関数を呼び出して、call_models() 関数を使用して各システムプロンプトごとに各モデルを呼び出します。

モデルを呼び出した後、応答は受信され、送信した順序で prompt_responses 変数に保存されます。そのため、モデルからの実際の応答を新しい変数に割り当てる必要があります。

run_agentic_workflow 関数への追加を続けてください:

    # Map responses to their respective prompts
    pricing_response = prompt_responses[prompt_names.index("pricing_prompt")]
    scheduling_response = prompt_responses[prompt_names.index("scheduling_prompt")]
    listing_response = prompt_responses[prompt_names.index("listing_prompt")]

ステップ5 — ロジックの定義

最後に、この関数にロジック層を追加して、レスポンスを確認し、次に何をすべきかを判断できるようにする必要があります。データベースで物件の価格を調べたり、電話を人間の担当者に転送したり、不動産エージェントとの予約をスケジュールしたり、ユーザーに詳細を尋ねるフォローアップの質問を送ることができます。

agentic_workflows.py の run_agentic_workflow 関数に追加を続けてください:

    # Route 1: Handle pricing inquiries
    if pricing_response.lower() != "false":
        if pricing_response.startswith("listing_id:"):
            # Extract listing ID from response
            listing_id = pricing_response.split("listing_id: ")[1].strip()
            
            # Simulate database/API lookup
            # You can add database querying logic here to look for a target listing ID. For our example, we will use a simple Python dictionary
            example_price_database = {
                "123456": "$350,000",
                "654321": "$450,000", 
                "112233": "$550,000"
            }
            
            found_price = example_price_database.get(listing_id)
            if found_price:
                final_response = f"The price for listing {listing_id} is {found_price}."
            else:
                final_response = "We are unable to find that listing ID in our records. Are you sure you have the correct listing ID?"
        else:
            final_response = pricing_response  # Response asking for listing ID
    
    # Route 2: Handle scheduling requests
    elif scheduling_response.lower() != "false":
        if scheduling_response.startswith("date:"):
            final_response = f"Perfect! I've scheduled a call for you on that date and time. A sales representative will reach out to you at that time."
            # In production: Add logic to actually book the appointment, and consider customizing the message to confirm the date and time selected.
        else:
            final_response = scheduling_response  # Response asking for specific time
    
    # Route 3: Handle general listing questions  
    elif listing_response.lower() != "false":
        final_response = "Please hold while I transfer you to a specialist for further assistance."
        # In production: Add logic to transfer chat to human representative
        # You could alternatively add logic to access listing details and answer the user's specific question

    else:
        final_response = "I apologize, I'm not sure how I can help with that. Let me transfer you to a human representative who can better assist you."
        # In production: Add logic to transfer chat to human representative
    
    return final_response

ステップ6 — ワークフローのテスト

基本的なエージェントワークフローを3つのカテゴリーの質問に対応できるようコードを書いたので、次にリスティングの価格に関する会話でテストします。この会話では、ユーザーは最初の質問でリスティングIDを提供せず、モデルはそのIDを求めるフォローアップの質問で応答します。ユーザーがIDを提供すると、モデルは正しい価格で応答します。

test_workflow.pyという新しいファイルを作成し、以下のテストコードを追加してください:

from agentic_workflows import run_agentic_workflow

conversation_history = [
    {"role": "user", "content": "Hi, can you tell me the price of one of your listings?"},
]
final_response = run_agentic_workflow(conversation_history)
print(f"Response: {final_response}")

テストスクリプトを実行します:

python3 test_workflow.py

次のような出力が表示されます。

Response: Could you please provide the listing_id of the item you're asking about?

では、ユーザーがリスティングIDを提供する完全な会話でテストしてみましょう:

conversation_history = [
    {"role": "user", "content": "Hi, can you tell me the price of one of your listings?"},
    {"role": "assistant", "content": "Could you please provide the listing_id of the item you're asking about?"},
    {"role": "user", "content": "Yes, the listing ID is 123456"},
]
final_response = run_agentic_workflow(conversation_history)
print(f"Response: {final_response}")

もう一度スクリプトを実行してください:

python3 test_workflow.py

料金情報が表示されます：

Response: The price for listing 123456 is $350,000.

結論

このチュートリアルでは、Pythonで最初から並列のエージェントワークフローを構築しました。複数のLLM呼び出しを同時に行う非同期関数を作成し、異なるタスク用に複数のシステムプロンプトを設定し、ユーザーのリクエストを処理するルーティングロジック層を実装しました。

このフレームワークを本番運用可能な状態にするには、テストを行い、プロンプトを書き直し、モデルを変更し、異なる会話で再度テストを行い、要求に見合った精度と速度を達成するまで繰り返す必要があります。このようなワークフローでは、各プロンプトを個別にテストし、時間をかけて改善を続ける必要があります。我々の例では、低遅延（中央値の応答時間が0.5秒未満）で、特定の指示に従う能力があり、単一GPUにデプロイ可能であることから、Mistral-Small-3.2-24Bモデルを選びました。また、GPT-OSS-120b、GPT-OSS-20b、Llama 3.1 8bでもテストを行い、この特定のユースケースではMistralモデルが最も優れた性能を示しました。各プロンプトやタスクに適したモデルを選ぶことは、反復的なプロセスです。

もしより自律的なエージェントの使用に興味がある場合、システムプロンプトにより多くのコンテキストやツールへのアクセスを与えることで、その自律性を高めたり低めたりすることができます。また、理由付けやその他の目的のために、LLMへのフォローアップ呼び出しを「チェーン」としてつなげることも可能です。しかし、プロンプトが複数のAPI呼び出しをチェーンで行う必要がある場合、もはや1つの非同期呼び出しグループを待つだけではないため、ワークフローの遅延が増加します。

私たちは vLLM で Mistral モデルをデプロイしたため、異なるモデルで動作させるためにプロンプトの形式を変更する必要はありませんでした。しかし、OpenAI、Claude、そしてオープンソースのモデルでは、それぞれの API に送信されるプロンプトの形式に異なる要件がある場合があります。もし呼び出すたびに異なるモデル提供者を使用する場合は、プロンプトの形式を呼び出すモデルの要件に合わせて変換できる Python 関数を書く必要があります。例えば、vLLM 形式のプロンプトを Claude 形式のプロンプトに変換する関数が必要になるかもしれません。

いくつかのモデルは他のモデルよりもはるかに時間がかかることがあり、すべてのレスポンスを受け取るために最も遅いレスポンスを待たなければならないことがあります。Python の asyncio ライブラリを使用すると、asyncio.gather() の代わりに asyncio.as_completed() を使って、モデルのレスポンスを届き次第処理することができます。そうすれば、最初に届いたレスポンスを確認して希望する進行ルートかどうかを判断し、他のレスポンスを待たずに先に進むことができます。これにはロジック層の変更も必要です。

より多くのシステムプロンプトを追加する際に直面する主なボトルネックは、LLM APIのレート制限やGPUでの処理制限です。これを回避する最良の方法は、GPUのDroplet上で自分のモデルをホストすることですが、APIのレート制限のクォータを増やしたり、呼び出しを複数のAPIに分散させたり、レート制限が発生した場合のリトライロジック、タイムアウト処理、エラーハンドリングを実装することで、これらの問題を緩和することも可能です。

Linux-Console.netコミュニティで学んでいただきありがとうございます。