Cara Membangun Alur Kerja Agen Paralel dengan Python | Tutorial Langkah demi Langkah

Pendahuluan

Model bahasa besar bisa menjadi tantangan untuk digunakan di lingkungan produksi karena mereka bisa mengalami ketidakakuratan, ketidakkonsistenan, dan latensi. Seiring model menjadi lebih besar dan lebih canggih, sering tergoda untuk memasukkan lebih banyak konteks dan instruksi yang lebih kompleks di setiap prompt, dengan harapan model dapat menangani itu, tetapi dengan satu prompt sistem menggunakan model yang relatif besar, latensi bisa berkisar antara 3-15 detik. Jika Anda sedang membangun agen yang sensitif terhadap waktu, seperti bot percakapan telepon, ini terlalu lama untuk seorang agen berhenti sejenak.

Opsi dengan latensi lebih rendah adalah membagi tugas menjadi tugas-tugas yang lebih kecil dan membiarkan alur kerja LLM Anda menangani setiap bagian secara paralel, sehingga lebih mudah bagi model yang lebih kecil untuk menangani dengan benar. Ini akan memungkinkan Anda meningkatkan latensi secara drastis, dan dalam banyak kasus, biaya Anda juga.

Ada beberapa cara berbeda untuk menjalankan alur kerja agen secara paralel, tetapi dalam banyak alur kerja agen, hambatan utama adalah pemanggilan ke LLM. Anda dapat mengurangi hambatan ini dengan melakukan beberapa pemanggilan sekaligus ke API LLM menggunakan pustaka asyncio dan aiohttp di Python.

Dalam tutorial ini, Anda akan menggunakan Python untuk membangun alur kerja agen yang melakukan beberapa panggilan LLM secara asynchronous. Ini memungkinkan Anda membangun aplikasi LLM yang lebih cepat dan lebih akurat yang dapat di-deploy ke Droplet atau instance serverless sebagai endpoint API.

Anda akan membangun alur kerja layanan pelanggan untuk sistem telepon agen real estate yang dapat menjawab pertanyaan tentang daftar properti, menjadwalkan janji temu, dan menghubungkan penelepon dengan perwakilan manusia. Ini akan dilakukan dengan membuat beberapa panggilan asinkron ke berbagai endpoint LLM, mengurangi latency LLM dan memungkinkan lebih banyak fleksibilitas dalam pemilihan LLM untuk setiap tugas.

Poin Penting

  • Menggunakan panggilan asinkron ke model kecil dan cepat dapat membantu Anda membuat alur kerja agen yang efektif dan latensi rendah untuk aplikasi yang memerlukan waktu sensitif
  • Alur kerja multi-model memungkinkan Anda menyesuaikan setiap prompt untuk model yang berbeda, mengurangi kebutuhan komputasi secara keseluruhan dan menurunkan biaya pengelolaan model LLM sendiri serta penggunaan API pihak ketiga

Prasyarat

Untuk menyelesaikan tutorial ini, Anda akan membutuhkan:

  • Lingkungan pengembangan lokal untuk Python 3.
  • Akses ke API LLM. Dalam tutorial ini, Anda akan menggunakan model Mistral-Small-3.2-24B yang dijalankan pada GPU Droplet menggunakan vLLM. Anda dapat menyiapkan GPU Droplet menggunakan vLLM untuk melayani sebuah model dengan mengikuti panduan Cara Menjalankan GPT-OSS dan vLLM pada GPU Droplet AMD dengan ROCm.
  • Keterampilan dengan sintaks async/await di Python (opsional tetapi bermanfaat).

Langkah 1 — Menyiapkan Lingkungan Anda

Pertama, siapkan lingkungan Anda dan instal dependensi Anda. Dalam tutorial ini, panggilan akan dilakukan ke model Mistral-Small-3.2-24B yang telah dipasang pada Droplet GPU H200, tetapi Anda harus membuat panggilan ke model Anda sendiri yang telah dipasang atau API LLM pihak ketiga. Tutorial ini hanya melakukan panggilan ke satu LLM, tetapi Anda akan dapat menggunakan model berbeda dari penyedia yang berbeda untuk setiap prompt jika Anda mau.

Buat lingkungan virtual dan instal dependensi yang diperlukan:

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

Anda akan membuat panggilan model secara asynchronous, jadi Anda perlu membuat beberapa fungsi asynchronous yang dapat mengelola panggilan ini. Anda akan menggunakan pustaka asyncio dan aiohttp di Python. Jika Anda menggunakan API LLM yang berbeda, Anda perlu menginstal klien mereka untuk mengelola panggilan ke endpoint mereka.

