Construindo o Xabironelson Codex

Até agora fizemos a configuração inicial do Xabiro. Bom, neste post aqui, construiremos a camada que interagirá com a LLM.

Vamos tentar não complicar o que deve ser simples. Imaginemos três camadas: uma será responsável por lidar com as interações do usuário; outra, por concentrar os objetos que modelarão as “regras de negócio”; e a terceira, por lidar com os provedores de dados

Qual é o benefício disso?

Eu sei que toda aplicação deve levar em consideração uma arquitetura que resolva os desafios relacionados ao seu contexto. Maaas, eu sou fã do Clean, e será ela mesma.

Diante de tudo isso, focaremos na comunicação com a LLM. Para simplificarmos nossa vida, trataremos a LLM como um “Datasource”.

Nossa maritaca

Reduzir a LLM a uma maritaca tem o mesmo efeito prático de reduzi-la a um DataSource. Porém, “maritaca” é um exemplo mais legal, além de fazermos uma propaganda a Maritaca AI.

Mas, como faremos para nos comunicar com os provedores de LLM? Agora vamos usar a decisão que tomamos quando ponderamos sobre “make or buy”. No entanto, iremos mapear o objeto para uma entidade do nosso domínio para não ficarmos acoplados ao Lite… Pera, mas que Lite?

El Lite 🥸

Cada provedor vai puxar a sardinha para o seu lado e tentar te jogar em um vendor lock-in.

Sobre vendor lock-in

Eu sei que Vendor Lock-In não é exatamente isso. Mas entenda que estou tentando lhe ajudar a não sair colocando objetos de terceiros ao longo da sua aplicação, para que você não sofra caso haja uma breaking change. Vulgo, to tentando salvar sua sexta.

Podemos imaginar o Lite como um tradutor. Em vez de aprendermos cinco idiomas diferentes, nós falamos com nosso tradutor, e ele faz o resto da comunicação.

graph TD
    subgraph A[Aplicação do Usuário]
        AP(Camada de Dados)
    end

    LiteLLM_Proxy[LiteLLM]

    subgraph C[APIs de Modelos de Linguagem]
        OpenAI[API OpenAI]
        Anthropic[API Anthropic]
        Diversos[API Xabiro]
    end

    AP --> |Request| LiteLLM_Proxy
    
    LiteLLM_Proxy --> |Request| OpenAI
    LiteLLM_Proxy --> |Request| Anthropic
    LiteLLM_Proxy --> |Request| Diversos

    OpenAI --> |Response| LiteLLM_Proxy
    Anthropic --> |Response| LiteLLM_Proxy
    Diversos --> |Response| LiteLLM_Proxy

    LiteLLM_Proxy --> |Response| AP

Além de facilitar a comunicação, o Lite nos fornece tracking de custo, técnicas para recuperação de erros e self-hosting.

Na concepção do Xabiro, utilizaremos essa facilidade de comunicação, as técnicas de failover automático e a visualização do custo por iteração.

Antes de começarmos a codificar, vamos desenhar as interações entre os componentes do nosso sistema:

