Function Calling e Structured Outputs no Gemini 3.8: Contratos Rígidos para Agentes de Software

Por /
Diagrama técnico ilustrando a integração de Pydantic, Function Calling e o motor de inferência do Gemini 3.8 para validação de dados em agentes.

O Problema Prático e Contexto Direto

A integração de modelos de linguagem em pipelines de software exige determinismo na estrutura das respostas. Quando um agente autônomo consome payloads não estruturados ou textos livres em formato JSON não garantido, parsers de backend falham devido a inconsistências de tipos, chaves ausentes ou alucinações de formatação. Em arquiteturas de microsserviços, a quebra de um contrato de interface interrompe a execução do pipeline de automação e introduz estados inconsistentes em bancos de dados e filas de mensagens.

Para contornar essa fragilidade, o Gemini 3.8 disponibiliza dois recursos nativos acoplados ao ciclo de inferência: Structured Outputs (utilizando schemas JSON/Pydantic) e Function Calling (invocação declarativa de ferramentas). A aplicação de esquemas rígidos força o decodificador do modelo a gerar tokens que respeitem a gramática estipulada. Essa abordagem garante a interoperabilidade direta com modelos de dados tipados em Python, permitindo a orquestração segura descrita na Arquitetura de Agentes e Sub-agentes com Antigravity e Gemini 3.8: Guia Definitivo de Sistemas Autônomos em Produção.

Pré-requisitos e Instalação

A execução do código requer ambiente Python 3.10 ou superior, o SDK oficial google-genai e a biblioteca de validação pydantic.

Execute os comandos abaixo para configurar o ambiente virtual e instalar as dependências necessárias:

python3 -m venv venv
source venv/bin/activate
pip install google-genai pydantic

Defina a chave de API do Google Gemini na variável de ambiente antes de executar os scripts:

export GEMINI_API_KEY="Sua_Chave_API_Aqui"

Implementação Completa em Python (Código Funcional)

O código a seguir demonstra o padrão completo de implementação. Ele cobre a definição de estruturas tipadas com Pydantic, a aplicação do contrato via Structured Outputs no Gemini 3.8, a definição de funções para Function Calling e o tratamento explícito de exceções de validação (ValidationError).

Crie um arquivo chamado agent_contracts.py e insira a implementação completa:

import json
import os
import sys
import typing
from pydantic import BaseModel, Field, ValidationError
from google import genai
from google.genai import types

# ------------------------------------------------------------------------------
# 1. Definição do Modelo de Saída Estruturada via Pydantic
# ------------------------------------------------------------------------------
class TaskStep(BaseModel):
    step_number: int = Field(
        ge=1, 
        description="Número sequencial indicando a ordem de execução da etapa."
    )
    command: str = Field(
        description="Comando explícito a ser executado no ambiente de destino."
    )
    timeout_seconds: int = Field(
        ge=5, 
        le=300, 
        description="Tempo limite de execução do comando em segundos."
    )

class ExecutionPipeline(BaseModel):
    pipeline_id: str = Field(
        description="Identificador único em formato string para o pipeline."
    )
    target_environment: str = Field(
        pattern="^(production|staging|development)$",
        description="Ambiente de destino válido para a execução."
    )
    steps: typing.List[TaskStep] = Field(
        description="Lista sequencial contendo os passos do pipeline."
    )

# ------------------------------------------------------------------------------
# 2. Definição da Função para Function Calling NATIVO
# ------------------------------------------------------------------------------
def provision_container(image_name: str, port: int) -> str:
    """Provisiona um container no cluster de destino.

    Args:
        image_name: Nome da imagem do container no registry (ex: redis:7.0).
        port: Porta TCP exposta pelo serviço.
    """
    # Exemplo simulado de integração com API de orquestração de infraestrutura
    execution_result = {
        "status": "PROVISIONED",
        "image": image_name,
        "port": port,
        "container_id": "cnt-8f92a10c99"
    }
    return json.dumps(execution_result)

# ------------------------------------------------------------------------------
# 3. Execução de Structured Outputs com Validação de Contrato
# ------------------------------------------------------------------------------
def run_structured_output_pipeline(client: genai.Client) -> ExecutionPipeline:
    prompt = (
        "Crie um pipeline para implantação da aplicação Web. "
        "O ambiente deve ser 'production'. "
        "Passo 1: executar 'docker pull app:v2' com timeout de 60s. "
        "Passo 2: executar 'docker run -d -p 80:80 app:v2' com timeout de 30s."
    )

    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=ExecutionPipeline,
            temperature=0.0
        )
    )

    try:
        raw_json = response.text
        parsed_data = json.loads(raw_json)
        validated_pipeline = ExecutionPipeline.model_validate(parsed_data)
        return validated_pipeline
    except ValidationError as err:
        print(f"[ERRO] Falha de validação no schema Pydantic: {err}", file=sys.stderr)
        raise
    except json.JSONDecodeError as err:
        print(f"[ERRO] Resposta retornada não é um JSON válido: {err}", file=sys.stderr)
        raise

