Kuinka rakentaa rinnakkaisia agenttipohjaisia työnkulkuja Pythonilla | Vaiheittainen opas

Johdanto

Suurten kielimallien käyttö tuotantoympäristössä voi olla haastavaa, koska ne saattavat kärsiä virheistä, epäjohdonmukaisuuksista ja viiveestä. Mallien kasvaessa ja kehittyessä on houkuttelevaa sisällyttää jokaiseen kehotteeseen enemmän kontekstia ja monimutkaisempia ohjeita toivoen, että malli pystyy käsittelemään ne, mutta yhden järjestelmäkehoteen käyttäminen suhteellisen suuren mallin kanssa voi aiheuttaa 3–15 sekunnin viiveen. Jos rakennat aikakriittistä agenttia, kuten puhelinkeskustelubottia, tämä on liian pitkä aika agentin pysähtymiselle.

Matalamman viiveen vaihtoehto on jakaa tehtävä pienempiin osiin ja antaa LLM-työnkulkujesi käsitellä kukin osa rinnakkain, jolloin pienemmät mallit pystyvät hoitamaan sen oikein. Tämä mahdollistaa viiveen merkittävän parantamisen ja monissa tapauksissa myös kustannusten alentamisen.

On olemassa muutamia eri tapoja suorittaa agenttipohjaisia työnkulkuja rinnakkain, mutta monissa agenttipohjaisissa työnkuluissa pääasiallinen pullonkaula on kutsu LLM:ään. Voit vähentää tätä pullonkaulaa tekemällä useita samanaikaisia kutsuja LLM:n API:iin Pythonin asyncio- ja aiohttp-kirjastojen avulla.

Tässä oppaassa käytät Pythonia rakentaaksesi agenttipohjaisia työnkulkuja, jotka tekevät useita LLM-kutsuja samanaikaisesti. Tämä mahdollistaa nopeampien ja tarkempien LLM-sovellusten rakentamisen, jotka voidaan ottaa käyttöön Dropletissa tai palvelimettomassa instanssissa API-päätepisteenä.

Rakennat asiakaspalvelun työnkulun kiinteistövälitystoimiston puhelinjärjestelmälle, joka voi vastata kysymyksiin listauksista, varata aikoja ja yhdistää soittajat ihmisedustajiin. Se tekee tämän tekemällä useita asynkronisia kutsuja eri LLM-päätepisteisiin, mikä vähentää LLM:n viivettä ja antaa enemmän joustavuutta LLM:n valinnassa kutakin tehtävää varten.

Keskeiset kohdat

Pienten, nopeiden mallien asynkronisten kutsujen käyttäminen voi auttaa sinua luomaan tehokkaita, matalan viiveen agenttipohjaisia työnkulkuja aikakriittisiin sovelluksiin
Monimalliprosessit antavat sinun mukauttaa jokaisen kehotteen eri mallille, vähentäen kokonaislaskentatarpeita ja alentamalla itseisännöityjen LLM-mallien sekä kolmannen osapuolen API:n käytön kustannuksia

Edellytykset

Tämän oppaan suorittamiseen tarvitset:

Paikallinen kehitysympäristö Python 3:lle.
Pääsy LLM-rajapintaan. Tässä opetusohjelmassa käytät Mistral-Small-3.2-24B -mallia, joka on otettu käyttöön GPU-Dropletissa vLLM:n avulla. Voit ottaa GPU-Dropletin käyttöön vLLM:llä mallin tarjoamiseksi seuraamalla ohjeita: How To Run GPT-OSS and vLLM on an AMD GPU Droplet with ROCm.
Tuttuus Pythonin async/await-syntaksin kanssa (valinnainen mutta hyödyllinen).

Vaihe 1 — Ympäristön määrittäminen

Ensiksi, ota ympäristösi käyttöön ja asenna riippuvuudet. Tässä oppaassa tehdään kutsuja Mistral-Small-3.2-24B -malliin, joka on otettu käyttöön H200 GPU Dropletissä, mutta sinun täytyy joko tehdä kutsuja omaan otettuun malliin tai kolmannen osapuolen LLM-rajapintaan. Tämä opas tekee kutsuja vain yhteen LLM:ään, mutta voit halutessasi käyttää eri mallia eri palveluntarjoajilta jokaisessa kehotteessa.

Luo virtuaaliympäristö ja asenna tarvittavat riippuvuudet:

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

Teet mallikutsusi asynkronisesti, joten sinun täytyy luoda joitakin asynkronisia funktioita, jotka voivat hallita näitä kutsuja. Käytät Pythonissa asyncio- ja aiohttp-kirjastoja. Jos käytät toista LLM-rajapintaa, sinun täytyy asentaa heidän asiakasohjelmansa hallitaksesi kutsuja heidän päätepisteeseensä.

Luo uusi tiedosto nimeltä agentic_workflows.py ja lisää seuraava koodi tuomaan tarvittavat riippuvuudet ja määrittämään API-päätepisteen.