sequenceDiagram
    box Camada de Apresentação (Presentation - CLI)
        actor Usuario
        participant CLI as "CLI - Interface"
    end

    box Camada de Domínio (Domain)
        participant UseCase as "GenerateCompletionUseCase"
        participant IRepository as "LLMRepository (Interface)"
    end

    box Camada de Dados (Data - Infraestrutura)
        participant RepositoryImpl as "LLMRepositoryImpl"
        participant LiteLLMClient as "LiteLLMClient"
        participant ResponseMapper as "ResponseMapper"
        participant LiteLLM as "LiteLLM (Biblioteca)"
    end

    box Provedores Externos
        participant LLMExterna as "API LLM (Gemini)"
    end

    Usuario->>CLI: python main.py solve "Quanto é 2+2?"
    activate CLI
    
    Note over CLI: Carrega .env, valida config YAML, verifica API key, inicializa dependências.
    CLI->>UseCase: execute("Quanto é 2+2?")
    activate UseCase

    Note over UseCase: Lógica de aplicação (validações futuras).
    UseCase->>IRepository: complete(user_input)
    
    Note over IRepository: Contrato do domínio para conclusão.
    
    IRepository->>RepositoryImpl: complete(user_input)
    activate RepositoryImpl
    
    Note over RepositoryImpl: Formata prompt com template, adiciona à memória.
    RepositoryImpl->>LiteLLMClient: complete(messages)
    activate LiteLLMClient

    Note over LiteLLMClient: Prepara chamada para LiteLLM.
    LiteLLMClient->>LiteLLM: completion(model, messages, ...)
    activate LiteLLM

    LiteLLM->>LLMExterna: Requisição HTTP (ex: POST para Gemini API)
    activate LLMExterna

    LLMExterna-->>LiteLLM: Resposta JSON
    deactivate LLMExterna

    LiteLLM-->>LiteLLMClient: CompletionResponseDTO
    deactivate LiteLLM

    LiteLLMClient-->>RepositoryImpl: CompletionResponseDTO
    deactivate LiteLLMClient

    Note over RepositoryImpl: Adiciona resposta à memória, mapeia para domínio.
    RepositoryImpl->>ResponseMapper: to_domain(dto)
    activate ResponseMapper
    ResponseMapper-->>RepositoryImpl: ResponseModel (Domínio)
    deactivate ResponseMapper

    RepositoryImpl-->>UseCase: ResponseModel
    deactivate RepositoryImpl

    Note over UseCase: Processamento final (se necessário).
    UseCase-->>CLI: ResponseModel
    deactivate UseCase
    
    Note over CLI: Exibe conteúdo e tokens.
    CLI-->>Usuario: Resposta formatada
    deactivate CLI

Com tudo desenhado, sigamos para o código. Primeiro, vamos adicionar a dependência do litellm

uv add litellm

As nossas camadas

O objetivo primordial da arquitetura é ~~facilitar a vida na sexta-feira permitindo que seu sistema seja testável~~ minimizar os recursos humanos necessários para construir e manter um sistema¹. Ao isolarmos o código, garantimos que as regras de negócio sejam independentes de frameworks, fazendo com que alterações nas tecnologias usadas não exijam que trabalhemos no sábado.

A camada de dados

Comecemos modelando nosso Repository

from litellm import completion
from litellm.exceptions import APIError, AuthenticationError
from litellm.types.utils import ModelResponse
from domain.boundary.llm_repository import LLMRepository
from domain.models.model_errors import LLMUnavailableError
from domain.models.response_model import ResponseModel
 
class LLMRepositoryImpl(LLMRepository):
	# TODO: Tem um problema aqui relacionado a testabilidade e de SRP
    def __init__(self, llm_config, prompt_template: str):
        self._llm_config = llm_config
        self._prompt = prompt_template
        self._short_term_memory = []
 
    def complete(self, user_input: str) -> ResponseModel:
        self._short_term_memory.append({"role": "user", "content": self._prompt.format(user_input=user_input)})
        try:
            response = completion(
                model=self._llm_config.model,
                messages=self._short_term_memory,
                max_tokens=self._llm_config.max_tokens,
                temperature=self._llm_config.temperature,
            )
 
            assert(response is ModelResponse)
            content = response.choices[0].message.content
            tokens = response.usage.total_tokens
            return ResponseModel(
                content=content,
                tokens_used=tokens,
                
            )
        except (AuthenticationError, APIError) as e:
            raise LLMUnavailableError("O serviço LLM está indisponível no momento.") from e

A camada de domínio

A interface do nosso Repository:

from abc import ABC, abstractmethod
from domain.models.response_model import ResponseModel
 
class LLMRepository(ABC):
    """Abstract base class for LLM repository."""
 
    @abstractmethod
    def complete(self, user_input: str) -> ResponseModel:
        """Generate a completion for the given prompt."""
        pass

Nosso modelo de resposta da LLM

from pydantic import BaseModel
 
class ResponseModel(BaseModel):
    content: str
    tokens_used: int

Nossos modelos de erro

class ModelErros(Exception):
    """Base class for exceptions in this module."""
    pass
 
class LLMUnavailableError(ModelErros):
    """Exception raised when the LLM service is unavailable."""
    def __init__(self, message="LLM service is unavailable"):
        self.message = message
        super().__init__(self.message)

Nosso UseCase:

from domain.boundary.llm_repository import LLMRepository
from domain.models.model_errors import LLMUnavailableError
from utils.logger import Logger
 
