Hogyan építsünk párhuzamos agentikus munkafolyamatokat Pythonban | Lépésről lépésre útmutató

Bevezetés

A nagy nyelvi modelleket nehéz lehet termelési környezetben használni, mivel pontatlanságokkal, következetlenségekkel és késleltetéssel küzdhetnek. Ahogy a modellek egyre nagyobbak és képzettebbek lesznek, csábító lehet több kontextust és összetettebb utasításokat belefoglalni minden promptba abban a reményben, hogy a modell képes lesz kezelni ezeket, de egyetlen rendszerprompt esetén, relatíve nagy modell használatával, a késleltetés 3-15 másodperc között változhat. Ha időérzékeny ügynököt építesz, például egy telefonos beszélgetés botot, ez túl hosszú, hogy az ügynök megálljon.

A kisebb késleltetésű lehetőség az, ha a feladatot kisebb részekre bontjuk, és az LLM-folyamatainkat minden rész párhuzamos kezelésére használjuk, így a kisebb modellek is helyesen tudják kezelni. Ez lehetővé teszi, hogy drasztikusan javítsuk a késleltetést, és sok esetben a költségeket is.

Az ügynöki munkafolyamatok párhuzamos futtatásának többféle módja van, de sok ügynöki munkafolyamatnál a fő szűk keresztmetszet a LLM hívása. Ezt a szűk keresztmetszetet csökkentheted, ha több, egyidejű hívást hajtasz végre a LLM API-khoz a Python asyncio és aiohttp könyvtárainak használatával.

Ebben az oktatóanyagban Python segítségével fogsz ügynöki munkafolyamatokat építeni, amelyek több LLM-hívást hajtanak végre aszinkron módon. Ez lehetővé teszi, hogy gyorsabb és pontosabb LLM-alkalmazásokat készíts, amelyeket egy Dropletre vagy szerver nélküli példányra API végpontként telepíthetsz.

Készíteni fogsz egy ügyfélszolgálati munkafolyamatot egy ingatlanügynökség telefonrendszeréhez, amely képes válaszolni a listázásokkal kapcsolatos kérdésekre, időpontokat egyeztetni, és a hívókat élő ügyfélszolgálati munkatársakhoz kapcsolni. Ezt úgy fogja megvalósítani, hogy több aszinkron hívást hajt végre különböző LLM végpontokra, csökkentve ezzel a LLM késleltetését, és lehetővé téve nagyobb rugalmasságot az egyes feladatokhoz választott LLM tekintetében.

Főbb tanulságok

Kisebb, gyors modellek aszinkron hívásának használata segíthet hatékony, alacsony késleltetésű ügynöki munkafolyamatok létrehozásában időérzékeny alkalmazásokhoz.
A többmodellű munkafolyamatok lehetővé teszik, hogy minden kérést egy-egy külön modellhez igazíts, csökkentve az összes számítási igényt és az önállóan hosztolt LLM-modellek, valamint harmadik fél API-használatának költségeit.

Előfeltételek

A bemutató befejezéséhez szükséged lesz:

Egy helyi fejlesztői környezet Python 3-hoz.
Hozzáférés egy LLM API-hoz. Ebben a bemutatóban a Mistral-Small-3.2-24B modellt fogja használni, amely egy GPU Dropleten van telepítve a vLLM segítségével. Beállíthat egy GPU Dropletet a vLLM használatával a modell kiszolgálására, ha követi a Hogyan futtassunk GPT-OSS-t és vLLM-et egy AMD GPU Dropleten ROCm-mel című útmutatót.
Ismeret a Python async/await szintaxisával kapcsolatban (opcionális, de hasznos).

1. lépés — A környezet beállítása

Először állítsd be a környezetedet, és telepítsd a függőségeidet. Ebben a bemutatóban egy Mistral-Small-3.2-24B modellhez történnek hívások, amely egy H200 GPU Dropleten lett telepítve, de neked vagy a saját telepített modelljeidhez, vagy egy harmadik fél LLM API-jához kell hívásokat intézned. Ez a bemutató csak egy LLM-hez intéz hívásokat, de ha szeretnéd, minden prompthoz különböző szolgáltatók különböző modelljeit is használhatod.

Hozz létre egy virtuális környezetet, és telepítsd a szükséges függőségeket:

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

Az API-hívásokat aszinkron módon fogod végezni, ezért létre kell hoznod néhány aszinkron függvényt, amelyek képesek kezelni ezeket a hívásokat. Pythonban az asyncio és az aiohttp könyvtárakat fogod használni. Ha egy másik LLM API-t használsz, telepítened kell a kliensüket a végpontjukhoz történő hívások kezeléséhez.

