Website-Suche

Wie man parallele agentische Workflows mit Python erstellt | Schritt-für-Schritt-Anleitung

Einführung

Große Sprachmodelle können in einer Produktionsumgebung schwer zu verwenden sein, da sie unter Ungenauigkeiten, Inkonsistenzen und Verzögerungen leiden können. Wenn Modelle größer und leistungsfähiger werden, ist es verlockend, mehr Kontext und komplexere Anweisungen in jede Eingabeaufforderung aufzunehmen, in der Hoffnung, dass das Modell damit umgehen kann. Bei einer einzelnen Systemaufforderung mit einem relativ großen Modell kann die Verzögerung jedoch zwischen 3 und 15 Sekunden liegen. Wenn Sie einen zeitkritischen Agenten entwickeln, wie zum Beispiel einen Telefon-Konversations-Bot, ist dies zu lang, als dass ein Agent pausieren könnte.

Eine Option mit niedrigerer Latenz besteht darin, die Aufgabe in kleinere Aufgaben zu unterteilen und Ihre LLM-Workflows jede einzelne Aufgabe parallel bearbeiten zu lassen, wodurch es für kleinere Modelle einfacher wird, diese korrekt zu bearbeiten. Dies ermöglicht es Ihnen, Ihre Latenz erheblich zu verbessern und in vielen Fällen auch Ihre Kosten.

Es gibt einige verschiedene Möglichkeiten, agentische Workflows parallel auszuführen, aber in vielen agentischen Workflows ist der Hauptengpass der Aufruf des LLM. Sie können diesen Engpass verringern, indem Sie mehrere gleichzeitige Aufrufe an LLM-APIs unter Verwendung der Python-Bibliotheken asyncio und aiohttp durchführen.

In diesem Tutorial werden Sie Python verwenden, um agentische Workflows zu erstellen, die mehrere LLM-Aufrufe asynchron durchführen. Dadurch können Sie schnellere, genauere LLM-Anwendungen erstellen, die auf einem Droplet oder einer serverlosen Instanz als API-Endpunkt bereitgestellt werden können.

Sie werden einen Kundenservice-Workflow für das Telefonsystem einer Immobilienagentur erstellen, der Fragen zu Immobilienangeboten beantworten, Termine vereinbaren und Anrufer mit menschlichen Mitarbeitern verbinden kann. Dies geschieht durch mehrere asynchrone Aufrufe verschiedener LLM-Endpunkte, wodurch die LLM-Latenz reduziert und mehr Flexibilität bei der Auswahl von LLMs für jede Aufgabe ermöglicht wird.

Wichtige Erkenntnisse

  • Die Verwendung asynchroner Aufrufe kleiner, schneller Modelle kann Ihnen helfen, effektive, latenzarme agentische Arbeitsabläufe für zeitkritische Anwendungen zu erstellen.
  • Multi-Modell-Workflows ermöglichen es Ihnen, jedes Prompt für ein anderes Modell anzupassen, wodurch Ihr gesamter Rechenbedarf reduziert und die Kosten für das Selbst-Hosting von LLM-Modellen sowie die Nutzung von Drittanbieter-APIs gesenkt werden.

Voraussetzungen

Um dieses Tutorial abzuschließen, benötigen Sie:

  • Eine lokale Entwicklungsumgebung für Python 3.
  • Zugriff auf eine LLM-API. In diesem Tutorial werden Sie das Modell Mistral-Small-3.2-24B verwenden, das auf einem GPU-Droplet unter Nutzung von vLLM bereitgestellt wird. Sie können ein GPU-Droplet mit vLLM einrichten, um ein Modell bereitzustellen, indem Sie How To Run GPT-OSS and vLLM on an AMD GPU Droplet with ROCm befolgen.
  • Vertrautheit mit der Python async/await-Syntax (optional, aber hilfreich).

Schritt 1 — Einrichten Ihrer Umgebung

Richte zuerst deine Umgebung ein und installiere deine Abhängigkeiten. In diesem Tutorial werden Aufrufe an ein Mistral-Small-3.2-24B-Modell vorgenommen, das auf einem H200 GPU Droplet bereitgestellt wurde, aber Sie müssen entweder Ihr eigenes bereitgestelltes Modell oder eine Drittanbieter-LLM-API aufrufen. Dieses Tutorial ruft nur ein LLM auf, aber du kannst für jeden Prompt ein anderes Modell von verschiedenen Anbietern verwenden, wenn du möchtest.

Erstellen Sie eine virtuelle Umgebung und installieren Sie die erforderlichen Abhängigkeiten:

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

Sie werden Ihre Modellaufrufe asynchron durchführen, daher müssen Sie einige asynchrone Funktionen erstellen, die diese Aufrufe verwalten können. Sie werden die Bibliotheken asyncio und aiohttp in Python verwenden. Wenn Sie eine andere LLM-API verwenden, müssen Sie deren Client installieren, um Aufrufe an deren Endpunkt zu verwalten.

Erstellen Sie eine neue Datei mit dem Namen agentic_workflows.py und fügen Sie den folgenden Code hinzu, um die erforderlichen Abhängigkeiten zu importieren und den API-Endpunkt zu definieren.

