Pesquisa de site

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

Introdução

Modelos de linguagem grandes podem ser desafiadores de usar em um ambiente de produção porque podem apresentar 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 de 3 a 15 segundos. Se você estiver construindo um agente sensível ao tempo, como um bot para conversas telefônicas, esse intervalo é longo demais para que o agente faça uma pausa.

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

Existem algumas maneiras diferentes de executar fluxos de trabalho agentes em paralelo, mas em muitos fluxos de trabalho agentes, o principal gargalo é a chamada para o LLM. Você pode reduzir esse gargalo fazendo múltiplas chamadas simultâneas para APIs de LLM usando as bibliotecas asyncio e aiohttp do Python.

Neste tutorial, você usará Python para construir fluxos de trabalho agentes que fazem múltiplas chamadas de LLM de forma assíncrona. Isso permitirá que você crie aplicativos de LLM mais rápidos e precisos, que podem ser implantados em uma Droplet ou instância sem servidor como um endpoint de API.

Você vai criar um fluxo de atendimento ao cliente para o sistema telefônico de uma imobiliária que possa responder perguntas sobre imóveis, agendar compromissos e conectar chamadas a representantes humanos. Isso será feito realizando múltiplas chamadas assíncronas a diferentes endpoints de LLM, reduzindo a latência do LLM e permitindo mais 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 de baixa latência para aplicações sensíveis ao tempo
  • Fluxos de trabalho multimodelo permitem que você personalize cada prompt para um modelo diferente, reduzindo suas necessidades gerais de computação e diminuindo o custo de hospedar modelos LLM por conta própria e de usar APIs de terceiros.

Pré-requisitos

Para completar este tutorial, você precisará de:

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

Passo 1 — Configurando Seu Ambiente

Primeiro, configure seu ambiente e instale suas dependências. Neste tutorial, serão feitas chamadas a um modelo Mistral-Small-3.2-24B que foi implantado em um Droplet com GPU H200, mas você precisará fazer chamadas para seu próprio modelo implantado ou para uma API LLM de terceiros. Este tutorial faz chamadas para apenas um LLM, mas você poderá usar um modelo diferente de provedores diferentes 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á suas chamadas ao modelo de forma assíncrona, então precisará criar algumas funções assíncronas que possam gerenciar essas chamadas. Você usará as bibliotecas asyncio e aiohttp no Python. Se estiver usando uma API de LLM diferente, será necessário instalar o cliente deles para gerenciar as chamadas ao endpoint.

Crie um novo arquivo 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 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 — Escrevendo a Lógica da Chamada Assíncrona

Você quer que suas chamadas assíncronas possam receber uma lista de dicionários, cada um contendo o ID do modelo e o prompt que você deseja enviar para o 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ê 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 requisições POST para o endpoint da API e lida com a resposta HTTP de forma assíncrona. Isso permite que múltiplas chamadas de API de LLM sejam feitas de forma concorrente em vez de sequencial.

A função _call_models_async() reúne todos os detalhes de múltiplas solicitações que fornecemos e faz todas as chamadas de API sem esperar por cada resposta. Em seguida, à medida que as respostas retornam, ela adiciona as respostas a uma lista na mesma ordem em que foram enviadas. A função call_models() é uma função síncrona que facilita a chamada das chamadas assíncronas.

Passo 3 — Escrevendo Seus Prompts

Agora que você escreveu a funcionalidade básica para fazer chamadas assíncronas a um endpoint de API, você precisa escrever seus prompts. Você terá três prompts de sistema separados. Um ajudará com perguntas sobre preços, outro com agendamento e outro com a identificação de outras perguntas sobre listagens, direcionando para um representante humano. Você pode adicionar mais, incluindo alguns com mais poder de tomada de decisão autônoma, mas esses três funcionarão por enquanto.

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

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

Etapa 4 — Processando a Entrada do Usuário pelos Modelos

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

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

Essa 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, ela chama as funções assíncronas para acionar cada modelo para cada prompt do sistema usando a função call_models().

