Vyhledávání na webu

Jak vytvořit paralelní agentní workflow pomocí Pythonu | Krok za krokem návod

Úvod

Velké jazykové modely mohou být obtížné používat v produkčním prostředí, protože mohou trpět nepřesnostmi, rozpory a zpožděním. Jak se modely zvětšují a stávají schopnějšími, je lákavé zahrnout do každého promptu více kontextu a složitější instrukce v naději, že je model zvládne, ale u jediného systémového promptu při použití relativně velkého modelu může zpoždění dosahovat 3–15 sekund. Pokud vytváříte časově citlivého agenta, například bota pro telefonní hovory, je to pro agenta příliš dlouhá pauza.

Možností s nižší latencí je rozdělit úkol na menší části a nechat vaše pracovní postupy LLM zpracovat každý díl paralelně, což usnadní správné zpracování i menšími modely. To vám umožní výrazně zlepšit latenci a v mnoha případech i náklady.

Existuje několik různých způsobů, jak spouštět agentní pracovní postupy paralelně, ale v mnoha agentních pracovních postupech je hlavní úzké hrdlo volání LLM. Toto úzké hrdlo můžete snížit tím, že provedete více současných volání LLM API pomocí knihoven asyncio a aiohttp v Pythonu.

V tomto tutoriálu použijete Python k vytváření agentních pracovních postupů, které provádějí více volání LLM asynchronně. To vám umožní vytvářet rychlejší a přesnější aplikace LLM, které lze nasadit na Droplet nebo bezserverovou instanci jako API endpoint.

Vytvoříte pracovní postup zákaznického servisu pro telefonní systém realitní kanceláře, který bude schopen odpovídat na dotazy ohledně nemovitostí, plánovat schůzky a spojovat volající s lidskými zástupci. Bude to dělat prostřednictvím několika asynchronních volání na různé LLM endpointy, čímž se sníží latence LLM a umožní větší flexibilita při výběru LLM pro každý úkol.

Hlavní body

  • Používání asynchronních volání na malé, rychlé modely vám může pomoci vytvářet efektivní workflow s nízkou latencí pro citlivé aplikace na čas.
  • Vícemodelové pracovní postupy vám umožňují přizpůsobit každý dotaz jinému modelu, což snižuje vaše celkové výpočetní nároky a snižuje náklady na vlastní provoz modelů LLM a použití API třetích stran

Předpoklady

K dokončení tohoto tutoriálu budete potřebovat:

  • Místní vývojové prostředí pro Python 3.
  • Přístup k API LLM. V tomto tutoriálu budete používat model Mistral-Small-3.2-24B nasazený na GPU Dropletu pomocí vLLM. GPU Droplet pro poskytování modelu pomocí vLLM si můžete nastavit podle pokynů v návodu Jak spustit GPT-OSS a vLLM na AMD GPU Dropletu s ROCm.
  • Znalost syntaxe Python async/await (volitelné, ale užitečné).

Krok 1 — Nastavení vašeho prostředí

Nejprve nastavte své prostředí a nainstalujte potřebné závislosti. V tomto tutoriálu budou volání směrována na model Mistral-Small-3.2-24B, který byl nasazen na H200 GPU Dropletu, ale budete muset buď volat svůj vlastní nasazený model, nebo API třetí strany LLM. Tento tutoriál volá pouze jeden LLM, ale pokud chcete, budete moci pro každý prompt použít jiný model od různých poskytovatelů.

Vytvořte virtuální prostředí a nainstalujte požadované závislosti:

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

Budete volat svůj model asynchronně, takže budete muset vytvořit některé asynchronní funkce, které tyto volání zvládnou. Budete používat knihovny asyncio a aiohttp v Pythonu. Pokud používáte jiné API pro LLM, budete muset nainstalovat jejich klienta, aby bylo možné spravovat volání na jejich koncový bod.

Vytvořte nový soubor s názvem agentic_workflows.py a přidejte následující kód pro import potřebných závislostí a definování API endpointu.

