Wyszukiwanie w witrynie

Jak tworzyć równoległe agentowe przepływy pracy w Pythonie | Samouczek krok po kroku

Wprowadzenie

Duże modele językowe mogą być trudne w użyciu w środowisku produkcyjnym, ponieważ mogą cierpieć na niedokładności, niespójności i opóźnienia. W miarę jak modele stają się większe i bardziej zaawansowane, kuszące jest dodawanie większej ilości kontekstu i bardziej skomplikowanych instrukcji w każdym zapytaniu, mając nadzieję, że model sobie z tym poradzi, ale przy użyciu pojedynczego zapytania systemowego z relatywnie dużym modelem opóźnienie może wynosić od 3 do 15 sekund. Jeśli tworzysz agenta reagującego w czasie rzeczywistym, na przykład bota do rozmów telefonicznych, to zbyt długi czas na wstrzymanie działania agenta.

Opcją o niższych opóźnieniach jest podzielenie zadania na mniejsze zadania i pozwolenie, aby przepływy pracy Twojego LLM obsługiwały każdy element równolegle, co ułatwia prawidłowe przetwarzanie przez mniejsze modele. Pozwoli to znacznie poprawić opóźnienia, a w wielu przypadkach także koszty.

Istnieje kilka różnych sposobów równoległego uruchamiania przepływów pracy z agentami, ale w wielu takich przepływach głównym wąskim gardłem jest wywołanie LLM. Można zmniejszyć to wąskie gardło, wykonując wiele równoczesnych wywołań API LLM przy użyciu bibliotek Python asyncio i aiohttp.

W tym samouczku użyjesz Pythona do tworzenia agentowych przepływów pracy, które wykonują wiele wywołań LLM asynchronicznie. Pozwoli to na tworzenie szybszych, bardziej dokładnych aplikacji LLM, które można wdrożyć na droplecie lub w instancji bezserwerowej jako punkt końcowy API.

Stworzysz workflow obsługi klienta dla systemu telefonicznego agencji nieruchomości, który będzie odpowiadał na pytania dotyczące ofert, umawiał spotkania i łączył rozmówców z ludzkimi przedstawicielami. Będzie to realizowane poprzez wykonywanie wielu asynchronicznych wywołań do różnych punktów końcowych LLM, co zmniejszy opóźnienia LLM i pozwoli na większą elastyczność w wyborze LLM dla każdego zadania.

Najważniejsze wnioski

  • Korzystanie z wywołań asynchronicznych do małych, szybkich modeli może pomóc w tworzeniu efektywnych, niskolatencyjnych przepływów pracy agentowych dla aplikacji wrażliwych na czas
  • Wielomodelowe przepływy pracy pozwalają dostosować każde zapytanie do innego modelu, zmniejszając ogólne zapotrzebowanie na moc obliczeniową i obniżając koszty samodzielnego hostowania modeli LLM oraz korzystania z zewnętrznych interfejsów API.

Wymagania wstępne

Aby ukończyć ten samouczek, będziesz potrzebować:

  • Lokalne środowisko deweloperskie dla Pythona 3.
  • Dostęp do API LLM. W tym samouczku użyjesz modelu Mistral-Small-3.2-24B uruchomionego na GPU Droplecie za pomocą vLLM. Możesz skonfigurować GPU Droplet z vLLM, aby obsługiwał model, postępując zgodnie z instrukcją Jak uruchomić GPT-OSS i vLLM na GPU Droplecie AMD z ROCm.
  • Znajomość składni async/await w Pythonie (opcjonalna, ale przydatna).

Krok 1 — Konfigurowanie środowiska

Najpierw skonfiguruj swoje środowisko i zainstaluj zależności. W tym poradniku wywoływane będą zapytania do modelu Mistral-Small-3.2-24B, który został wdrożony na GPU H200 Droplet, ale będziesz musiał albo wywoływać zapytania do własnego wdrożonego modelu, albo do API LLM firm trzecich. Ten poradnik korzysta tylko z jednego modelu LLM, ale będziesz mógł używać różnych modeli od różnych dostawców dla każdego zapytania, jeśli chcesz.

Utwórz środowisko wirtualne i zainstaluj wymagane zależności:

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

Będziesz wykonywać wywołania swojego modelu asynchronicznie, więc musisz utworzyć kilka funkcji asynchronicznych, które będą zarządzać tymi wywołaniami. Będziesz używać bibliotek asyncio i aiohttp w Pythonie. Jeśli używasz innego API LLM, będziesz musiał zainstalować ich klienta, aby zarządzać wywołaniami do ich punktu końcowego.