# ------------------------------------------------------------------------------
# 4. Execução de Function Calling Determinístico
# ------------------------------------------------------------------------------
def run_function_calling_pipeline(client: genai.Client) -> None:
    prompt = "Provisione um container Redis utilizando a imagem redis:7.0 na porta 6379."

    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(
            tools=[provision_container],
            temperature=0.0
        )
    )

    if response.function_calls:
        for call in response.function_calls:
            print(f"[FUNCTION CALL] Invocando função: {call.name}")
            print(f"[ARGUMENTOS]: {call.args}")

            if call.name == "provision_container":
                image = str(call.args["image_name"])
                port = int(call.args["port"])
                output = provision_container(image_name=image, port=port)
                print(f"[RESPOSTA DA FERRAMENTA]: {output}")
    else:
        print("[INFO] Nenhuma chamada de função foi disparada pelo modelo.")

# ------------------------------------------------------------------------------
# 5. Ponto de Entrada Principal
# ------------------------------------------------------------------------------
def main():
    api_key = os.environ.get("GEMINI_API_KEY")
    if not api_key:
        print("[ERRO CRÍTICO] Variável GEMINI_API_KEY ausente.", file=sys.stderr)
        sys.exit(1)

    client = genai.Client(api_key=api_key)

    print("=== Teste 1: Execução com Structured Output ===")
    pipeline = run_structured_output_pipeline(client)
    print(f"Pipeline ID: {pipeline.pipeline_id}")
    print(f"Ambiente Validado: {pipeline.target_environment}")
    print(f"Passos Validados ({len(pipeline.steps)}):")
    for step in pipeline.steps:
        print(f"  [{step.step_number}] {step.command} (Timeout: {step.timeout_seconds}s)")

    print("\n=== Teste 2: Execução com Function Calling ===")
    run_function_calling_pipeline(client)

if __name__ == "__main__":
    main()

Como Testar e Verificar a Saída

Para executar o script e verificar o contrato estático e o disparo de funções, execute o comando abaixo no terminal:

python agent_contracts.py

Exemplo de Saída Esperada no Console

=== Teste 1: Execução com Structured Output ===
Pipeline ID: pipe-prod-001
Ambiente Validado: production
Passos Validados (2):
  [1] docker pull app:v2 (Timeout: 60s)
  [2] docker run -d -p 80:80 app:v2 (Timeout: 30s)

=== Teste 2: Execução com Function Calling ===
[FUNCTION CALL] Invocando função: provision_container
[ARGUMENTOS]: {'image_name': 'redis:7.0', 'port': 6379.0}
[RESPOSTA DA FERRAMENTA]: {"status": "PROVISIONED", "image": "redis:7.0", "port": 6379, "container_id": "cnt-8f92a10c99"}

A validação do Pydantic lança uma exceção ValidationError antes que dados corrompidos passem para a camada de execução caso o modelo retorne tipos incompatíveis ou descumpra as expressões regulares (pattern).

Armadilhas Operacionais e Boas Práticas

1. Latência por Construtores Gramaticais

O uso de response_schema com restrições rígidas (expressões regulares complexas ou enumerações extensas) impõe uma penalidade de latência na fase de amostragem de tokens. O decodificador restringe o espaço de busca a cada token gerado. Em pipelines que exigem respostas sub-100ms, limite a complexidade das validações dentro do schema do modelo e efetue regras de negócio secundárias no código Python.

2. Mensagens Venenosas (Poison Messages) e Ciclos de Retry

Se o modelo gerar um payload que viole o schema Pydantic, o tratamento via try/except ValidationError deve ser acoplado a uma estratégia de re-prompting limitada (máximo de 2 a 3 tentativas). Evite reconsumir a mesma resposta em loops infinitos sem alterar o contexto ou ajustar a temperatura para 0.0.

3. Divergência de Tipos em Argumentos de Funções (Function Calling)

A API do Gemini pode serializar números inteiros como valores de ponto flutuante em JSON (ex: 6379.0 em vez de 6379). O código consumidor da chamada de função deve efetuar a coerção explícita de tipos (como feito em port = int(call.args["port"])) antes de repassar o argumento para subsistemas do SO ou drivers de banco de dados.

4. Controle de Timeouts e Deadlocks em Agentes

Chamadas externas de ferramentas acionadas por Function Calling devem ter timeouts estritos configurados no nível do cliente HTTP/gRPC. Ferramentas que bloqueiam a thread principal do agente paralisam a fila de execução do sistema autônomo.

Navegação da Trilha

Rolar para cima