Nahraďte your_server_ip IP adresou vaší GPU Droplet instance nebo definujte klienta pro vaše LLM API:

import asyncio
import aiohttp

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

Krok 2 — Psání logiky asynchronního volání

Chcete, aby vaše asynchronní volání dokázala zpracovat seznam slovníků, z nichž každý obsahuje ID modelu a prompt, který chcete poslat LLM, a odeslala je všechny bez čekání na odpověď každého jednotlivého volání. Prompt bude obsahovat historii konverzace a systémový prompt pro toto konkrétní volání. Přidáte synchronní funkci call_models, kterou bude možné importovat a volat ve vašem hlavním skriptu.

Přidejte následující asynchronní funkce do 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))

Metoda aiohttp.ClientSession() vytváří asynchronní klientskou relaci, která odesílá POST požadavky na API endpoint a asynchronně zpracovává HTTP odpověď. To umožňuje provádět více volání LLM API současně, místo po jednom za sebou.

Funkce _call_models_async() shromažďuje všechny podrobnosti o více požadavcích, které poskytujeme, a provádí všechny volání API, aniž by čekala na každou odpověď. Poté, jakmile přicházejí odpovědi, přidává je do seznamu ve stejném pořadí, v jakém byly odeslány. Funkce call_models() je synchronní funkce, která usnadňuje volání asynchronních funkcí.

Krok 3 — Psaní vašich výzev

Teď, když máte základní funkčnost napsanou pro asynchronní volání do API, potřebujete napsat své výzvy. Budete mít tři samostatné systémové výzvy. Jedna vám pomůže s otázkami ohledně cen, jedna s plánováním a jedna s identifikací dalších otázek týkajících se seznamu a nasměrováním na lidského zástupce. Můžete přidat další, včetně těch s větší autonomií při rozhodování, ale prozatím tyto tři budou stačit.

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'."
    }
}

V těchto konfiguracích systémových pokynů zahrnujete název pokynu pro pozdější odkazování, ID modelu, kterému tento pokyn posíláte, a znění samotného systémového pokynu, ke kterému bude připojena historie konverzace.

Krok 4 — Zpracování uživatelského vstupu prostřednictvím modelů

Nyní musíte vytvořit funkci, která dokáže zpracovat uživatelův rozhovor, přidat do něj každý systémový prompt a provést vaše asynchronní volání na správný model pro každý prompt pomocí funkce call_models(), kterou jste napsali výše. Tento příklad předpokládá, že historie rozhovoru je uchovávána na front-endu ve stejném formátu, který naše API přijímá, a je odesílána spolu s nejnovějším vstupem uživatele.

Přidejte následující funkci do 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)

Tato funkce vytvoří seznam objektů slovníku pro každý systémový prompt, který obsahuje ID modelu a systémový prompt + historii konverzace. Poté zavolá asynchronní funkce pro volání každého modelu pro každý systémový prompt pomocí funkce call_models().

Po zavolání modelů jsou odpovědi přijaty a uloženy do proměnné prompt_responses ve stejném pořadí, v jakém byly odeslány, takže je potřeba přiřadit nové proměnné s aktuálními odpověďmi od modelů.

Pokračujte v přidávání do funkce 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")]

Krok 5 — Definování logiky

Nakonec je potřeba do této funkce přidat logickou vrstvu, která bude procházet vaše odpovědi a určovat, co dál. Můžete vyhledat cenu inzerátu ve vaší databázi, přesměrovat hovor na lidského zástupce, naplánovat schůzku s jedním z vašich realitních makléřů, nebo poslat uživateli doplňující otázku, aby poskytl více informací.

Pokračujte v přidávání do funkce run_agentic_workflow v agentic_workflows.py:

    # 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

Krok 6 — Testování vašeho pracovního postupu

Teď, když máte napsaný kód pro základní agentní pracovní postup, který zvládá tři kategorie otázek, otestujete ho v konverzaci o ceně inzerátu. V této konverzaci uživatel při své počáteční otázce neposkytne ID inzerátu a model odpoví doplňující otázkou, aby ho uživatel zadal. Uživatel ho poskytne a model odpoví správnou cenou.

