Cara Membina Aliran Kerja Agenetik Selari dengan Python | Tutorial Langkah demi Langkah

Pengenalan

Model bahasa besar boleh menjadi mencabar untuk digunakan dalam persekitaran pengeluaran kerana mereka boleh mengalami ketidakakuratan, ketidakkonsistenan, dan kelewatan. Apabila model menjadi lebih besar dan lebih berkeupayaan, agak menggoda untuk memasukkan lebih banyak konteks dan arahan yang lebih kompleks dalam setiap prompt, dengan harapan model dapat menguruskannya, tetapi dengan satu prompt sistem menggunakan model yang agak besar, kelewatan boleh berkisar antara 3-15 saat. Jika anda membina ejen yang sensitif terhadap masa, seperti bot perbualan telefon, ini terlalu lama untuk ejen berhenti sebentar.

Pilihan dengan latensi lebih rendah adalah dengan memecahkan tugas kepada tugas-tugas yang lebih kecil dan membiarkan aliran kerja LLM anda mengendalikan setiap bahagian secara selari, menjadikannya lebih mudah bagi model yang lebih kecil untuk menanganinya dengan betul. Ini akan membolehkan anda meningkatkan latensi dengan ketara, dan dalam banyak kes, mengurangkan kos anda.

Terdapat beberapa cara berbeza untuk menjalankan aliran kerja agen secara selari, tetapi dalam banyak aliran kerja agen, halangan utama adalah panggilan ke LLM. Anda boleh mengurangkan halangan ini dengan membuat beberapa panggilan serentak ke API LLM menggunakan perpustakaan asyncio dan aiohttp Python.

Dalam tutorial ini, anda akan menggunakan Python untuk membina aliran kerja agenik yang membuat panggilan LLM berbilang secara tidak serentak. Ini akan membolehkan anda membina aplikasi LLM yang lebih pantas dan lebih tepat yang boleh digunakan pada Droplet atau instance tanpa pelayan sebagai titik akhir API.

Anda akan membina aliran kerja perkhidmatan pelanggan untuk sistem telefon agensi hartanah yang boleh menjawab soalan mengenai senarai hartanah, menjadualkan temu janji, dan menghubungkan pemanggil kepada wakil manusia. Ia akan melakukan ini dengan membuat beberapa panggilan tak serentak ke pelbagai titik akhir LLM, mengurangkan kelewatan LLM dan membenarkan lebih fleksibiliti dalam pemilihan LLM untuk setiap tugas.

Kesimpulan Utama

  • Menggunakan panggilan tak segerak kepada model kecil dan pantas boleh membantu anda mencipta aliran kerja agen yang berkesan dan berlatensi rendah untuk aplikasi yang sensitif kepada masa
  • Aliran kerja pelbagai model membolehkan anda menyesuaikan setiap arahan untuk model yang berbeza, mengurangkan keperluan pengiraan keseluruhan anda dan menurunkan kos penggunaan model LLM yang dihos sendiri serta penggunaan API pihak ketiga

Prasyarat

Untuk melengkapkan tutorial ini, anda akan memerlukan:

  • Persekitaran pembangunan tempatan untuk Python 3.
  • Akses kepada API LLM. Dalam tutorial ini, anda akan menggunakan model Mistral-Small-3.2-24B yang dipasang pada GPU Droplet menggunakan vLLM. Anda boleh menyediakan GPU Droplet menggunakan vLLM untuk menyajikan model dengan mengikuti Cara Menjalankan GPT-OSS dan vLLM pada GPU Droplet AMD dengan ROCm.
  • Biasa dengan sintaks async/await Python (pilihan tetapi berguna).

Langkah 1 — Menyediakan Persekitaran Anda

Mula-mula, sediakan persekitaran anda dan pasang kebergantungan anda. Dalam tutorial ini, panggilan akan dibuat kepada model Mistral-Small-3.2-24B yang telah dikerahkan pada GPU Droplet H200, tetapi anda perlu membuat panggilan kepada model yang anda kerahkan sendiri atau API LLM pihak ketiga. Tutorial ini hanya membuat panggilan kepada satu LLM, tetapi anda boleh menggunakan model berbeza daripada penyedia berbeza untuk setiap arahan jika anda mahu.

Cipta persekitaran maya dan pasang kebergantungan yang diperlukan:

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

Anda akan membuat panggilan model anda secara tidak segerak, jadi anda perlu membuat beberapa fungsi tidak segerak yang dapat mengurus panggilan ini. Anda akan menggunakan pustaka asyncio dan aiohttp dalam Python. Jika anda menggunakan API LLM yang berbeza, anda perlu memasang klien mereka untuk mengurus panggilan ke titik akhir mereka.

