Zoeken op website

Hoe Parallelle Agentische Workflows Bouwen met Python | Stapsgewijze Tutorial

Inleiding

Grote taalmodellen kunnen moeilijk te gebruiken zijn in een productieve omgeving omdat ze kunnen lijden aan onnauwkeurigheden, inconsistenties en vertraging. Naarmate modellen groter en capabeler worden, is het verleidelijk om meer context en complexere instructies in elke prompt op te nemen, in de hoop dat het model dit aankan, maar bij één enkele systeem-prompt met een relatief groot model kan de vertraging overal tussen de 3 en 15 seconden liggen. Als je een tijdgevoelige agent bouwt, zoals een telefoonconversatie-bot, is dit te lang voor een agent om te pauzeren.

Een optie met lagere latency is om de taak op te splitsen in kleinere taken en je LLM-workflows elk onderdeel parallel te laten afhandelen, waardoor het voor kleinere modellen makkelijker wordt om het correct te verwerken. Dit stelt je in staat om je latency aanzienlijk te verbeteren en in veel gevallen ook je kosten te verlagen.

Er zijn een paar verschillende manieren om agent-gestuurde workflows parallel uit te voeren, maar in veel agent-gestuurde workflows is de belangrijkste bottleneck de oproep naar de LLM. Je kunt deze bottleneck verminderen door meerdere gelijktijdige oproepen naar LLM-API's te doen met behulp van de Python-bibliotheken asyncio en aiohttp.

In deze handleiding gebruik je Python om agentische workflows te bouwen die meerdere LLM-aanroepen asynchroon uitvoeren. Hiermee kun je snellere en nauwkeurigere LLM-toepassingen bouwen die kunnen worden ingezet op een Droplet of serverloze instantie als een API-endpoint.

Je zult een klantenservice workflow opbouwen voor het telefoonsysteem van een makelaarskantoor die vragen over vastgoedaanbiedingen kan beantwoorden, afspraken kan plannen en bellers kan doorverbinden naar menselijke medewerkers. Dit zal gebeuren door meerdere asynchrone oproepen naar verschillende LLM-eindpunten te maken, waardoor de LLM-latentie wordt verminderd en er meer flexibiliteit is in de keuze van LLM voor elke taak.

Belangrijkste punten

  • Het gebruik van asynchrone oproepen naar kleine, snelle modellen kan je helpen om effectieve, lage-latentie agentische workflows te creëren voor tijdgevoelige toepassingen
  • Multi-model workflows stellen je in staat om elke prompt aan te passen voor een ander model, waardoor je de totale rekencapaciteit vermindert en de kosten voor het zelf hosten van LLM-modellen en het gebruik van API's van derden verlaagt.

Vereisten

Om deze tutorial te voltooien, heb je het volgende nodig:

  • Een lokale ontwikkelomgeving voor Python 3.
  • Toegang tot een LLM-API. In deze tutorial gebruik je het Mistral-Small-3.2-24B-model dat is gedeployed op een GPU Droplet met vLLM. Je kunt een GPU Droplet instellen met vLLM om een model te hosten door de instructies te volgen in How To Run GPT-OSS and vLLM on an AMD GPU Droplet with ROCm.
  • Bekendheid met de Python async/await-syntaxis (optioneel maar handig).

Stap 1 — Je omgeving instellen

Stel eerst je omgeving in en installeer je afhankelijkheden. In deze tutorial zullen er oproepen worden gedaan naar een Mistral-Small-3.2-24B-model dat is uitgerold op een H200 GPU Droplet, maar je zult ofwel oproepen moeten doen naar je eigen uitgerolde model of naar een LLM-API van een derde partij. Deze tutorial doet alleen oproepen naar één LLM, maar je kunt voor elke prompt een ander model van verschillende aanbieders gebruiken als je dat wilt.

Maak een virtuele omgeving aan en installeer de vereiste afhankelijkheden:

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

Je zult je modelaanroepen asynchroon uitvoeren, dus je moet enkele asynchrone functies maken die deze aanroepen kunnen beheren. Je zult de asyncio- en aiohttp-bibliotheken in Python gebruiken. Als je een andere LLM-API gebruikt, moet je hun client installeren om de aanroepen naar hun endpoint te beheren.

Maak een nieuw bestand genaamd agentic_workflows.py en voeg de volgende code toe om de benodigde afhankelijkheden te importeren en het API-eindpunt te definiëren.