Ersetzen Sie your_server_ip durch die IP-Adresse Ihres GPU-Droplet-Deployments oder definieren Sie den Client für Ihre LLM-API:

import asyncio
import aiohttp

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

Schritt 2 — Schreiben der Logik für den asynchronen Aufruf

Sie möchten, dass Ihre asynchronen Aufrufe in der Lage sind, eine Liste von Wörterbüchern zu übernehmen, die jeweils die Modell-ID und den Prompt enthalten, den Sie an das LLM senden möchten, und sie alle senden, ohne auf die individuelle Antwort jedes Aufrufs zu warten. Der Prompt wird die Gesprächshistorie und den System-Prompt für diesen bestimmten Aufruf enthalten. Sie werden eine synchrone call_models-Funktion hinzufügen, die importiert und in Ihrem Hauptskript aufgerufen werden kann.

Fügen Sie die folgenden asynchronen Funktionen zu agentic_workflows.py hinzu:

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

Die Methode aiohttp.ClientSession() erstellt eine asynchrone Client-Session, die POST-Anfragen an den API-Endpunkt sendet und die HTTP-Antwort asynchron verarbeitet. Dies ermöglicht es, mehrere LLM-API-Aufrufe gleichzeitig statt nacheinander zu tätigen.

Die _call_models_async()-Funktion sammelt alle Details mehrerer Anfragen, die wir bereitstellen, und führt alle API-Aufrufe aus, ohne auf jede Antwort zu warten. Dann, wenn die Antworten zurückkommen, fügt sie die Antworten einer Liste in derselben Reihenfolge hinzu, in der sie gesendet wurden. Die call_models()-Funktion ist eine synchrone Funktion, die die asynchronen Aufrufe leichter aufrufbar macht.

Schritt 3 — Schreiben Ihrer Eingabeaufforderungen

Da Sie nun die grundlegende Funktionalität geschrieben haben, um asynchrone Aufrufe an einen API-Endpunkt zu machen, müssen Sie Ihre Prompts schreiben. Sie werden drei separate System-Prompts haben. Einer wird bei Preisfragen helfen, einer bei Terminplanung und einer bei der Identifizierung anderer Listing-Fragen und der Weiterleitung an einen menschlichen Vertreter. Sie können weitere hinzufügen, einschließlich einiger mit mehr autonomer Entscheidungsbefugnis, aber diese drei werden vorerst ausreichen.

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 diesen System-Prompt-Konfigurationen fügen Sie den Titel des Prompts zur späteren Bezugnahme, die Modell-ID, an die Sie diesen Prompt senden, und die Formulierung des System-Prompts selbst hinzu, an die der Gesprächsverlauf angehängt wird.

Schritt 4 — Verarbeitung der Benutzereingaben durch die Modelle

Jetzt müssen Sie eine Funktion erstellen, die die Unterhaltung des Benutzers übernehmen, jede der Systemaufforderungen hinzufügen und Ihre asynchronen Aufrufe an das richtige Modell für jede Aufforderung mit der Funktion call_models() ausführen kann, die Sie oben geschrieben haben. Dieses Beispiel geht davon aus, dass der Gesprächsverlauf im Frontend im gleichen Format gespeichert wird, das unsere API akzeptiert, und zusammen mit der jüngsten Eingabe des Benutzers gesendet wird.

Fügen Sie die folgende Funktion zu agentic_workflows.py hinzu:

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)

Diese Funktion erstellt eine Liste von Wörterbuchobjekten für jede Systemaufforderung, die die Modell-ID und die Systemaufforderung + den Gesprächsverlauf enthält. Anschließend ruft sie die asynchronen Funktionen auf, um jedes Modell für jede Systemaufforderung mit der Funktion call_models() aufzurufen.

Nachdem die Modelle aufgerufen wurden, werden die Antworten empfangen und in der Variable prompt_responses in derselben Reihenfolge gespeichert, in der sie gesendet wurden, sodass Sie neue Variablen mit den tatsächlichen Antworten der Modelle zuweisen müssen.

Fahren Sie fort, die Funktion run_agentic_workflow zu erweitern:

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

Schritt 5 — Die Logik definieren

Schließlich müssen Sie dieser Funktion eine Logikschicht hinzufügen, die Ihre Antworten durchsuchen und bestimmen kann, was als Nächstes zu tun ist. Sie können einen Listenpreis in Ihrer Datenbank nachschlagen, den Anruf an einen menschlichen Vertreter weiterleiten, einen Termin mit einem Ihrer Immobilienmakler vereinbaren oder dem Benutzer eine Folgefrage senden, in der nach weiteren Details gefragt wird.

Fahren Sie fort, die Funktion run_agentic_workflow in agentic_workflows.py zu erweitern:

    # 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

Schritt 6 — Testen Ihres Workflows

Jetzt, da Sie den Code für einen grundlegenden agentischen Arbeitsablauf zur Handhabung von drei Kategorien von Fragen geschrieben haben, werden Sie ihn mit einem Gespräch über den Preis eines Eintrags testen. In diesem Gespräch gibt der Benutzer die Eintrags-ID in seiner Anfangsfrage nicht an, und das Modell antwortet mit einer Folgefrage, die diese anfordert. Der Benutzer gibt sie an, und das Modell antwortet mit dem richtigen Preis.

