Pesquisa de sites

Como Construir Fluxos de Trabalho Agenciais Paralelos com Python | Tutorial Passo a Passo

Introdução

Modelos de linguagem grandes podem ser desafiantes de usar num ambiente de produção porque podem sofrer de imprecisões, inconsistências e latência. À medida que os modelos se tornam maiores e mais capazes, é tentador incluir mais contexto e instruções mais complexas em cada prompt, na esperança de que o modelo consiga lidar com isso, mas com um único prompt de sistema usando um modelo relativamente grande, a latência pode variar entre 3 a 15 segundos. Se estiver a construir um agente sensível ao tempo, como um bot de conversação telefónica, este tempo é demasiado longo para o agente fazer uma pausa.

Uma opção com menor latência é dividir a tarefa em tarefas mais pequenas e fazer com que os seus fluxos de trabalho LLM tratem cada parte em paralelo, tornando mais fácil para modelos mais pequenos lidarem com elas corretamente. Isto permitirá que melhore drasticamente a sua latência e, em muitos casos, os seus custos.

Existem várias formas de executar fluxos de trabalho agentes em paralelo, mas em muitos fluxos de trabalho agentes, o principal gargalo é a chamada ao LLM. Pode reduzir este gargalo fazendo múltiplas chamadas simultâneas às APIs do LLM usando as bibliotecas asyncio e aiohttp do Python.

Neste tutorial, você vai usar Python para construir fluxos de trabalho agentivos que fazem múltiplas chamadas a LLM de forma assíncrona. Isto permitirá que você crie aplicações LLM mais rápidas e precisas, que podem ser implementadas num Droplet ou numa instância sem servidor como um endpoint de API.

Vai criar um fluxo de trabalho de atendimento ao cliente para o sistema telefónico de uma agência imobiliária que possa responder a perguntas sobre anúncios, agendar compromissos e ligar os chamadores a representantes humanos. Fará isto através de múltiplas chamadas assíncronas a diferentes endpoints de LLM, reduzindo a latência do LLM e permitindo maior flexibilidade na escolha do LLM para cada tarefa.

Principais Conclusões

  • Usar chamadas assíncronas para modelos pequenos e rápidos pode ajudá-lo a criar fluxos de trabalho agentivos eficazes e com baixa latência para aplicações sensíveis ao tempo
  • Os fluxos de trabalho multimodelo permitem personalizar cada prompt para um modelo diferente, reduzindo as necessidades de computação globais e diminuindo o custo de hospedar localmente modelos LLM e de utilizar APIs de terceiros.

Pré-requisitos

Para completar este tutorial, você precisará de:

  • Um ambiente de desenvolvimento local para Python 3.
  • Acesso a uma API LLM. Neste tutorial, irá usar o modelo Mistral-Small-3.2-24B implementado num Droplet com GPU utilizando vLLM. Pode configurar um Droplet com GPU usando vLLM para servir um modelo seguindo o guia Como Executar GPT-OSS e vLLM num Droplet AMD GPU com ROCm.
  • Familiaridade com a sintaxe async/await do Python (opcional, mas útil).

Passo 1 — Configurar o Seu Ambiente

Primeiro, configure o seu ambiente e instale as dependências. Neste tutorial, serão feitas chamadas a um modelo Mistral-Small-3.2-24B que foi implementado num Droplet com GPU H200, mas você precisará fazer chamadas ao seu próprio modelo implementado ou a uma API LLM de terceiros. Este tutorial faz chamadas apenas a um LLM, mas poderá usar um modelo diferente de diversos fornecedores para cada prompt, se desejar.

Crie um ambiente virtual e instale as dependências necessárias:

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

Você fará as chamadas ao seu modelo de forma assíncrona, por isso precisa criar algumas funções assíncronas que possam gerir essas chamadas. Irá utilizar as bibliotecas asyncio e aiohttp em Python. Se estiver a usar uma API de LLM diferente, precisará de instalar o cliente correspondente para gerir as chamadas ao seu endpoint.