Buat file baru bernama agentic_workflows.py dan tambahkan kode berikut untuk mengimpor dependensi yang diperlukan serta mendefinisikan endpoint API.

Gantilah your_server_ip dengan alamat IP dari deployment 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 Logika Panggilan Asinkron

Anda ingin panggilan asinkron Anda dapat mengambil daftar kamus yang masing-masing memiliki ID model dan prompt yang ingin Anda kirim ke LLM, dan mengirim semuanya tanpa menunggu respons masing-masing panggilan. Prompt akan mencakup riwayat percakapan dan prompt sistem untuk panggilan tersebut. Anda akan menambahkan fungsi call_models sinkron yang dapat diimpor dan dipanggil di 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))

Metode aiohttp.ClientSession() membuat sesi klien asinkron yang mengirim permintaan POST ke endpoint API dan menangani respons HTTP secara asinkron. Ini memungkinkan beberapa panggilan API LLM dilakukan secara bersamaan daripada secara berurutan.

Fungsi _call_models_async() mengumpulkan semua detail dari beberapa permintaan yang kami berikan dan melakukan semua panggilan API tanpa menunggu masing-masing respons. Kemudian, saat respons datang, fungsi ini menambahkan respons ke dalam daftar dengan urutan yang sama seperti saat dikirim. Fungsi call_models() adalah fungsi sinkron yang memudahkan pemanggilan panggilan asinkron.

Langkah 3 — Menulis Prompt Anda

Sekarang setelah Anda memiliki fungsi dasar yang ditulis untuk melakukan panggilan asinkron ke endpoint API, Anda perlu menulis prompt Anda. Anda akan memiliki tiga prompt sistem yang terpisah. Satu akan membantu dengan pertanyaan harga, satu dengan penjadwalan, dan satu untuk mengidentifikasi pertanyaan lain tentang listing dan mengarahkan ke perwakilan manusia. Anda bisa menambahkan lebih banyak lagi, termasuk beberapa dengan kekuatan pengambilan keputusan yang lebih mandiri, tetapi ketiga ini akan berfungsi untuk saat 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 prompt sistem ini, Anda menyertakan judul prompt untuk referensi nanti, ID model yang akan Anda kirimkan prompt tersebut, dan kata-kata dari prompt sistem itu sendiri yang akan dilampirkan ke riwayat percakapan.

Langkah 4 — Memproses Masukan Pengguna melalui Model

Sekarang Anda perlu membuat fungsi yang dapat mengambil percakapan pengguna, menambahkan setiap prompt sistem ke dalamnya, dan melakukan panggilan asinkron ke model yang tepat untuk setiap prompt menggunakan fungsi call_models() yang telah Anda tulis di atas. Contoh ini mengasumsikan bahwa riwayat percakapan disimpan di front-end dalam format yang sama dengan yang diterima API kami dan dikirim bersama dengan input terbaru dari 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 membuat daftar objek kamus untuk setiap prompt sistem yang mencakup ID model dan prompt sistem + riwayat percakapan. Kemudian fungsi ini memanggil fungsi asinkron untuk memanggil setiap model untuk setiap prompt sistem menggunakan fungsi call_models().

Setelah memanggil model, respons diterima dan disimpan dalam variabel prompt_responses dengan urutan yang sama seperti saat dikirim, jadi Anda perlu menetapkan variabel baru dengan respons sebenarnya dari model tersebut.

Lanjutkan menambahkan ke 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 Logika

Akhirnya, Anda perlu menambahkan lapisan logika ke fungsi ini yang dapat menelusuri respons Anda dan menentukan langkah selanjutnya. Anda bisa mencari harga listing di database Anda, mengarahkan panggilan ke perwakilan manusia, menjadwalkan pertemuan dengan salah satu agen real estat Anda, atau mengirim pertanyaan lanjutan kepada pengguna untuk meminta detail lebih lanjut.

Lanjutkan menambahkan ke fungsi run_agentic_workflow di 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 Alur Kerja Anda

Sekarang setelah Anda menulis kode untuk alur kerja agen dasar yang menangani tiga kategori pertanyaan, Anda akan mengujinya dengan percakapan tentang harga suatu listing. Dalam percakapan ini, pengguna tidak memberikan ID listing dalam pertanyaan awal mereka, dan model merespons dengan pertanyaan lanjutan memintanya. Pengguna memberikannya, dan model merespons dengan harga yang benar.

