Do Notebook ao Sistema de Producao
Criar um agente IA que funciona no notebook e uma coisa. Coloca-lo em producao com alta disponibilidade, monitoramento e escalabilidade e outra. Este tutorial cobre o caminho completo.
---
1. Estrutura do Projeto
meu-agente/
|-- app/
| |-- main.py # FastAPI app
| |-- agent.py # LangChain agent
| |-- monitoring.py # Prometheus metrics
|-- Dockerfile
|-- docker-compose.yml
+-- requirements.txt---
2. O Agente (agent.py)
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferWindowMemory
from langchain.tools import tool
@tool
def buscar_informacao(query: str) -> str:
# Integre com sua fonte de dados real
return f"Resultado para: {query}"
@tool
def calcular(expressao: str) -> str:
try:
allowed = set('0123456789+-*/.() ')
return str(eval(expressao)) if all(c in allowed for c in expressao) else 'Invalido'
except Exception as e:
return f'Erro: {e}'
def criar_agente(session_id):
llm = ChatOpenAI(model='gpt-4o-mini', temperature=0.1, request_timeout=30, max_retries=2)
prompt = ChatPromptTemplate.from_messages([
('system', 'Voce e um assistente especializado. Responda em portugues.'),
MessagesPlaceholder(variable_name='chat_history'),
('human', '{input}'),
MessagesPlaceholder(variable_name='agent_scratchpad'),
])
tools = [buscar_informacao, calcular]
agent = create_openai_tools_agent(llm, tools, prompt)
return AgentExecutor(
agent=agent, tools=tools,
memory=ConversationBufferWindowMemory(k=10, return_messages=True, memory_key='chat_history'),
verbose=False, max_iterations=5, handle_parsing_errors=True,
)---
3. A API (main.py)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import asyncio, time, uuid, logging
from .agent import criar_agente
from .monitoring import REQUESTS_TOTAL, LATENCY_HISTOGRAM, REQUEST_IN_PROGRESS
logger = logging.getLogger(__name__)
app = FastAPI(title='Meu Agente IA', version='1.0.0')
class ChatRequest(BaseModel):
message: str = Field(..., min_length=1, max_length=4000)
session_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
class ChatResponse(BaseModel):
response: str
session_id: str
latency_ms: float
@app.get('/health')
async def health_check():
return {'status': 'healthy', 'version': '1.0.0'}
@app.post('/chat', response_model=ChatResponse)
async def chat(request: ChatRequest):
REQUEST_IN_PROGRESS.inc()
start = time.perf_counter()
try:
agente = criar_agente(request.session_id)
resultado = await asyncio.get_event_loop().run_in_executor(
None, lambda: agente.invoke({'input': request.message})
)
latencia = (time.perf_counter() - start) * 1000
REQUESTS_TOTAL.labels(status='success').inc()
return ChatResponse(response=resultado['output'],
session_id=request.session_id, latency_ms=round(latencia, 2))
except Exception as e:
REQUESTS_TOTAL.labels(status='error').inc()
logger.error(f'Erro: {e}', exc_info=True)
raise HTTPException(status_code=500, detail='Erro interno')
finally:
REQUEST_IN_PROGRESS.dec()---
4. Monitoramento (monitoring.py)
from prometheus_client import Counter, Histogram, Gauge, make_asgi_app
REQUESTS_TOTAL = Counter('agent_requests_total', 'Total requests', ['status'])
LATENCY_HISTOGRAM = Histogram('agent_request_duration_seconds', 'Latency',
buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 30.0])
REQUEST_IN_PROGRESS = Gauge('agent_requests_in_progress', 'In progress')
metrics_app = make_asgi_app()
app.mount('/metrics', metrics_app)---
5. Dockerfile
FROM python:3.12-slim
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends curl \\
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app/ ./app/
RUN useradd -m -u 1000 agente && chown -R agente:agente /app
USER agente
HEALTHCHECK --interval=30s --timeout=10s --start-period=30s \\
CMD curl -f http://localhost:8000/health || exit 1
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]---
6. Docker Compose
version: "3.9"
services:
agente:
build: .
ports: ["8000:8000"]
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- REDIS_URL=redis://redis:6379
depends_on:
redis:
condition: service_healthy
restart: unless-stopped
deploy:
resources:
limits: {memory: 512M, cpus: "0.5"}
redis:
image: redis:7-alpine
command: redis-server --save 60 1 --maxmemory 128mb
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 5s
retries: 3
restart: unless-stopped
prometheus:
image: prom/prometheus:latest
volumes: ["./prometheus.yml:/etc/prometheus/prometheus.yml"]
ports: ["9090:9090"]---
7. Deploy
Railway (mais simples):
npm i -g @railway/cli
railway login && railway init && railway up
railway variables set OPENAI_API_KEY=sk-...VPS com Docker:
docker-compose up -d
curl http://localhost:8000/health
docker-compose logs -f agente---
Checklist de Producao
- Variaveis de ambiente via .env (nunca hardcoded)
- Health check endpoint respondendo
- Logging estruturado JSON configurado
- Rate limiting ativo (max 10 req/min por IP)
- Timeout de 30s nas chamadas ao LLM
- Retry logic com backoff exponencial
- Metricas Prometheus em /metrics
- Docker com usuario nao-root
- Limites de memoria/CPU definidos
Parabens - voce chegou ao fim da Trilha Python IA com habilidades para construir, otimizar e colocar em producao sistemas completos de IA.