Crie um novo ficheiro chamado agentic_workflows.py e adicione o seguinte código para importar as dependências necessárias e definir o endpoint da API.

Substitua your_server_ip pelo endereço IP do seu Droplet GPU ou defina o cliente para a sua API LLM:

import asyncio
import aiohttp

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

Passo 2 — Escrever a Lógica da Chamada Assíncrona

Você quer que as suas chamadas assíncronas consigam receber uma lista de dicionários, cada um contendo o ID do modelo e o prompt que deseja enviar ao LLM, e enviá-los todos sem esperar pela resposta individual de cada chamada. O prompt incluirá o histórico da conversa e o prompt do sistema para essa chamada específica. Você irá adicionar uma função síncrona call_models que pode ser importada e chamada no seu script principal.

Adicione as seguintes funções assíncronas ao 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))

O método aiohttp.ClientSession() cria uma sessão de cliente assíncrona que envia pedidos POST para o endpoint da API e trata a resposta HTTP de forma assíncrona. Isto permite que múltiplas chamadas à API LLM sejam feitas concorrentemente em vez de sequencialmente.

A função _call_models_async() recolhe todos os detalhes de múltiplos pedidos que fornecemos e realiza todas as chamadas à API sem esperar por cada resposta. Depois, à medida que as respostas vão chegando, adiciona-as a uma lista pela mesma ordem em que foram enviadas. A função call_models() é uma função síncrona que torna mais fácil chamar as funções assíncronas.

Passo 3 — Escrever os Seus Prompts

Agora que já tem a funcionalidade básica escrita para fazer chamadas assíncronas a um endpoint de API, precisa de escrever os seus prompts. Terá três prompts de sistema separados. Um ajudará com questões de preços, outro com agendamento, e outro com a identificação de outras questões relacionadas com listagens, encaminhando para um representante humano. Pode adicionar mais, incluindo alguns com maior poder de decisão autónoma, mas estes três funcionarão por agora.

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

Nestas configurações de prompt do sistema, inclui o título do prompt para referência futura, o ID do modelo a que está a enviar esse prompt e a redação do próprio prompt do sistema ao qual o histórico da conversa será anexado.

Passo 4 — Processamento da Entrada do Utilizador através dos Modelos

Agora precisa de criar uma função que possa receber a conversa do utilizador, adicionar cada uma das instruções do sistema a ela e fazer as suas chamadas assíncronas ao modelo correto para cada instrução usando a função call_models() que escreveu acima. Este exemplo assume que o histórico da conversa é guardado no front-end no mesmo formato que a nossa API aceita e é enviado junto com a entrada mais recente do utilizador.

Adicione a seguinte função a 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)

Esta função cria uma lista de objetos dicionário para cada prompt do sistema que inclui o ID do modelo e o prompt do sistema + histórico da conversa. Em seguida, chama as funções assíncronas para invocar cada modelo para cada prompt do sistema utilizando a função call_models().

Após chamar os modelos, as respostas são recebidas e guardadas na variável prompt_responses na mesma ordem em que foram enviadas, por isso é necessário atribuir novas variáveis com as respostas reais dos modelos.

Continue a adicionar à função 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")]

Passo 5 — Definir a Lógica

Finalmente, você precisa adicionar uma camada lógica a esta função que possa analisar as suas respostas e determinar o que fazer a seguir. Pode consultar o preço de uma propriedade na sua base de dados, redirecionar a chamada para um representante humano, agendar uma reunião com um dos seus agentes imobiliários ou enviar uma pergunta de seguimento ao utilizador pedindo mais detalhes.

Continue a adicionar à função run_agentic_workflow em 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

Passo 6 — Testar o Seu Fluxo de Trabalho

Agora que já escreveu o código para um fluxo de trabalho agente básico para lidar com três categorias de perguntas, irá testá-lo com uma conversa sobre o preço de um anúncio. Nesta conversa, o utilizador não fornece o ID do anúncio na sua pergunta inicial, e o modelo responde com uma pergunta de seguimento a solicitá-lo. O utilizador fornece-o, e o modelo responde com o preço correto.