Buat file baru bernama test_workflow.py dan tambahkan kode pengujian 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 pengujian:

python3 test_workflow.py

Anda akan melihat output yang mirip dengan ini:

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

Sekarang uji dengan percakapan lengkap di mana pengguna memberikan ID daftar:

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 lagi:

python3 test_workflow.py

Anda akan melihat informasi harga:

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

Kesimpulan

Dalam tutorial ini, Anda membangun alur kerja agen paralel dari awal menggunakan Python. Anda membuat fungsi asinkron untuk melakukan panggilan LLM secara bersamaan, mengonfigurasi beberapa prompt sistem untuk tugas yang berbeda, dan mengimplementasikan lapisan logika routing untuk menangani permintaan pengguna.

Untuk membuat kerangka kerja ini siap digunakan dalam produksi, Anda perlu menguji, menulis ulang prompt, mengganti model, dan menguji kembali dengan berbagai percakapan sampai Anda mencapai tingkat akurasi dan kecepatan yang sesuai dengan kebutuhan Anda. Dengan jenis alur kerja ini, Anda perlu menguji setiap prompt secara individual dan terus meningkatkannya seiring waktu. Dalam contoh kami, kami memilih model Mistral-Small-3.2-24B karena latensinya yang rendah (waktu respons median kurang dari 0,5 detik), kemampuannya mengikuti instruksi spesifik ini, dan kemampuannya untuk dijalankan pada satu GPU. Kami juga menguji dengan GPT-OSS-120b, GPT-OSS-20b, dan Llama 3.1 8b, dan model Mistral menunjukkan performa terbaik untuk kasus penggunaan tertentu ini. Memilih model yang tepat untuk setiap prompt atau tugas adalah proses iteratif.

Jika Anda tertarik menggunakan agen dengan otonomi lebih, prompt sistem dapat diberikan lebih banyak atau lebih sedikit otonomi dengan memberi mereka akses ke lebih banyak konteks atau lebih banyak alat. Anda bahkan dapat melakukan panggilan lanjutan ke LLM yang "dirangkai" bersama untuk tujuan penalaran atau tujuan lainnya. Namun, jika sebuah prompt memerlukan beberapa panggilan API dalam sebuah rantai, ini akan meningkatkan latensi alur kerja karena Anda tidak lagi hanya menunggu satu kelompok panggilan asinkron ke model.

Kami menerapkan model Mistral dengan vLLM, jadi kami tidak perlu mengubah format prompt kami agar dapat digunakan untuk model yang berbeda. Namun, OpenAI, Claude, dan model open-source mungkin memiliki persyaratan berbeda untuk format prompt yang dikirim ke API mereka. Jika Anda menggunakan penyedia model yang berbeda untuk setiap panggilan, Anda akan perlu menulis fungsi Python yang dapat mengubah format prompt agar sesuai dengan persyaratan model yang Anda panggil. Misalnya, Anda mungkin memerlukan fungsi yang dapat mengambil prompt berformat vLLM dan mengubah sintaksnya menjadi prompt berformat Claude.

Anda sering kali akan menemukan beberapa model membutuhkan waktu lebih lama daripada yang lain, sehingga Anda harus menunggu respons yang paling lambat untuk mendapatkan semua respons kembali. Dengan pustaka asyncio di Python, Anda dapat memproses respons model saat mereka datang, menggunakan asyncio.as_completed() alih-alih asyncio.gather(). Kemudian Anda dapat memeriksa respons pertama untuk melihat apakah itu jalur yang ingin Anda lanjutkan dan melanjutkan tanpa menunggu sisa respons. Ini juga akan memerlukan perubahan pada lapisan logika.

Saat menambahkan lebih banyak prompt sistem, hambatan utama yang akan Anda temui di sini adalah pembatasan kecepatan pada API LLM atau batas pemrosesan pada GPU Anda. Cara terbaik untuk mengatasi ini adalah dengan menjalankan model Anda sendiri di GPU Droplet, tetapi Anda juga dapat mengurangi masalah ini dengan meningkatkan kuota batas kecepatan API Anda, menyebarkan panggilan Anda ke beberapa API, serta menerapkan logika pengulangan, penanganan waktu tunggu, dan penanganan kesalahan ketika batas kecepatan tercapai.

Terima kasih telah belajar bersama Komunitas Linux-Console.net.