Após chamar os modelos, as respostas são recebidas e salvas na variável prompt_responses na mesma ordem em que foram enviadas, então você precisa atribuir novas variáveis com as respostas reais dos modelos.

Continue adicionando à 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 — Definindo a Lógica

Finalmente, você precisa adicionar uma camada de lógica a esta função que possa analisar suas respostas e determinar o que fazer a seguir. Você pode consultar o preço de um imóvel em seu banco de dados, redirecionar a chamada para um representante humano, agendar uma reunião com um de seus corretores de imóveis ou enviar uma pergunta de acompanhamento ao usuário pedindo mais detalhes.

Continue adicionando à 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

Etapa 6 — Testando seu fluxo de trabalho

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

Crie um novo arquivo 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

Você 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 usuário 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

Você verá as informações de preço:

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. Você criou funções assíncronas para fazer chamadas simultâneas ao LLM, configurou múltiplos prompts do sistema para diferentes tarefas e implementou uma camada de lógica de roteamento para lidar com as solicitações dos usuários.

Para tornar este framework pronto para produção, você precisará testar, reescrever prompts, mudar de modelos e testar novamente com diferentes conversas até alcançar um nível de precisão e velocidade que atenda aos seus requisitos. Com esses tipos de fluxos de trabalho, você precisará testar cada prompt individualmente e continuar melhorando-os ao longo do tempo. Em nosso exemplo, escolhemos o modelo Mistral-Small-3.2-24B devido à sua baixa latência (tempo de resposta mediano de menos de 0,5 segundos), sua capacidade de seguir essas instruções específicas e sua capacidade de ser implantado em uma única GPU. Também testamos com GPT-OSS-120b, GPT-OSS-20b e Llama 3.1 8b, e o modelo Mistral teve o melhor desempenho para este caso de uso específico. Escolher o modelo correto para cada prompt ou tarefa é um processo iterativo.

Se você estiver interessado em usar agentes com mais autonomia, os prompts do sistema podem receber mais ou menos autonomia ao lhes dar acesso a mais contexto ou a mais ferramentas. Você pode até fazer chamadas de acompanhamento para o LLM "encadeadas" juntas para raciocínio ou outros propósitos. No entanto, se um prompt exigir múltiplas chamadas de API em cadeia, isso aumentará a latência do fluxo de trabalho, porque você não estará mais apenas esperando por um grupo de chamadas assíncronas aos modelos.

Implantamos o modelo Mistral com o vLLM, então não precisávamos alterar o formato de nossos prompts para que funcionassem com diferentes modelos. No entanto, OpenAI, Claude e modelos de código aberto podem ter requisitos diferentes para a formatação dos prompts que são enviados para suas APIs. Se você usar provedores de modelos diferentes para cada chamada, será necessário escrever funções em Python que possam converter o formato dos prompts para corresponder aos requisitos do modelo que você está chamando. Por exemplo, pode ser necessário uma função que possa pegar um prompt formatado para vLLM e converter a sintaxe para um prompt formatado para Claude.

Você frequentemente perceberá que alguns modelos levam muito mais tempo que outros, fazendo você esperar pela resposta mais lenta para receber todas as respostas. Com a biblioteca asyncio do Python, você pode processar as respostas dos modelos à medida que elas chegam, usando asyncio.as_completed() em vez de asyncio.gather(). Então, você pode verificar as primeiras respostas para ver se elas são o caminho que você deseja seguir e avançar sem esperar pelas demais respostas. Isso também exigiria alterações na camada de lógica.

Ao adicionar mais prompts do sistema, o principal gargalo que você encontrará aqui é o limite de taxa na API do LLM ou os limites de processamento da sua GPU. A melhor forma de contornar isso é hospedar seu próprio modelo em um Droplet com GPU, mas você também pode mitigar esses problemas aumentando suas cotas de limite de taxa da API, distribuindo suas chamadas entre APIs e implementando lógica de tentativa, gerenciamento de tempo limite e tratamento de erros quando os limites de taxa forem atingidos.

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