Vervang your_server_ip door het IP-adres van je GPU Droplet-implementatie of definieer de client voor je LLM API:

import asyncio
import aiohttp

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

Stap 2 — Het schrijven van de asynchrone aanroeplogica

Je wilt dat je asynchrone oproepen een lijst van woordenboeken kunnen aannemen, die elk de model-ID en prompt bevatten die je naar het LLM wilt sturen, en ze allemaal verzenden zonder te wachten op de individuele reactie van elke oproep. De prompt zal de gespreksgeschiedenis en systeem-prompt voor die specifieke oproep bevatten. Je zult een synchrone call_models-functie toevoegen die kan worden geïmporteerd en aangeroepen in je hoofdscripts.

Voeg de volgende asynchrone functies toe aan 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))

De aiohttp.ClientSession()-methode creëert een asynchrone clientsessie die POST-verzoeken naar de API-endpoint stuurt en de HTTP-respons asynchroon verwerkt. Hierdoor kunnen meerdere LLM API-aanroepen gelijktijdig worden uitgevoerd in plaats van achter elkaar.

De _call_models_async() functie verzamelt alle details van meerdere verzoeken die we aanleveren en voert alle API-aanroepen uit zonder te wachten op elke reactie. Vervolgens, zodra de reacties terugkomen, voegt het de reacties toe aan een lijst in dezelfde volgorde als ze zijn verzonden. De call_models() functie is een synchrone functie die het gemakkelijker maakt om de asynchrone oproepen uit te voeren.

Stap 3 — Het schrijven van je prompts

Nu je de basisfunctionaliteit hebt geschreven om asynchrone oproepen naar een API-eindpunt te doen, moet je je prompts schrijven. Je zult drie afzonderlijke systeem-prompts hebben. Eén zal helpen met prijsvragen, één met planning en één met het identificeren van andere vragen over vermeldingen en verwijzen naar een menselijke vertegenwoordiger. Je kunt er meer toevoegen, inclusief sommige met meer autonome besluitvormingsbevoegdheid, maar deze drie zijn voorlopig voldoende.

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

In deze systeempromptconfiguraties voeg je de titel van de prompt toe voor latere verwijzing, het model-ID waar je die prompt naartoe stuurt, en de formulering van de systeemprompt zelf waaraan de gespreksgeschiedenis zal worden gekoppeld.

Stap 4 — Verwerken van gebruikersinvoer via de modellen

Nu moet je een functie maken die het gesprek van de gebruiker kan nemen, elk van de systeemopdrachten eraan kan toevoegen en je asynchrone oproepen naar het juiste model kan doen voor elke opdracht met behulp van de functie call_models() die je hierboven hebt geschreven. Dit voorbeeld gaat ervan uit dat de gespreksgeschiedenis aan de voorkant wordt opgeslagen in hetzelfde formaat dat onze API accepteert en wordt meegestuurd met de meest recente invoer van de gebruiker.

Voeg de volgende functie toe aan 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)

Deze functie maakt een lijst van woordenboekobjecten voor elke systeemprompt die de model-ID en de systeemprompt + gespreksgeschiedenis bevatten. Vervolgens roept het de asynchrone functies aan om elk model voor elke systeemprompt aan te roepen met behulp van de call_models()-functie.

Na het aanroepen van de modellen worden de reacties ontvangen en opgeslagen in de prompt_responses-variabele in dezelfde volgorde als waarin ze zijn verzonden, dus je moet nieuwe variabelen toewijzen met de daadwerkelijke reacties van de modellen.

Blijf toevoegen aan de run_agentic_workflow functie:

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

Stap 5 — Het definiëren van de logica

Ten slotte moet je een logisch laag aan deze functie toevoegen die door je reacties kan kijken en kan bepalen wat de volgende stap is. Je kunt een verkoopprijs in je database opzoeken, het gesprek doorverwijzen naar een menselijke medewerker, een afspraak plannen met een van je makelaars, of een vervolgvraag naar de gebruiker sturen om meer details te vragen.

Blijf toevoegen aan de run_agentic_workflow-functie in 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

Stap 6 — Je workflow testen