class GenerateCompletionUseCase:
    def __init__(self, repository: LLMRepository, logger: Logger):
        self._repository = repository
        self._logger = logger
 
    def execute(self, user_input: str):
        self._logger.info("Processando entrada do usuário.", context={"length": len(user_input)})
        # TODO: Podemos aplicar nossas GuardRails aqui. Logo faremos isso.
        try:
            completion = self._repository.complete(user_input)
            self._logger.info("Geração de conclusão bem-sucedida.", context={"completion_length": len(completion.content)})
            return completion
        except LLMUnavailableError as e:
            self._logger.error("Serviço LLM indisponível.", context={"error": str(e)})
            raise e

A camanda de apresentação

Por agora, não criaremos um Container para recuperar nossas dependências. Vamos no bom e velho “abre e fecha”:

import os
from pathlib import Path
 
import typer
import yaml
from dotenv import load_dotenv
 
from constants import CONFIG_FILE_NAME, DEFAULT_CONFIG
from data.client.lite_llm_client import LiteLLMClient
from data.repository.llm_repository import LLMRepositoryImpl
from domain.models.model_errors import LLMError
from domain.use_case.generate_completion_use_case import GenerateCompletionUseCase
from models.config import LLMConfig
from utils.logger import BasicLogger
 
app = typer.Typer(help="Xabironelson Codex")
 
 
@app.callback()
def main_callback():
    pass
 
 
@app.command()
def test():
    typer.echo("Hello from Xabironelson Codex")
 
 
@app.command()
def init():
    config_path = Path.cwd() / CONFIG_FILE_NAME
    if config_path.exists():
        typer.confirm(
            f"O arquivo {CONFIG_FILE_NAME} já existe. Deseja sobrescrevê-lo?",
            abort=True,
        )
    with open(config_path, "w") as f:
        yaml.dump(DEFAULT_CONFIG, f, sort_keys=False)
 
    typer.secho(
        f"Arquivo de configuração '{CONFIG_FILE_NAME}' criado com sucesso!",
        fg=typer.colors.GREEN,
    )
 
 
@app.command()
def solve(
    task: str = typer.Argument(
        ..., help="Descrição da tarefa que o agente deve realizar."
    ),
    config_file: Path = typer.Option(
        CONFIG_FILE_NAME,
        "--config",
        "-c",
        exists=True,
        file_okay=True,
        dir_okay=False,
        writable=False,
        help="Caminho para o arquivo de configuração.",
    ),
):
    typer.secho("Carregando variáveis de ambiente...", fg=typer.colors.BLUE)
    load_dotenv()
 
    typer.secho(f"Carregando configurações de: {config_file}", fg=typer.colors.CYAN)
 
    try:
        with open(config_file, "r") as f:
            config = yaml.safe_load(f)
    except yaml.YAMLError as e:
        typer.secho(
            f"Erro ao carregar o arquivo de configuração: {e}", fg=typer.colors.RED
        )
        raise typer.Exit(code=1)
    except FileNotFoundError:
        typer.secho(
            f"Erro: Arquivo de configuração '{config_file}' não encontrado.",
            fg=typer.colors.RED,
        )
        raise typer.Exit(code=1)
 
    llm_configuration = config.get("llm", {})
    prompt_config = config.get("prompt", {})
    verbose = config.get("verbose", False)
 
    env_var_name = llm_configuration.get("api_key_env", "LLM_API_KEY")
    api_key = os.getenv(env_var_name)
 
    if not api_key:
        typer.secho("\nERRO: Chave API não encontrada!", fg=typer.colors.RED)
        typer.echo(f"A variável de ambiente '{env_var_name}' não foi definida.")
        typer.echo("Certifique-se de criar um arquivo .env ou exportar a variável.")
        raise typer.Exit(code=1)
 
    typer.echo("\n[CONFIGURAÇÕES CARREGADAS]")
    typer.secho(f" Tarefa: {task}", fg=typer.colors.YELLOW)
 
    if llm_configuration:
        typer.echo(f" Modelo: {llm_configuration.get('model', 'N/A')}")
        typer.echo(f" Temperatura: {llm_configuration.get('temperature', 'N/A')}")
        typer.echo(f" Max Tokens: {llm_configuration.get('max_tokens', 'N/A')}")
        typer.secho(
            f" Chave API carregada de: {env_var_name} (OK)", fg=typer.colors.GREEN
        )
        typer.echo(f" Modo Verbose: {verbose}")
 
    typer.echo("\n[INICIALIZANDO SISTEMA]")
    typer.secho("Configurando dependências manualmente...", fg=typer.colors.BLUE)
 
    logger = BasicLogger()
 
    llm_config = LLMConfig(
        model=llm_configuration.get("model", "gpt-4"),
        temperature=llm_configuration.get("temperature", 0.7),
        max_tokens=llm_configuration.get("max_tokens", 1500),
        api_key_env=llm_configuration.get("api_key_env", "LLM_API_KEY"),
    )
 
    llm_config_adapter = llm_config
 
    llm_client = LiteLLMClient(
        llm_config=llm_config_adapter,
        logger=logger,
        prompt=prompt_config,
    )
 
    repository = LLMRepositoryImpl(
        llm_client=llm_client,
        logger=logger,
    )
 
    use_case = GenerateCompletionUseCase(
        repository=repository,
        logger=logger,
    )
 
    typer.secho("Sistema inicializado com sucesso!", fg=typer.colors.GREEN)
    typer.echo("\n[EXECUTANDO TAREFA]")
 
    try:
        typer.secho("Processando...", fg=typer.colors.CYAN)
        result = use_case.execute(task)
 
        typer.echo("\n[RESULTADO]")
        typer.secho(result.content, fg=typer.colors.GREEN)
        typer.echo(f"\nTokens utilizados: {result.tokens_used}")
 
    except LLMError as e:
        typer.secho(f"\n[ERRO LLM] {e.message}", fg=typer.colors.RED)
        if verbose and e.cause:
            typer.echo(f"Causa: {e.cause}")
        raise typer.Exit(code=1)
    except Exception as e:
        typer.secho(f"\n[ERRO] {str(e)}", fg=typer.colors.RED)
        if verbose:
            import traceback
 
            typer.echo(traceback.format_exc())
        raise typer.Exit(code=1)