Hozz létre egy új fájlt agentic_workflows.py néven, és add hozzá a következő kódot a szükséges függőségek importálásához és az API végpont definiálásához.

Cseréld le a your_server_ip-t a GPU Droplet telepítésed IP-címére, vagy definiáld a klienst az LLM API-dhoz:

import asyncio
import aiohttp

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

2. lépés — Az aszinkron hívás logikájának megírása

Azt szeretnéd, hogy az aszinkron hívásaid képesek legyenek egy olyan szótárakból álló listát fogadni, amelyben mindegyik tartalmazza a modell azonosítóját és a promptot, amit az LLM-nek küldeni szeretnél, és mindet elküldeni anélkül, hogy megvárnád az egyes hívások egyéni válaszát. A prompt tartalmazni fogja a beszélgetés előzményeit és az adott híváshoz rendelt rendszer promptot. Hozzáadsz egy szinkron call_models funkciót, amely importálható és meghívható a fő scriptedben.

Add hozzá a következő aszinkron függvényeket a agentic_workflows.py fájlhoz:

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

A aiohttp.ClientSession() metódus létrehoz egy aszinkron kliens munkamenetet, amely POST kéréseket küld az API végpontra, és aszinkron módon kezeli a HTTP választ. Ez lehetővé teszi, hogy több LLM API hívás párhuzamosan, egymás után történő végrehajtás helyett történjen.

A _call_models_async() függvény összegyűjti a megadott több kérés részleteit, és az összes API-hívást végrehajtja anélkül, hogy minden válaszra várna. Ezután, ahogy a válaszok visszaérkeznek, hozzáadja őket egy listához ugyanabban a sorrendben, ahogyan elküldték őket. A call_models() függvény egy szinkron függvény, amely könnyebbé teszi az aszinkron hívások meghívását.

3. lépés — A promptok írása

Most, hogy megírtad az alapvető funkciókat az API végpont aszinkron hívásainak kezelésére, meg kell írnunk a promptjaidat. Három külön rendszerpromptod lesz. Az egyik az árkérdéseknél segít, a másik az ütemezéssel foglalkozik, a harmadik pedig más listázással kapcsolatos kérdéseket azonosít és emberi képviselőhöz irányít. Továbbiakat is hozzáadhatsz, beleértve olyanokat is, amelyek nagyobb autonóm döntési jogkörrel rendelkeznek, de ezek a hármak egyelőre működni fognak.

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

Ezekben a rendszerparancs-beállításokban megadod a parancs címét későbbi hivatkozás céljából, a modell azonosítóját, amelyhez a parancsot küldöd, valamint magát a rendszerparancs szövegét, amelyhez a beszélgetés előzményei csatolva lesznek.

4. lépés — Felhasználói bemenet feldolgozása a modelleken keresztül

Most létre kell hoznod egy függvényt, amely képes a felhasználó beszélgetését átvenni, hozzáadni minden rendszer üzenetet, és aszinkron hívásokat végrehajtani a megfelelő modellhez az előzőleg írt call_models() függvény segítségével. Ez a példa feltételezi, hogy a beszélgetési előzményeket az előtérben ugyanabban a formátumban tárolják, ahogyan az API-nk fogadja, és a felhasználó legutóbbi bemenetével együtt küldik el.

Add hozzá a következő függvényt a agentic_workflows.py fájlhoz:

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)

Ez a függvény létrehoz egy szótár objektumokból álló listát minden egyes rendszerüzenethez, amely tartalmazza a modellazonosítót és a rendszerüzenetet + a beszélgetés előzményeit. Ezután meghívja az aszinkron függvényeket, hogy minden modellhez minden rendszerüzenet esetén meghívja a call_models() függvényt.

Miután meghívtuk a modelleket, a válaszok megérkeznek és elmentődnek a prompt_responses változóba ugyanabban a sorrendben, ahogyan elküldtük őket, így új változókat kell létrehozni a modellektől kapott tényleges válaszok tárolására.

Folytassa a run_agentic_workflow függvény bővítését:

    # 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. lépés — A logika meghatározása

Végül hozzá kell adnod egy logikai réteget ehhez a funkcióhoz, amely képes átnézni a válaszaidat és meghatározni, mi legyen a következő lépés. Lekérdezheted az árlistát az adatbázisodból, átirányíthatod a hívást egy emberi képviselőhöz, időpontot egyeztethetsz az egyik ingatlanügynököddel, vagy küldhetsz egy követő kérdést a felhasználónak további részletek kérése céljából.

Folytasd a run_agentic_workflow függvény bővítését a agentic_workflows.py fájlban:

    # 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. lépés — A munkafolyamat tesztelése