Korvaa your_server_ip GPU Droplet -asennuksesi IP-osoitteella tai määritä asiakas LLM-rajapintaasi varten:

import asyncio
import aiohttp

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

Vaihe 2 — Asynkronisen kutsulogiikan kirjoittaminen

Haluat, että asynkroniset kutsusi voivat ottaa listan sanakirjoja, joista jokaisessa on mallin tunnus ja kehotus, jonka haluat lähettää LLM:lle, ja lähettää ne kaikki odottamatta kunkin kutsun yksittäistä vastausta. Kehotus sisältää keskusteluhistorian ja järjestelmäkehotuksen kyseistä kutsua varten. Lisäät synkronisen call_models-funktion, joka voidaan tuoda ja kutsua pääskriptissäsi.

Lisää seuraavat asynkroniset funktiot tiedostoon 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()-metodi luo asynkronisen asiakassession, joka lähettää POST-pyyntöjä API-päätepisteeseen ja käsittelee HTTP-vastaukset asynkronisesti. Tämä mahdollistaa useiden LLM API -kutsujen tekemisen samanaikaisesti sen sijaan, että ne suoritettaisiin peräkkäin.

_call_models_async()-funktio kerää kaikki antamiemme useiden pyyntöjen tiedot ja suorittaa kaikki API-kutsut odottamatta kunkin vastauksen saapumista. Sitten, kun vastaukset palautuvat, se lisää vastaukset listaan samassa järjestyksessä, jossa ne lähetettiin. call_models()-funktio on synkroninen funktio, joka tekee asynkronisten kutsujen tekemisestä helpompaa.

Vaihe 3 — Kehotustesi kirjoittaminen

Nyt kun sinulla on perustoiminnallisuus kirjoitettuna asynkronisten kutsujen tekemiseen API-päätepisteeseen, sinun täytyy kirjoittaa komennot. Sinulla tulee olemaan kolme erillistä järjestelmäkomentoa. Yksi auttaa hinnoittelukysymyksissä, yksi aikataulutuksessa ja yksi muiden ilmoitusaiheisten kysymysten tunnistamisessa ja ohjaamisessa ihmisedustajalle. Voit lisätä lisää, mukaan lukien joitakin, joilla on enemmän autonomista päätöksentekovaltaa, mutta nämä kolme toimivat toistaiseksi.

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

Näissä järjestelmän kehotuksen asetuksissa sisällytät kehotuksen otsikon myöhempää viittausta varten, sen mallin tunnisteen, jolle lähetät kehotuksen, sekä itse järjestelmän kehotuksen sanamuodon, johon keskusteluhistoria liitetään.

Vaihe 4 — Käyttäjän syötteen käsittely mallien kautta

Nyt sinun täytyy luoda funktio, joka voi ottaa käyttäjän keskustelun, lisätä siihen jokaisen järjestelmän kehotteen ja tehdä asynkroniset kutsut oikealle mallille kullekin kehotteelle käyttäen yläpuolella kirjoittamaasi call_models()-funktiota. Tämä esimerkki olettaa, että keskusteluhistoria tallennetaan käyttöliittymässä samaan muotoon, jota API-mme hyväksyy, ja lähetetään yhdessä käyttäjän viimeisimmän syötteen kanssa.

Lisää seuraava funktio tiedostoon 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)

Tämä funktio luo sanakirjaobjekteista koostuvan listan jokaiselle järjestelmän kehotteelle, joka sisältää mallin tunnuksen sekä järjestelmän kehotteen ja keskusteluhistorian. Sitten se kutsuu asynkronisia funktioita kutsuakseen jokaista mallia jokaiselle järjestelmän kehotteelle käyttämällä call_models()-funktiota.

Kun mallit on kutsuttu, vastaukset vastaanotetaan ja tallennetaan muuttujaan prompt_responses samassa järjestyksessä kuin ne lähetettiin, joten sinun täytyy määrittää uudet muuttujat mallien todellisille vastauksille.

Jatka lisäämistä run_agentic_workflow-funktioon:

    # 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")]

Vaihe 5 — Logiikan määrittäminen

Lopuksi sinun täytyy lisätä tähän funktioon looginen kerros, joka voi käydä läpi vastauksiasi ja päättää, mitä tehdä seuraavaksi. Voit tarkistaa myyntihinnan tietokannastasi, ohjata puhelun ihmisedustajalle, varata ajan yhdelle kiinteistönvälittäjistäsi tai lähettää käyttäjälle jatkokysymyksen ja pyytää lisätietoja.

Jatka lisäämistä run_agentic_workflow-funktioon tiedostossa 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

Vaihe 6 — Työnkulun testaaminen

Nyt kun olet kirjoittanut koodin perusagenttiprosessille, joka käsittelee kolmen tyyppisiä kysymyksiä, testaat sitä keskustelulla kohteen hinnasta. Tässä keskustelussa käyttäjä ei anna kohteen tunnusta ensimmäisessä kysymyksessään, ja malli vastaa jatkokysymyksellä, jossa sitä pyydetään. Käyttäjä antaa tunnuksen, ja malli vastaa oikealla hinnalla.