Tá quase pronto, só falta testar

Quem nunca falou essa na daily?

Imagem

Em alguma parte do texto, mencionei que a arquitetura é a facilitadora de nossos testes.

Teste garante qualidade?

Puta treta boa! A meu ver, não. Teste garante comportamento, que, se garantido ao longo do tempo, cria a percepção de qualidade.

Se continuarmos com o Repository da forma que está, teremos um problema na hora de realizar o teste por conta do El Lite 🥸. Existe alguma forma de testar com ele ali, diretamente? Provavelmente, mas vamos utilizar a mágica da arquitetura para nos salvar.

Criando nosso Client

Vamos começar do começo! Para conseguirmos testar nosso Repository teremos de aplicar DIP para removermos o completion

from abc import ABC, abstractmethod
from litellm import completion
from litellm.exceptions import APIError, AuthenticationError, RateLimitError, Timeout
from data.dtos import CompletionResponseDTO
from domain.boundary.llm_configuration import LLMConfiguration
from domain.models.model_errors import (
	LLMAuthenticationError,
	LLMRateLimitError,
	LLMTimeoutError,
	LLMUnavailableError,
)
from utils.logger import Logger
 
class LLMClient(ABC):
	"""Abstract base class for LLM client."""
 
	@abstractmethod
	def complete(self, messages: list[dict[str, str]]) -> CompletionResponseDTO:
		"""Generate a response from the model based on the given messages."""
		pass
 