Most, hogy megírtad az alapvető ügynöki munkafolyamat kódját, amely három kategóriájú kérdés kezelésére alkalmas, tesztelni fogod egy beszélgetéssel a hirdetés áráról. Ebben a beszélgetésben a felhasználó nem adja meg a hirdetés azonosítóját a kezdeti kérdésében, és a modell egy további kérdéssel kéri azt. A felhasználó megadja, és a modell a helyes árral válaszol.

Hozz létre egy új fájlt test_workflow.py néven, és add hozzá a következő tesztkódot:

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

Futtasd a teszt szkriptet:

python3 test_workflow.py

Ehhez hasonló kimenetet fogsz látni:

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

Most próbáld ki egy teljes beszélgetéssel, ahol a felhasználó megadja a listázási azonosítót:

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

Futtasd újra a szkriptet:

python3 test_workflow.py

Látni fogja az árakat:

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

Következtetés

Ebben az útmutatóban nulláról építettél fel egy párhuzamos, ügynöki munkafolyamatot Pythonban. Aszinkron függvényeket hoztál létre egyszerre történő LLM-hívásokhoz, több rendszerpromptot konfiguráltál különböző feladatokhoz, és implementáltál egy irányítási logikai réteget a felhasználói kérések kezelésére.

Ahhoz, hogy ezt a keretrendszert éles használatra alkalmassá tedd, tesztelned kell, újra kell írnod a promptokat, modelleket kell cserélned, majd különböző beszélgetésekkel újra tesztelned kell, amíg el nem éred a kívánt pontosságot és sebességet. Az ilyen munkafolyamatoknál minden promptot egyénileg kell tesztelni, és idővel folyamatosan javítani kell azokat. Példánkban a Mistral-Small-3.2-24B modellt választottuk az alacsony késleltetése miatt (középidős válaszidő kevesebb, mint 0,5 másodperc), a konkrét utasítások követésének képessége, valamint az egy GPU-ra történő telepíthetősége miatt. Teszteltünk továbbá a GPT-OSS-120b, GPT-OSS-20b és Llama 3.1 8b modellekkel is, és a Mistral modell teljesített legjobban ebben az adott esetben. Minden prompthoz vagy feladathoz a megfelelő modell kiválasztása iteratív folyamat.

Ha érdekel az önállóbb ügynökök használata, a rendszerparancsoknak adható többé-kevésbé autonómia azáltal, hogy hozzáférést kapnak több kontextushoz vagy több eszközhöz. Akár követő hívásokat is végezhetsz az LLM‑hez, amelyeket „láncolt” módon kapcsolsz össze az érvelés vagy más célok érdekében. Azonban ha egy parancs több API-hívást igényel láncban, ez növeli a munkafolyamat késleltetését, mert már nem csak egy aszinkron híváscsoport eredményére vársz a modellektől.

A Mistral modellt a vLLM-mel telepítettük, így nem kellett megváltoztatnunk a promptjaink formátumát a különböző modellekhez. Azonban az OpenAI, Claude és a nyílt forráskódú modellek eltérő követelményeket támaszthatnak a promptok API-jukhoz történő küldésének formátumára. Ha minden híváshoz más-más modell szolgáltatót használ, Python függvényeket kell írnia, amelyek képesek a promptok formátumát átalakítani az éppen hívott modell követelményeinek megfelelően. Például szüksége lehet egy olyan függvényre, amely képes egy vLLM-formátumú promptot Claude-formátumúvá alakítani.

Gyakran tapasztalhatod, hogy egyes modellek sokkal tovább tartanak, mint mások, ami azt eredményezi, hogy a leglassabb válaszra kell várnod, hogy megkapd az összes választ. A Python asyncio könyvtárának segítségével feldolgozhatod a modellválaszokat, ahogyan érkeznek, a asyncio.as_completed() használatával a asyncio.gather() helyett. Így először megnézheted az első válaszokat, hogy azok megfelelnek-e az útnak, amin haladni szeretnél, és folytathatod a munkát anélkül, hogy a többi válaszra várnál. Ehhez a logikai réteget is módosítani kellene.

Amikor további rendszerüzeneteket adsz hozzá, a fő szűk keresztmetszet, amibe itt ütközni fogsz, az a LLM API korlátozása vagy a GPU-d feldolgozási kapacitásának határa. A legjobb módja ennek áthidalására, ha a saját modelljeidet egy GPU Dropleten futtatod, de ezek a problémák csökkenthetők az API-korlát kvóták növelésével, a hívások szétterítésével az API-k között, valamint újrapróbálkozási logika, időkorlát-kezelés és hibakezelés bevezetésével, amikor elérik a korlátozásokat.

Köszönjük, hogy a Linux-Console.net közösséggel tanultál.