Luo uusi tiedosto nimeltä test_workflow.py ja lisää siihen seuraava testikoodi:

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

Suorita testiskripti:

python3 test_workflow.py

Näet samankaltaisen tulosteen kuin tämä:

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

Testaa nyt täydellä keskustelulla, jossa käyttäjä antaa ilmoituksen tunnisteen:

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

Suorita skripti uudelleen:

python3 test_workflow.py

Näet hinnatiedot:

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

Johtopäätös

Tässä opetusohjelmassa loit rinnakkaisen agenttipohjaisen työnkulun kokonaan alusta Pythonilla. Loit asynkronisia funktioita samanaikaisten LLM-kutsujen tekemistä varten, konfiguroit useita järjestelmäkehotteita eri tehtäviä varten ja toteutit reitityslogiikkakerroksen käyttäjäpyyntöjen käsittelemiseksi.

Jotta tämä kehys olisi tuotantovalmis, sinun täytyy testata, kirjoittaa kehotteet uudelleen, vaihtaa malleja ja testata uudelleen eri keskustelujen kanssa, kunnes saavutat tarkkuuden ja nopeuden tason, joka täyttää vaatimuksesi. Tällaisissa työnkuluissa sinun täytyy testata jokaista kehotetta yksitellen ja jatkaa niiden parantamista ajan myötä. Esimerkissämme valitsimme Mistral-Small-3.2-24B -mallin sen pienen viiveen (mediaanivasteaika alle 0,5 sekuntia), kyvyn seurata näitä tiettyjä ohjeita ja sen kyvyn asentaa yhdelle GPU:lle vuoksi. Testasimme myös GPT-OSS-120b:llä, GPT-OSS-20b:llä ja Llama 3.1 8b:llä, ja Mistral-malli suoriutui parhaiten tässä erityisessä käyttötapauksessa. Oikean mallin valitseminen jokaiselle kehotteelle tai tehtävälle on iteratiivinen prosessi.

Jos olet kiinnostunut käyttämään agenteja, joilla on enemmän autonomiaa, järjestelmäkehotteille voidaan antaa enemmän tai vähemmän autonomiaa antamalla niille pääsy enemmän kontekstiin tai useampiin työkaluihin. Voit jopa tehdä jatkopyyntöjä LLM:lle, jotka on "ketjutettu" yhteen päättelyä tai muita tarkoituksia varten. Kuitenkin, jos kehotteessa vaaditaan useita API-kutsuja ketjussa, tämä lisää työnkulun viivettä, koska et enää odota vain yhtä ryhmää mallien asynkronisia kutsuja.

Otimme Mistral-mallin käyttöön vLLM:n kanssa, joten meidän ei tarvinnut muuttaa kehotteidemme formaattia eri mallien toimivuuden varmistamiseksi. Kuitenkin OpenAI:lla, Claudella ja avoimen lähdekoodin malleilla voi olla erilaisia vaatimuksia kehotteiden muotoilulle, jotka lähetetään niiden API:hin. Jos käytät eri mallitoimittajia jokaisessa kutsussa, sinun täytyy kirjoittaa Python-funktioita, jotka voivat muuntaa kehotteiden formaatin vastaamaan sen mallin vaatimuksia, jota kutsut. Esimerkiksi saatat tarvita funktion, joka voi ottaa vLLM-formaatissa olevan kehotteen ja muuntaa syntaksin Claude-formaatin mukaiseksi kehotteeksi.

Huomaat usein, että jotkut mallit vievät paljon kauemmin kuin toiset, mikä saa sinut odottamaan hitaimman vastauksen saadaksesi kaikki vastaukset takaisin. Pythonin asyncio-kirjaston avulla voit käsitellä mallin vastauksia sitä mukaa kuin ne saapuvat, käyttämällä asyncio.as_completed() sijaan asyncio.gather(). Tämän jälkeen voit tarkistaa ensimmäiset vastaukset nähdäksesi, ovatko ne haluamasi reitin mukaisia, ja edetä ilman, että odotat kaikkia muita vastauksia. Tämä vaatisi myös muutoksia logiikkakerrokseen.

Kun lisäät lisää järjestelmäkehotteita, tärkein pullonkaula, johon tulet törmäämään, on LLM-rajapinnan nopeusrajoitukset tai GPU:si käsittelyrajoitukset. Paras tapa kiertää tämä on isännöidä oma mallisi GPU-Dropletissä, mutta voit myös lieventää näitä ongelmia lisäämällä API-nopeusrajoituskiintiöitäsi, jakamalla kutsuja eri rajapintojen välillä sekä toteuttamalla uudelleenkutsulogiikan, aikakatkaisujen käsittelyn ja virheenkäsittelyn, kun nopeusrajoituksiin törmätään.

Kiitos, että opiskelet Linux-Console.net-yhteisön kanssa.