Utwórz nowy plik o nazwie agentic_workflows.py i dodaj następujący kod, aby zaimportować niezbędne zależności i zdefiniować punkt końcowy API.

Zastąp your_server_ip adresem IP swojej instalacji GPU Droplet lub zdefiniuj klienta dla swojego API LLM:

import asyncio
import aiohttp

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

Krok 2 — Pisanie logiki wywołań asynchronicznych

Chcesz, aby twoje wywołania asynchroniczne mogły przyjmować listę słowników, z których każdy zawiera ID modelu i prompt, który chcesz wysłać do LLM, i wysyłać je wszystkie bez czekania na odpowiedź z każdego wywołania. Prompt będzie zawierał historię rozmowy oraz systemowy prompt dla tego konkretnego wywołania. Dodasz funkcję synchroniczną call_models, którą można zaimportować i wywołać w głównym skrypcie.

Dodaj następujące asynchroniczne funkcje do 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))

Metoda aiohttp.ClientSession() tworzy asynchroniczną sesję klienta, która wysyła żądania POST do punktu końcowego API i obsługuje odpowiedź HTTP w sposób asynchroniczny. Umożliwia to wykonywanie wielu wywołań API LLM jednocześnie, zamiast sekwencyjnie.

Funkcja _call_models_async() zbiera wszystkie szczegóły wielu żądań, które dostarczamy, i wykonuje wszystkie wywołania API bez oczekiwania na każdą odpowiedź. Następnie, gdy odpowiedzi wracają, dodaje je do listy w tej samej kolejności, w jakiej zostały wysłane. Funkcja call_models() jest funkcją synchroniczną, która ułatwia wywoływanie funkcji asynchronicznych.

Krok 3 — Pisanie swoich promptów

Teraz, gdy masz już napisaną podstawową funkcjonalność umożliwiającą wykonywanie asynchronicznych wywołań do punktu końcowego API, musisz napisać swoje prośby. Będziesz mieć trzy oddzielne prośby systemowe. Jedna będzie pomagać w pytaniach dotyczących cen, druga w planowaniu, a trzecia w identyfikowaniu innych pytań związanych z ofertami i kierowaniu do przedstawiciela człowieka. Możesz dodać więcej, w tym niektóre z większą autonomią w podejmowaniu decyzji, ale te trzy wystarczą na początek.

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

W tych konfiguracjach poleceń systemowych należy uwzględnić tytuł polecenia do późniejszego odniesienia, identyfikator modelu, do którego wysyłasz to polecenie, oraz treść samego polecenia systemowego, do którego zostanie dołączona historia rozmowy.

Krok 4 — Przetwarzanie danych wejściowych użytkownika przez modele

Teraz musisz stworzyć funkcję, która będzie mogła przyjąć rozmowę użytkownika, dodać do niej każde z poleceń systemowych i wykonywać asynchroniczne wywołania do odpowiedniego modelu dla każdego polecenia, korzystając z funkcji call_models(), którą napisałeś powyżej. Ten przykład zakłada, że historia rozmowy jest zapisywana w interfejsie front-end w tym samym formacie, który akceptuje nasze API, i jest wysyłana wraz z najnowszym wejściem użytkownika.

Dodaj następującą funkcję do 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)

Ta funkcja tworzy listę obiektów słowników dla każdego komunikatu systemowego, która zawiera identyfikator modelu oraz komunikat systemowy i historię rozmowy. Następnie wywołuje funkcje asynchroniczne, aby uruchomić każdy model dla każdego komunikatu systemowego przy użyciu funkcji call_models().

Po wywołaniu modeli, odpowiedzi są odbierane i zapisywane w zmiennej prompt_responses w tej samej kolejności, w jakiej zostały wysłane, więc musisz przypisać nowe zmienne z faktycznymi odpowiedziami od modeli.

Kontynuuj dodawanie do funkcji run_agentic_workflow:

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

Krok 5 — Definiowanie logiki

Na koniec musisz dodać do tej funkcji warstwę logiki, która będzie mogła przeglądać Twoje odpowiedzi i określać, co zrobić dalej. Możesz sprawdzić cenę oferty w swojej bazie danych, przekierować rozmowę do przedstawiciela, umówić spotkanie z jednym z agentów nieruchomości lub wysłać użytkownikowi pytanie uzupełniające, prosząc o więcej szczegółów.

Kontynuuj dodawanie do funkcji run_agentic_workflow w pliku 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

Krok 6 — Testowanie Twojego przepływu pracy

