Webbplatssökning

Hur man bygger parallella agentbaserade arbetsflöden med Python | Steg-för-steg-guide

Introduktion

Stora språkmodeller kan vara utmanande att använda i en produktionsmiljö eftersom de kan ha problem med felaktigheter, inkonsekvenser och fördröjning. När modeller blir större och mer kapabla är det frestande att inkludera mer kontext och mer komplexa instruktioner i varje prompt, i hopp om att modellen ska kunna hantera det, men med en enda systemprompt som använder en relativt stor modell kan fördröjningen vara allt från 3 till 15 sekunder. Om du bygger en tidskänslig agent, som en telefonkonversationsbot, är detta för lång tid för att agenten ska kunna pausa.

Ett alternativ med lägre latens är att dela upp uppgiften i mindre delar och låta dina LLM-arbetsflöden hantera varje del parallellt, vilket gör det lättare för mindre modeller att hantera korrekt. Detta kommer göra det möjligt för dig att drastiskt förbättra din latens och i många fall även dina kostnader.

Det finns några olika sätt att köra agentbaserade arbetsflöden parallellt, men i många agentbaserade arbetsflöden är den största flaskhalsen anropet till LLM. Du kan minska denna flaskhals genom att göra flera samtidigt anrop till LLM-API:er med hjälp av Pythons asyncio- och aiohttp-bibliotek.

I den här handledningen kommer du att använda Python för att bygga agentbaserade arbetsflöden som gör flera LLM-anrop asynkront. Detta gör det möjligt för dig att skapa snabbare och mer exakta LLM-applikationer som kan distribueras på en Droplet eller en serverlös instans som en API-slutpunkt.

Du kommer att bygga ett arbetsflöde för kundservice för en fastighetsbyrås telefonsystem som kan svara på frågor om objekt, boka tider och koppla samtalare till mänskliga representanter. Detta kommer att ske genom flera asynkrona anrop till olika LLM-endpoints, vilket minskar LLM-latensen och ger större flexibilitet i valet av LLM för varje uppgift.

Viktiga punkter

  • Att använda asynkrona anrop till små, snabba modeller kan hjälpa dig att skapa effektiva arbetsflöden med låg latens för tidskänsliga applikationer
  • Flermodellarbetsflöden låter dig anpassa varje prompt för en annan modell, vilket minskar dina totala beräkningsbehov och sänker kostnaden för att själv driva LLM-modeller och använda tredjeparts-API:er

Förutsättningar

För att slutföra denna handledning behöver du:

  • En lokal utvecklingsmiljö för Python 3.
  • Tillgång till en LLM API. I den här handledningen kommer du att använda Mistral-Small-3.2-24B-modellen som är distribuerad på en GPU Droplet med vLLM. Du kan sätta upp en GPU Droplet med vLLM för att tillhandahålla en modell genom att följa Hur man kör GPT-OSS och vLLM på en AMD GPU Droplet med ROCm.
  • Kännedom om Pythons async/await-syntax (valfritt men hjälpsamt).

Steg 1 — Ställa in din miljö

Först, ställ in din miljö och installera dina beroenden. I den här handledningen kommer samtal att göras till en Mistral-Small-3.2-24B-modell som har distribuerats på en H200 GPU Droplet, men du kommer antingen behöva göra samtal till din egen distribuerade modell eller till en tredjeparts LLM API. Den här handledningen gör endast samtal till en LLM, men du kommer att kunna använda en annan modell från olika leverantörer för varje prompt om du vill.

Skapa en virtuell miljö och installera de nödvändiga beroenden:

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

Du kommer att göra dina modellanrop asynkront, så du behöver skapa några asynkrona funktioner som kan hantera dessa anrop. Du kommer att använda biblioteken asyncio och aiohttp i Python. Om du använder ett annat LLM-API måste du installera deras klient för att hantera anrop till deras endpoint.

Skapa en ny fil som heter agentic_workflows.py och lägg till följande kod för att importera de nödvändiga beroendena och definiera API-slutpunkten.

Byt ut your_server_ip mot IP-adressen för din GPU Droplet-distribution eller definiera klienten för din LLM API:

import asyncio
import aiohttp

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

Steg 2 — Skriva logiken för asynkrona anrop

Du vill att dina asynkrona anrop ska kunna ta en lista med ordböcker som var och en innehåller modell-ID och prompt som du vill skicka till LLM, och skicka dem alla utan att vänta på varje anrops individuella svar. Prompten kommer att inkludera konversationshistoriken och systemprompten för det specifika anropet. Du kommer att lägga till en synkron call_models-funktion som kan importeras och anropas i ditt huvudscript.

Lägg till följande asynkrona funktioner i 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))

Metoden aiohttp.ClientSession() skapar en asynkron klient-session som skickar POST-förfrågningar till API-slutpunkten och hanterar HTTP-svaret asynkront. Detta gör att flera LLM API-anrop kan göras samtidigt istället för sekventiellt.

Funktionen _call_models_async() samlar alla detaljer för flera förfrågningar som vi tillhandahåller och gör alla API-anrop utan att vänta på varje svar. Sedan, när svaren kommer tillbaka, lägger den till svaren i en lista i samma ordning som de skickades. Funktionen call_models() är en synkron funktion som gör de asynkrona anropen enklare att använda.

Steg 3 — Skriv dina uppmaningar

Nu när du har skrivit den grundläggande funktionaliteten för att göra asynkrona anrop till en API-endpoint, behöver du skriva dina prompts. Du kommer att ha tre separata systemprompts. En kommer att hjälpa till med prisfrågor, en med schemaläggning och en med att identifiera andra frågor om listningar och vidarebefordra till en mänsklig representant. Du kan lägga till fler, inklusive några med mer autonomt beslutsfattande, men dessa tre fungerar för tillfället.

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