Buat fail baru bernama agentic_workflows.py dan tambah kod berikut untuk mengimport kebergantungan yang diperlukan dan menentukan titik akhir API.

Gantikan your_server_ip dengan alamat IP bagi penempatan GPU Droplet anda atau tentukan klien untuk API LLM anda:

import asyncio
import aiohttp

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

Langkah 2 — Menulis Logik Panggilan Asinkron

Anda ingin panggilan asynchronous anda dapat menerima senarai kamus yang masing-masing mempunyai ID model dan prompt yang ingin anda hantar ke LLM, dan menghantarnya semua tanpa menunggu maklum balas individu bagi setiap panggilan. Prompt itu akan termasuk sejarah perbualan dan prompt sistem untuk panggilan tersebut. Anda akan menambahkan fungsi call_models secara synchronous yang boleh diimport dan dipanggil dalam skrip utama anda.

Tambahkan fungsi asinkron berikut ke 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))

Kaedah aiohttp.ClientSession() mencipta sesi klien asinkron yang menghantar permintaan POST ke titik akhir API dan menguruskan respons HTTP secara asinkron. Ini membolehkan beberapa panggilan API LLM dilakukan serentak dan bukan secara bersiri.

Fungsi _call_models_async() mengumpulkan semua butiran bagi pelbagai permintaan yang kami berikan dan membuat semua panggilan API tanpa menunggu setiap respons. Kemudian, apabila respons diterima, ia menambah respons tersebut ke dalam senarai mengikut urutan yang sama seperti yang dihantar. Fungsi call_models() adalah fungsi segerak yang memudahkan pemanggilan panggilan asynchronous.

Langkah 3 — Menulis Arahan Anda

Sekarang bahawa anda telah menulis fungsi asas untuk membuat panggilan tak segerak ke titik akhir API, anda perlu menulis arahan anda. Anda akan mempunyai tiga arahan sistem yang berasingan. Satu akan membantu dengan soalan harga, satu dengan penjadualan, dan satu dengan mengenal pasti soalan senarai lain dan mengarahkan kepada wakil manusia. Anda boleh menambah lebih banyak, termasuk beberapa dengan kuasa membuat keputusan yang lebih autonomi, tetapi tiga ini sudah memadai buat masa ini.

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

Dalam konfigurasi arahan sistem ini, anda termasuk tajuk arahan untuk rujukan kemudian, ID model yang anda hantar arahan itu, dan perkataan arahan sistem itu sendiri yang sejarah perbualan akan disertakan.

Langkah 4 — Memproses Input Pengguna Melalui Model

Sekarang anda perlu mencipta fungsi yang boleh mengambil perbualan pengguna, menambah setiap arahan sistem kepadanya, dan membuat panggilan asynchronous anda ke model yang betul untuk setiap arahan menggunakan fungsi call_models() yang anda tulis di atas. Contoh ini mengandaikan bahawa sejarah perbualan disimpan di bahagian hadapan dalam format yang sama seperti yang diterima oleh API kami dan dihantar bersama-sama dengan input terkini pengguna.

Tambahkan fungsi berikut ke 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)

Fungsi ini mencipta senarai objek kamus untuk setiap prompt sistem yang termasuk ID model dan prompt sistem + sejarah perbualan. Ia kemudian memanggil fungsi tak segerak untuk memanggil setiap model bagi setiap prompt sistem menggunakan fungsi call_models().

Selepas memanggil model-model tersebut, respons diterima dan disimpan dalam pembolehubah prompt_responses mengikut turutan yang sama seperti yang dihantar, jadi anda perlu menetapkan pembolehubah baru dengan respons sebenar daripada model-model tersebut.

Teruskan menambah kepada fungsi 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")]

Langkah 5 — Menentukan Logik

Akhirnya, anda perlu menambah lapisan logik pada fungsi ini yang dapat menyemak respons anda dan menentukan apa yang perlu dilakukan seterusnya. Anda boleh menyemak harga senarai dalam pangkalan data anda, mengalihkan panggilan kepada wakil manusia, menjadualkan temu janji dengan salah seorang ejen hartanah anda, atau menghantar soalan susulan kepada pengguna untuk meminta maklumat lebih lanjut.

Teruskan menambah pada fungsi run_agentic_workflow dalam 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

Langkah 6 — Menguji Aliran Kerja Anda

Sekarang setelah anda menulis kod untuk aliran kerja ejen asas yang menangani tiga kategori soalan, anda akan mengujinya dengan perbualan mengenai harga sesebuah senarai. Dalam perbualan ini, pengguna tidak memberikan ID senarai dalam soalan awal mereka, dan model membalas dengan soalan susulan untuk memintanya. Pengguna memberikannya, dan model membalas dengan harga yang betul.