Erstellen Sie eine neue Datei mit dem Namen test_workflow.py und fügen Sie den folgenden Testcode hinzu:

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

Führen Sie das Testskript aus:

python3 test_workflow.py

Sie werden eine Ausgabe ähnlich wie diese sehen:

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

Testen Sie nun mit einem vollständigen Gespräch, bei dem der Benutzer die Angebots-ID angibt:

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

Führen Sie das Skript erneut aus:

python3 test_workflow.py

Sie werden die Preisangaben sehen:

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

Fazit

In diesem Tutorial haben Sie einen parallelen agentischen Workflow von Grund auf in Python erstellt. Sie haben asynchrone Funktionen erstellt, um gleichzeitige LLM-Aufrufe zu tätigen, mehrere System-Prompts für verschiedene Aufgaben konfiguriert und eine Routing-Logikschicht implementiert, um Benutzeranfragen zu bearbeiten.

Um dieses Framework produktionsreif zu machen, müssen Sie testen, Prompts neu schreiben, Modelle ändern und erneut mit verschiedenen Unterhaltungen testen, bis Sie ein Niveau an Genauigkeit und Geschwindigkeit erreicht haben, das Ihren Anforderungen entspricht. Bei solchen Workflows müssen Sie jeden Prompt einzeln testen und im Laufe der Zeit kontinuierlich verbessern. In unserem Beispiel haben wir das Modell Mistral-Small-3.2-24B aufgrund seiner geringen Latenz (mittlere Antwortzeit von weniger als 0,5 Sekunden), seiner Fähigkeit, diesen spezifischen Anweisungen zu folgen, und seiner Möglichkeit, auf einer einzelnen GPU eingesetzt zu werden, gewählt. Wir haben auch mit GPT-OSS-120b, GPT-OSS-20b und Llama 3.1 8b getestet, und das Mistral-Modell schnitt bei diesem speziellen Anwendungsfall am besten ab. Die Wahl des richtigen Modells für jeden Prompt oder jede Aufgabe ist ein iterativer Prozess.

Wenn Sie daran interessiert sind, Agenten mit mehr Autonomie zu verwenden, kann den System-Prompts mehr oder weniger Autonomie gegeben werden, indem ihnen mehr Kontext oder mehr Werkzeuge zur Verfügung gestellt werden. Sie können sogar Folgeaufrufe an das LLM „verkettet“ durchführen, um Schlussfolgerungen oder andere Zwecke zu erreichen. Wenn jedoch ein Prompt mehrere API-Aufrufe in einer Kette erfordert, wird dies die Latenz des Workflows erhöhen, da Sie nicht mehr nur auf eine Gruppe asynchroner Aufrufe an Modelle warten.

Wir haben das Mistral-Modell mit vLLM bereitgestellt, sodass wir das Format unserer Prompts nicht ändern mussten, um mit verschiedenen Modellen zu arbeiten. Allerdings können OpenAI, Claude und Open-Source-Modelle unterschiedliche Anforderungen an die Formatierung der Prompts haben, die an ihre APIs gesendet werden. Wenn Sie für jeden Aufruf unterschiedliche Modellanbieter verwenden, müssen Sie Python-Funktionen schreiben, die das Format der Prompts so konvertieren können, dass es den Anforderungen des Modells entspricht, das Sie aufrufen. Zum Beispiel benötigen Sie möglicherweise eine Funktion, die ein im vLLM-Format vorliegendes Prompt nehmen und die Syntax in ein im Claude-Format vorliegendes Prompt umwandeln kann.

Sie werden oft feststellen, dass einige Modelle viel länger benötigen als andere, sodass Sie auf die langsamste Antwort warten müssen, um alle Antworten zurückzubekommen. Mit der Python-Bibliothek asyncio können Sie Modellantworten verarbeiten, sobald sie eintreffen, indem Sie asyncio.as_completed() anstelle von asyncio.gather() verwenden. Dann können Sie die ersten Antworten überprüfen, um zu sehen, ob sie der Weg sind, den Sie weiterverfolgen möchten, und voranschreiten, ohne auf die restlichen Antworten zu warten. Dies würde auch Änderungen an der Logikebene erfordern.

Beim Hinzufügen weiterer System-Prompts ist das Hauptproblem, auf das Sie stoßen werden, die Ratenbegrenzung bei der LLM-API oder Verarbeitungsgrenzen bei Ihrer GPU. Der beste Weg, dies zu umgehen, besteht darin, Ihr eigenes Modell auf einem GPU-Droplet zu hosten, aber Sie können diese Probleme auch abmildern, indem Sie Ihre API-Ratenlimitquoten erhöhen, Ihre Aufrufe auf verschiedene APIs verteilen und Logik für Wiederholungsversuche, Timeout-Behandlung und Fehlerbehandlung implementieren, wenn Ratenbegrenzungen erreicht werden.

Danke, dass Sie mit der Linux-Console.net Community lernen.