Crie um novo ficheiro chamado test_workflow.py e adicione o seguinte código de teste:

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

Execute o script de teste:

python3 test_workflow.py

Vais ver uma saída semelhante a esta:

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

Agora teste com uma conversa completa onde o utilizador fornece o ID do anúncio:

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

Execute o script novamente:

python3 test_workflow.py

Verá a informação de preços:

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

Conclusão

Neste tutorial, você construiu um fluxo de trabalho agente paralelo do zero em Python. Criou funções assíncronas para fazer chamadas simultâneas a LLM, configurou múltiplos prompts do sistema para diferentes tarefas e implementou uma camada de lógica de encaminhamento para lidar com os pedidos dos utilizadores.

Para tornar este framework pronto para produção, será necessário testar, reescrever prompts, alterar modelos e testar novamente com diferentes conversas até atingir um nível de precisão e rapidez que satisfaça os seus requisitos. Com este tipo de fluxos de trabalho, será necessário testar cada prompt individualmente e continuar a melhorá-los ao longo do tempo. No nosso exemplo, escolhemos o modelo Mistral-Small-3.2-24B devido à sua baixa latência (tempo de resposta médio inferior a 0,5 segundos), à sua capacidade de seguir estas instruções específicas e à sua possibilidade de ser implementado numa única GPU. Também testámos com GPT-OSS-120b, GPT-OSS-20b e Llama 3.1 8b, sendo que o modelo Mistral teve o melhor desempenho para este caso de utilização em particular. Escolher o modelo correto para cada prompt ou tarefa é um processo iterativo.

Se estiver interessado em utilizar agentes com mais autonomia, os prompts do sistema podem ter mais ou menos autonomia ao dar-lhes acesso a mais contexto ou mais ferramentas. Pode até ter chamadas subsequentes para o LLM “encadeadas” para raciocínio ou outros propósitos. No entanto, se um prompt necessitar de múltiplas chamadas de API em cadeia, isso aumentará a latência do fluxo de trabalho, pois deixa de estar apenas à espera de um único grupo de chamadas assíncronas aos modelos.

Implementámos o modelo Mistral com vLLM, por isso não precisámos de alterar o formato dos nossos prompts para que funcionassem com diferentes modelos. No entanto, a OpenAI, Claude e modelos open-source podem ter requisitos diferentes para a formatação dos prompts que são enviados para as suas APIs. Se usares diferentes fornecedores de modelos para cada chamada, serás obrigado a escrever funções em Python que possam converter o formato dos prompts para corresponder aos requisitos do modelo que estás a chamar. Por exemplo, podes precisar de uma função que consiga pegar num prompt formatado para vLLM e converter a sintaxe para um prompt formatado para Claude.

Você vai frequentemente perceber que alguns modelos demoram muito mais do que outros, fazendo com que tenha de esperar pela resposta mais lenta para receber todas as respostas. Com a biblioteca asyncio do Python, pode processar as respostas dos modelos à medida que chegam, usando asyncio.as_completed() em vez de asyncio.gather(). Depois, pode verificar as primeiras respostas para ver se são a direção que quer seguir e avançar sem esperar pelo restante das respostas. Isso também exigiria alterações na camada lógica.

Ao adicionar mais prompts do sistema, o principal obstáculo com que se vai deparar aqui é a limitação de taxa da API LLM ou os limites de processamento da sua GPU. A melhor forma de contornar isto é hospedar o seu próprio modelo num Droplet com GPU, mas também pode atenuar estes problemas aumentando as suas quotas de limite de taxa da API, distribuindo as suas chamadas entre várias APIs e implementando lógica de repetição, gestão de tempos de espera e tratamento de erros quando os limites de taxa são atingidos.

Obrigado por aprender com a comunidade Linux-Console.net.