Vytvořte nový soubor s názvem test_workflow.py a přidejte následující testovací kód:

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}")

Spusťte testovací skript:

python3 test_workflow.py

Uvidíte výstup podobný tomuto:

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

Nyní otestujte s úplným rozhovorem, kde uživatel poskytne ID inzerátu:

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}")

Spusťte skript znovu:

python3 test_workflow.py

Uvidíte informace o cenách:

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

Závěr

V tomto tutoriálu jste vytvořili paralelní agentní workflow od základu v Pythonu. Vytvořili jste asynchronní funkce pro současné volání LLM, nakonfigurovali více systémových výzev pro různé úkoly a implementovali logickou vrstvu směrování pro zpracování uživatelských požadavků.

Aby byl tento rámec připraven pro produkční použití, budete muset testovat, přepisovat příkazy, měnit modely a testovat znovu s různými konverzacemi, dokud nedosáhnete úrovně přesnosti a rychlosti, která splňuje vaše požadavky. U takovýchto pracovních postupů je potřeba každý příkaz testovat jednotlivě a průběžně jej vylepšovat. V našem příkladu jsme zvolili model Mistral-Small-3.2-24B kvůli jeho nízké latenci (medián odpovědí méně než 0,5 sekundy), schopnosti následovat tyto konkrétní pokyny a možnosti nasazení na jediný GPU. Testovali jsme také s modely GPT-OSS-120b, GPT-OSS-20b a Llama 3.1 8b a model Mistral podal nejlepší výkon pro tento konkrétní případ použití. Výběr správného modelu pro každý příkaz nebo úkol je iterativní proces.

Pokud máte zájem o použití agentů s větší autonomií, systémovým výzvám lze poskytnout více či méně autonomie tím, že jim dáte přístup k většímu množství kontextu nebo více nástrojům. Můžete dokonce provádět následné volání na LLM „spojené“ dohromady za účelem odvozování nebo jiných účelů. Nicméně, pokud výzva vyžaduje více volání API v řetězci, zvýší se tím latence pracovního postupu, protože již nečekáte pouze na jednu skupinu asynchronních volání na modely.

Nasadili jsme model Mistral s vLLM, takže jsme nemuseli měnit formát našich promptů, aby fungovaly pro různé modely. Nicméně OpenAI, Claude a open-source modely mohou mít odlišné požadavky na formátování promptů, které jsou zasílány jejich API. Pokud budete používat různé poskytovatele modelů pro každé volání, budete muset napsat Python funkce, které dokážou převést formát promptů tak, aby odpovídal požadavkům modelu, který voláte. Například můžete potřebovat funkci, která dokáže vzít prompt ve formátu vLLM a převést jeho syntaxi na prompt ve formátu Claude.

Často zjistíte, že některé modely zabírají mnohem více času než jiné, což vás nutí čekat na nejpomalejší odpověď, abyste získali všechny odpovědi. S knihovnou asyncio v Pythonu můžete zpracovávat odpovědi modelu, jakmile přicházejí, pomocí asyncio.as_completed() místo asyncio.gather(). Pak můžete zkontrolovat první odpovědi a zjistit, zda jsou cestou, kterou chcete pokračovat, a postupovat dál, aniž byste čekali na zbytek odpovědí. To by také vyžadovalo změny v logické vrstvě.

Při přidávání dalších systémových výzev narazíte hlavně na omezení rychlosti u API LLM nebo na limity zpracování u vašeho GPU. Nejlepším způsobem, jak se tomu vyhnout, je hostovat svůj vlastní model na GPU Dropletu, ale tyto problémy můžete také zmírnit zvýšením kvót pro rychlost API, rozložením volání mezi různé API a implementací logiky opakování, zpracování časového limitu a chyb pro případ, že jsou dosaženy limity rychlosti.

Děkujeme, že se učíte s komunitou Linux-Console.net.