I dessa systempromptkonfigurationer inkluderar du titeln på prompten för senare referens, modell-ID:t som du skickar prompten till, och formuleringen av själva systemprompten som konversationshistoriken kommer att kopplas till.

Steg 4 — Bearbeta användarinmatning genom modellerna

Nu behöver du skapa en funktion som kan ta användarens konversation, lägga till varje systemprompt till den, och göra dina asynkrona anrop till rätt modell för varje prompt med hjälp av call_models()-funktionen som du skrev ovan. Detta exempel förutsätter att konversationshistoriken sparas i front-end på samma sätt som vårt API accepterar och skickas tillsammans med användarens senaste input.

Lägg till följande funktion i 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)

Den här funktionen skapar en lista med ordboksobjekt för varje systemprompt som inkluderar modell-ID och systemprompten + konversationshistorik. Den anropar sedan de asynkrona funktionerna för att anropa varje modell för varje systemprompt med hjälp av call_models()-funktionen.

Efter att ha kallat på modellerna tas svaren emot och sparas i variabeln prompt_responses i samma ordning som de skickades, så du behöver tilldela nya variabler med de faktiska svaren från modellerna.

Fortsätt lägga till i run_agentic_workflow-funktionen:

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

Steg 5 — Definiera logiken

Slutligen behöver du lägga till ett logiklager i denna funktion som kan granska dina svar och avgöra vad som ska göras härnäst. Du kan slå upp ett listpris i din databas, vidarebefordra samtalet till en mänsklig representant, boka en tid med en av dina fastighetsmäklare, eller skicka en uppföljningsfråga till användaren för att få mer information.

Fortsätt att lägga till i run_agentic_workflow-funktionen i 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

Steg 6 — Testa ditt arbetsflöde

Nu när du har skrivit koden för ett grundläggande agentbaserat arbetsflöde för att hantera tre kategorier av frågor, kommer du att testa det med en konversation om priset på en annons. I denna konversation ger användaren inte annons-ID i sin initiala fråga, och modellen svarar med en följdfråga där den ber om det. Användaren lämnar det, och modellen svarar med det korrekta priset.

Skapa en ny fil som heter test_workflow.py och lägg till följande testkod:

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

Kör testscripts:

python3 test_workflow.py

Du kommer att se en output som liknar detta:

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

Testa nu med en fullständig konversation där användaren anger annons-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}")

Kör skriptet igen:

python3 test_workflow.py

Du kommer att se prisinformationen:

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

Slutsats

I den här handledningen byggde du ett parallellt agentbaserat arbetsflöde från grunden i Python. Du skapade asynkrona funktioner för att göra samtidiga LLM-anrop, konfigurerade flera systemprompter för olika uppgifter och implementerade ett routningslogiklager för att hantera användarförfrågningar.

För att göra detta ramverk produktionsklart behöver du testa, skriva om prompts, byta modeller och testa igen med olika konversationer tills du uppnår en nivå av noggrannhet och hastighet som uppfyller dina krav. Med den här typen av arbetsflöden behöver du testa varje prompt individuellt och fortsätta förbättra dem över tid. I vårt exempel valde vi modellen Mistral-Small-3.2-24B på grund av dess låga fördröjning (medianresponstid på mindre än 0,5 sekunder), dess förmåga att följa dessa specifika instruktioner och dess möjlighet att köras på en enda GPU. Vi testade också med GPT-OSS-120b, GPT-OSS-20b och Llama 3.1 8b, och Mistral-modellen presterade bäst för detta specifika användningsfall. Att välja rätt modell för varje prompt eller uppgift är en iterativ process.

Om du är intresserad av att använda agenter med mer självständighet kan systemuppmaningar ges mer eller mindre autonomi genom att ge dem tillgång till mer sammanhang eller fler verktyg. Du kan till och med ha uppföljningssamtal med LLM:n som är "kedjade" tillsammans för resonemang eller andra ändamål. Om en uppmaning däremot kräver flera API-anrop i en kedja kommer detta att öka latensen i arbetsflödet eftersom du inte längre bara väntar på en grupp asynkrona anrop till modeller.

Vi distribuerade Mistral-modellen med vLLM, så vi behövde inte ändra formatet på våra prompts för att fungera med olika modeller. Däremot kan OpenAI, Claude och öppen källkod-modeller ha olika krav på hur prompts som skickas till deras API:er ska formateras. Om du använder olika leverantörer av modeller för varje anrop, behöver du skriva Python-funktioner som kan konvertera formatet på prompts för att matcha kraven hos den modell du anropar. Till exempel kan du behöva en funktion som kan ta en prompt formaterad för vLLM och konvertera syntaxen till en prompt formaterad för Claude.

Du kommer ofta att märka att vissa modeller tar mycket längre tid än andra, vilket gör att du måste vänta på det långsammaste svaret för att få tillbaka alla svar. Med Pythons asyncio-bibliotek kan du bearbeta modelsvaren när de kommer in, genom att använda asyncio.as_completed() istället för asyncio.gather(). Då kan du kontrollera de första svaren för att se om de är den väg du vill gå vidare med och fortsätta utan att vänta på resten av svaren. Detta skulle också kräva ändringar i logiklagret.

När du lägger till fler systemprompter är den största flaskhalsen du kommer att stöta på här antingen begränsningar i LLM API:et eller processgränser med ditt GPU. Det bästa sättet att komma runt detta är att köra din egen modell på en GPU-droplet, men du kan också minska dessa problem genom att öka dina kvoter för API-ratebegränsning, sprida dina anrop över olika API:er samt implementera logik för försök på nytt, hantering av timeout och felhantering när rate limits uppnås.

Tack för att du lär dig tillsammans med Linux-Console.net-communityn.