Nu je de code hebt geschreven voor een basis agentische workflow om drie categorieën vragen te behandelen, ga je deze testen met een gesprek over de prijs van een aanbieding. In dit gesprek geeft de gebruiker in zijn eerste vraag de aanbieding-ID niet op, en het model reageert met een vervolgvraag waarin dit wordt gevraagd. De gebruiker geeft deze, en het model reageert met de juiste prijs.

Maak een nieuw bestand genaamd test_workflow.py en voeg de volgende testcode toe:

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

Voer het testscripts uit:

python3 test_workflow.py

Je zult uitvoer zien die hierop lijkt:

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

Test nu met een volledig gesprek waarin de gebruiker de vermelding-ID opgeeft:

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

Voer het script opnieuw uit:

python3 test_workflow.py

Je ziet de prijsinformatie:

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

Conclusie

In deze tutorial heb je een parallelle agentische workflow helemaal vanuit Python opgebouwd. Je hebt asynchrone functies gemaakt om gelijktijdige LLM-aanroepen te doen, meerdere systeem prompts geconfigureerd voor verschillende taken, en een routeringslogica laag geïmplementeerd om gebruikersverzoeken af te handelen.

Om dit framework productieklaar te maken, moet je testen, prompts herschrijven, modellen wijzigen en opnieuw testen met verschillende gesprekken totdat je een niveau van nauwkeurigheid en snelheid hebt bereikt dat aan je eisen voldoet. Bij dit soort workflows moet je elke prompt afzonderlijk testen en ze in de loop van de tijd blijven verbeteren. In ons voorbeeld kozen we voor het Mistral-Small-3.2-24B-model vanwege de lage latency (mediane reactietijd van minder dan 0,5 seconden), het vermogen om deze specifieke instructies te volgen en de mogelijkheid om op één enkele GPU te worden ingezet. We hebben ook getest met GPT-OSS-120b, GPT-OSS-20b en Llama 3.1 8b, en het Mistral-model presteerde het beste voor dit specifieke gebruik. Het kiezen van het juiste model voor elke prompt of taak is een iteratief proces.

Als je geïnteresseerd bent in het gebruik van agenten met meer autonomie, kan de systeemprompt meer of minder autonomie krijgen door ze toegang te geven tot meer context of meer hulpmiddelen. Je kunt zelfs vervolgoproepen naar de LLM hebben die 'aaneengeschakeld' zijn voor redenering of andere doeleinden. Echter, als een prompt meerdere API-oproepen in een keten vereist, zal dit de vertraging van de workflow vergroten omdat je niet langer alleen op één groep asynchrone oproepen naar modellen hoeft te wachten.

We hebben het Mistral-model geïmplementeerd met vLLM, dus we hoefden het formaat van onze prompts niet te wijzigen om met verschillende modellen te werken. OpenAI, Claude en open-source modellen kunnen echter verschillende vereisten hebben voor de opmaak van prompts die naar hun API’s worden verzonden. Als je voor elke oproep verschillende modelproviders gebruikt, zul je Python-functies moeten schrijven die het formaat van prompts kunnen omzetten zodat het voldoet aan de vereisten van het model dat je oproept. Bijvoorbeeld, je hebt mogelijk een functie nodig die een vLLM-geformatteerde prompt kan nemen en de syntaxis kan converteren naar een Claude-geformatteerde prompt.

Je zult vaak merken dat sommige modellen veel langer doen over een reactie dan andere, waardoor je moet wachten op de langzaamste reactie om alle antwoorden terug te krijgen. Met de asyncio-bibliotheek van Python kun je modelantwoorden verwerken zodra ze binnenkomen, door asyncio.as_completed() te gebruiken in plaats van asyncio.gather(). Vervolgens kun je de eerste antwoorden bekijken om te zien of dit de route is die je wilt volgen en doorgaan zonder te wachten op de resterende antwoorden. Dit zou ook aanpassingen aan de logischelaag vereisen.

Bij het toevoegen van meer systeemopdrachten is het belangrijkste knelpunt waar je tegenaan zult lopen de snelheidsbeperking van de LLM-API of de verwerkingslimieten van je GPU. De beste manier om dit te omzeilen is door je eigen model te hosten op een GPU-Droplet, maar je kunt deze problemen ook beperken door je API-snelheidslimietquota te verhogen, je oproepen over meerdere API's te verdelen en retry-logica, time-outbeheer en foutafhandeling te implementeren voor wanneer snelheidslimieten worden bereikt.

Bedankt dat je samen met de Linux-Console.net-community hebt geleerd.