class LiteLLMClient(LLMClient):
	def __init__(
		self, 
		llm_config: LLMConfiguration, 
		logger: Logger,
	):
		self._llm_config = llm_config
		self._logger = logger
 
	def complete(
		self, 
		messages: List[Dict[str, str]],
	) -> CompletionResponseDTO:
		try:
			self._logger.info(
				"Calling LLM completion",
				context={
					"model": self._llm_config.get_model(),
					"message_count": len(messages),
				},
			)
		
			response = completion(
				model=self._llm_config.get_model(),
				messages=messages,
				max_tokens=self._llm_config.get_max_tokens(),
				temperature=self._llm_config.get_temperature(),
				timeout=self._llm_config.get_timeout_seconds(),
				num_retries=self._llm_config.get_retry_config().get("max_retries", 3),
			)
		
		  
			content = response.choices[0].message.content
			tokens = response.usage.total_tokens
			finish_reason = response.choices[0].finish_reason
		
			self._logger.info(
				"LLM completion successful",
				context={
					"tokens_used": tokens, 
					"finish_reason": finish_reason
				},
			)
		
			return CompletionResponseDTO(
				content=content,
				tokens_used=tokens,
				model=self._llm_config.get_model(),
				finish_reason=finish_reason,
				raw_response=(
					response.model_dump() if hasattr(response, "model_dump") else None
				),
			)
		
		except AuthenticationError as e:
			self._logger.error("LLM authentication failed", context={"error": str(e)})
			raise LLMAuthenticationError(
				"Falha na autenticação com o serviço LLM. Verifique suas credenciais.",
				cause=e,
			)
		
		except RateLimitError as e:
			self._logger.error("LLM rate limit exceeded", context={"error": str(e)})
			raise LLMRateLimitError(
				"Limite de taxa excedido. Tente novamente mais tarde.", cause=e
			)
	
		except Timeout as e:
			self._logger.error("LLM request timed out", context={"error": str(e)})
			raise LLMTimeoutError(
				f"A requisição excedeu o tempo limite de {self._llm_config.get_timeout_seconds()} segundos.",
				cause=e,
			)
		
		except APIError as e:
			self._logger.error("LLM API error", context={"error": str(e)})
			raise LLMUnavailableError(
				"O serviço LLM está indisponível no momento.", cause=e
			)
	
		except Exception as e:
			self._logger.error(
				"Unexpected error during LLM completion",
				context={"error": str(e), "error_type": type(e).__name__},
			)
			raise LLMUnavailableError(
				f"Erro inesperado ao chamar o serviço LLM: {str(e)}", cause=e
			)

Com esse carinha feito, nosso Repository fica

from data.client.lite_llm_client import LLMClient
from data.mappers import ResponseMapper
from domain.boundary.llm_repository import LLMRepository
from domain.models.response_model import ResponseModel
from utils.logger import Logger
 
class LLMRepositoryImpl(LLMRepository):
	def __init__(
		self,
		llm_client: LLMClient,
		prompt_template: str,
		logger: Logger,
	):
		self._llm_client = llm_client
		self._prompt_template = prompt_template
		self._logger = logger
		self._short_term_memory: list[dict[str, str]] = []
 
	def complete(self, user_input: str) -> ResponseModel:
		formatted_prompt = self._prompt_template.format(user_input=user_input)
		self._short_term_memory.append({
			"role": "user", 
			"content": formatted_prompt
		})
		
		self._logger.info(
			"Generating completion",
			context={"memory_size": len(self._short_term_memory)},
		)
		
		response_dto = self._llm_client.complete(messages=self._short_term_memory)
		
		self._short_term_memory.append({
			"role": "assistant", 
			"content": response_dto.content
		})
		
		domain_response = ResponseMapper.to_domain(response_dto)
		
		self._logger.info(
			"Completion generated successfully",
			context={
				"tokens_used": domain_response.tokens_used,
				"memory_size": len(self._short_term_memory),
			},
		)
		
		return domain_response

Nosso mapper

from typing import Optional
 
from data.dtos import CompletionResponseDTO
from domain.models.response_model import ResponseModel
 
class ResponseMapper:
	@staticmethod
	def to_domain(dto: CompletionResponseDTO) -> ResponseModel:
		return ResponseModel(
			content=dto.content,
			tokens_used=dto.tokens_used,
		)
 
	@staticmethod
	def to_dto(
		domain_model: ResponseModel,
		model: str,
		finish_reason: Optional[str] = None,
	) -> CompletionResponseDTO:
		return CompletionResponseDTO(
			content=domain_model.content,
			tokens_used=domain_model.tokens_used,
			model=model,
			finish_reason=finish_reason,
		)

Se você percebeu, adicionamos dois novos modelos em nossa camada de dados

from typing import Any, Optional
from pydantic import BaseModel, Field
 
class MessageDTO(BaseModel):
	role: str = Field(
		..., 
		description="The role of the message sender (user/assistant/system)"
	)
	content: str = Field(
		..., 
		description="The content of the message"
	)
 