Buat fail baru bernama test_workflow.py dan tambah kod ujian berikut:

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

Jalankan skrip ujian:

python3 test_workflow.py

Anda akan melihat output yang serupa dengan ini:

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

Sekarang uji dengan perbualan lengkap di mana pengguna memberikan ID penyenaraian:

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

Jalankan skrip itu sekali lagi:

python3 test_workflow.py

Anda akan melihat maklumat harga:

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

Kesimpulan

Dalam tutorial ini, anda membina aliran kerja ejen selari dari awal menggunakan Python. Anda mencipta fungsi tak segerak untuk membuat panggilan LLM serentak, mengkonfigurasi pelbagai arahan sistem untuk tugasan yang berbeza, dan melaksanakan lapisan logik penghalaan untuk mengendalikan permintaan pengguna.

Untuk menjadikan rangka kerja ini sedia untuk produksi, anda perlu menguji, menulis semula arahan (prompts), menukar model, dan menguji semula dengan perbualan yang berbeza sehingga anda mencapai tahap ketepatan dan kelajuan yang memenuhi keperluan anda. Dengan jenis aliran kerja seperti ini, anda perlu menguji setiap arahan secara individu dan terus memperbaikinya dari masa ke masa. Dalam contoh kami, kami memilih model Mistral-Small-3.2-24B kerana kelewatan yang rendah (masa tindak balas median kurang daripada 0.5 saat), kebolehannya untuk mengikuti arahan khusus ini, dan kebolehannya untuk dikerahkan pada satu GPU. Kami juga menguji dengan GPT-OSS-120b, GPT-OSS-20b, dan Llama 3.1 8b, dan model Mistral menunjukkan prestasi terbaik untuk kes penggunaan tertentu ini. Memilih model yang betul untuk setiap arahan atau tugasan adalah proses berulang.

Jika anda berminat menggunakan ejen dengan lebih autonomi, arahan sistem boleh diberikan lebih atau kurang autonomi dengan memberi mereka akses kepada lebih banyak konteks atau lebih banyak alat. Anda juga boleh melakukan panggilan susulan kepada LLM yang "disambungkan" bersama untuk tujuan pemikiran atau tujuan lain. Walau bagaimanapun, jika satu arahan memerlukan pelbagai panggilan API secara berturutan, ini akan meningkatkan kelewatan aliran kerja kerana anda tidak lagi hanya menunggu satu kumpulan panggilan tak segerak kepada model.

Kami menggunakan model Mistral dengan vLLM, jadi kami tidak perlu menukar format arahan kami untuk berfungsi dengan model yang berbeza. Walau bagaimanapun, OpenAI, Claude, dan model sumber terbuka mungkin mempunyai keperluan yang berbeza untuk format arahan yang dihantar ke API mereka. Jika anda menggunakan penyedia model yang berbeza untuk setiap panggilan, anda perlu menulis fungsi Python yang boleh menukar format arahan supaya sesuai dengan keperluan model yang anda panggil. Sebagai contoh, anda mungkin memerlukan fungsi yang boleh mengambil arahan yang diformatkan untuk vLLM dan menukar sintaksnya menjadi arahan yang diformatkan untuk Claude.

Anda sering akan mendapati beberapa model mengambil masa lebih lama daripada yang lain, menyebabkan anda perlu menunggu respons paling lambat untuk mendapatkan semua respons kembali. Dengan perpustakaan asyncio Python, anda boleh memproses respons model sebaik sahaja ia sampai, menggunakan asyncio.as_completed() dan bukannya asyncio.gather(). Kemudian anda boleh memeriksa respons pertama untuk melihat sama ada ia adalah laluan yang ingin anda teruskan dan bergerak maju tanpa menunggu baki respons. Ini juga memerlukan perubahan pada lapisan logik.

Apabila menambah lebih banyak arahan sistem, halangan utama yang akan anda hadapi di sini ialah had kadar dengan API LLM atau had pemprosesan dengan GPU anda. Cara terbaik untuk mengatasinya adalah dengan mengehoskan model anda sendiri pada GPU Droplet, tetapi anda juga boleh mengurangkan masalah ini dengan meningkatkan kuota had kadar API anda, menyebarkan panggilan anda merentasi API, dan melaksanakan logik cuba semula, pengendalian tamat masa, dan pengendalian ralat apabila had kadar tercapai.

Terima kasih kerana belajar bersama Komuniti Linux-Console.net.