Teraz, gdy masz napisany kod dla podstawowego agentowego przepływu pracy obsługującego trzy kategorie pytań, przetestujesz go w rozmowie na temat ceny ogłoszenia. W tej rozmowie użytkownik nie podaje ID ogłoszenia w swoim początkowym pytaniu, a model odpowiada pytaniem uzupełniającym, prosząc o nie. Użytkownik je podaje, a model odpowiada poprawną ceną.

Utwórz nowy plik o nazwie test_workflow.py i dodaj następujący kod testowy:

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

Uruchom skrypt testowy:

python3 test_workflow.py

Zobaczysz wynik podobny do tego:

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

Teraz przetestuj pełną rozmowę, w której użytkownik podaje identyfikator ogłoszenia:

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

Uruchom ponownie skrypt:

python3 test_workflow.py

Zobaczysz informacje o cenach:

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

Wniosek

W tym samouczku stworzyłeś równoległy, agentowy przepływ pracy od podstaw w Pythonie. Utworzyłeś funkcje asynchroniczne do wykonywania jednoczesnych wywołań LLM, skonfigurowałeś różne komunikaty systemowe dla różnych zadań oraz wdrożyłeś warstwę logiki routingu do obsługi żądań użytkowników.

Aby przygotować ten framework do produkcji, będziesz musiał testować, przepisywać komendy, zmieniać modele i ponownie testować przy różnych rozmowach, aż osiągniesz poziom dokładności i szybkości spełniający Twoje wymagania. W tego rodzaju przepływach pracy konieczne jest testowanie każdej komendy osobno i ciągłe jej doskonalenie w czasie. W naszym przykładzie wybraliśmy model Mistral-Small-3.2-24B ze względu na jego niskie opóźnienia (mediana czasu odpowiedzi poniżej 0,5 sekundy), zdolność do podążania za tymi konkretnymi instrukcjami oraz możliwość wdrożenia na pojedynczej karcie GPU. Przetestowaliśmy również modele GPT-OSS-120b, GPT-OSS-20b i Llama 3.1 8b, a model Mistral sprawdził się najlepiej w tym konkretnym przypadku użycia. Wybór odpowiedniego modelu dla każdej komendy lub zadania jest procesem iteracyjnym.

Jeśli jesteś zainteresowany korzystaniem z agentów o większej autonomii, promptom systemowym można nadać większą lub mniejszą autonomię, dając im dostęp do większego kontekstu lub większej liczby narzędzi. Można nawet organizować kolejne wywołania do LLM „połączone w łańcuch” w celu rozumowania lub innych celów. Jednakże, jeśli prompt wymaga wielu wywołań API w łańcuchu, zwiększy to opóźnienie w przepływie pracy, ponieważ nie będziesz już czekać tylko na jedną grupę asynchronicznych wywołań do modeli.

Wdrożyliśmy model Mistral z vLLM, więc nie musieliśmy zmieniać formatu naszych promptów, aby działały na różnych modelach. Jednak OpenAI, Claude i modele open-source mogą mieć różne wymagania co do formatowania promptów wysyłanych do ich API. Jeśli używasz różnych dostawców modeli przy każdym wywołaniu, będziesz musiał napisać funkcje w Pythonie, które mogą konwertować format promptów tak, aby odpowiadał wymaganiom wywoływanego modelu. Na przykład możesz potrzebować funkcji, która może wziąć prompt sformatowany dla vLLM i przekonwertować jego składnię na prompt sformatowany dla Claude.

Często zauważysz, że niektóre modele potrzebują znacznie więcej czasu niż inne, co powoduje, że musisz czekać na najwolniejszą odpowiedź, aby otrzymać wszystkie odpowiedzi. Z biblioteką asyncio w Pythonie możesz przetwarzać odpowiedzi modelu w miarę ich napływania, używając asyncio.as_completed() zamiast asyncio.gather(). Dzięki temu możesz sprawdzić pierwsze odpowiedzi, aby zobaczyć, czy idą w kierunku, który chcesz kontynuować, i ruszyć dalej bez czekania na resztę odpowiedzi. Wymagałoby to również zmian w warstwie logiki.

Podczas dodawania kolejnych poleceń systemowych głównym ograniczeniem, na które natkniesz się tutaj, będzie limitowanie liczby żądań w API LLM lub ograniczenia przetwarzania w Twoim GPU. Najlepszym sposobem na obejście tego problemu jest uruchomienie własnego modelu na GPU Droplecie, ale możesz również zmniejszyć te problemy, zwiększając limity żądań API, rozdzielając wywołania między różne API oraz implementując logikę ponawiania, obsługę limitów czasu i błędów w przypadku osiągnięcia limitów żądań.

Dziękujemy za naukę z społecznością Linux-Console.net.