class CompletionResponseDTO(BaseModel):
	content: str = Field(
		..., 
		description="The generated content"
	)
	tokens_used: int = Field(
		..., 
		description="Number of tokens used in the completion"
	)
	model: str = Field(
		..., 
		description="The model used for completion"
	)
	finish_reason: Optional[str] = Field(
		None, 
		description="Reason why the completion finished"
	)
	raw_response: Optional[dict[str, Any]] = Field(
		None, 
		description="Raw response from the LLM API"
	)

Esses modelos serão necessários para que possamos isolar nossa aplicação do Lite.

Habemus as primeiras palavras

Vamo lá, bora fazer o Xabiro falar

uv run main.py "Quanto é 2+2?

O resultado

 Tarefa: Quanto é 2+2?
 Modelo: gemini/gemini-2.5-flash
 Temperatura: 0.7
 Max Tokens: 1500
 Chave API carregada de: LLM_API_KEY (OK)
 Modo Verbose: False
 
[INICIALIZANDO SISTEMA]
Configurando dependências manualmente...
Sistema inicializado com sucesso!
 
[EXECUTANDO TAREFA]
Processando...
2025-10-13 10:22:20,519 - INFO - Processing user input. | Context: {'length': 13}
2025-10-13 10:22:20,519 - INFO - Generating completion | Context: {'memory_size': 1}
2025-10-13 10:22:20,519 - INFO - Calling LLM completion | Context: {'model': 'gemini/gemini-2.5-flash', 'message_count': 1, 'prompt': 'Você é um agente de código pessoal chamado Xabironelson Codex. Sua tarefa é ajudar o usuário a escrever, revisar e depurar código. Sempre que possível, forneça exemplos de código e explique conceitos técnicos de maneira clara e concisa.'}
10:22:20 - LiteLLM:INFO: utils.py:3386 - 
LiteLLM completion() model= gemini-2.5-flash; provider = gemini
2025-10-13 10:22:20,521 - INFO - 
LiteLLM completion() model= gemini-2.5-flash; provider = gemini
10:22:21 - LiteLLM:INFO: utils.py:1295 - Wrapper: Completed Call, calling success_handler
2025-10-13 10:22:21,525 - INFO - Wrapper: Completed Call, calling success_handler
2025-10-13 10:22:21,527 - INFO - LLM completion successful | Context: {'tokens_used': 41, 'finish_reason': 'stop'}
2025-10-13 10:22:21,528 - INFO - Completion generated successfully | Context: {'tokens_used': 41, 'memory_size': 2}
2025-10-13 10:22:21,528 - INFO - Completion generation successful. | Context: {'completion_length': 20}
 
[RESULTADO]
2+2 é igual a **4**.
 
Tokens utilizados: 41

Mas e agora?

A partir de agora, os textos ficarão mais densos e cheios de código. Por conta disso, e pelo feedback do LT, abrirei um GitHub e poderemos acompanhar todas as alterações relacionadas a um post por meio do PR associado a ele.

Nos próximos textos, começaremos a implementar o REPL, o uso de ferramentas e o aumento da memória da nossa Dolly. Além disso, discutiremos se faz sentido nosso Repository ter a responsabilidade de gerenciar a memória curta.

Até lá!

Referências

MARTIN, Robert C. (Uncle Bob). Arquitetura limpa: O guia do artesão para estrutura e design de software. Tradução de Marco S. P. Ferreira. Rio de Janeiro: Alta Books, 2019. ↩

Cake Notes

Explorador

Construindo o Xabironelson Codex - Parte 4

Nossa maritaca

El Lite 🥸

As nossas camadas

A camada de dados

A camada de domínio

A camanda de apresentação

Tá quase pronto, só falta testar

Criando nosso Client

Habemus as primeiras palavras

Mas e agora?

Referências

Visão de gráfico

Sumário

Cake Notes

Explorador

Construindo o Xabironelson Codex - Parte 4

Nossa maritaca

El Lite 🥸

As nossas camadas

A camada de dados

A camada de domínio

A camanda de apresentação

Tá quase pronto, só falta testar

Criando nosso Client

Habemus as primeiras palavras

Mas e agora?

Referências

Footnotes

Visão de gráfico

Sumário