Padrões de Implantação em Produção
Implemente aplicações chatflow-agent em produção, serviços ASGI com FastAPI, padrões de concorrência no SQLite, conteinerização com Docker e gestão de processos.
A implantação de enxames do chatflow-agent em ambientes de produção exige uma arquitetura que equilibre operações de E/S não bloqueantes, concorrência segura de sessões e resiliência operacional.
Este guia detalha os padrões recomendados de implantação, abrangendo a integração ASGI com FastAPI, considerações de armazenamento no SQLite, conteinerização com Docker e monitoramento por verificações de integridade (health checks).
Esquema Arquitetural
Tráfego de Entrada (Clientes / Webhooks)
│
▼
Reverse Proxy (Nginx / Cloudflare / Traefik)
├── Terminação TLS (HTTPS / WSS)
├── Timeout de Requisição >= 120s
└── Rate Limiting e Proteção DDoS
│
▼
Servidor de Aplicação ASGI (Uvicorn / Gunicorn)
├── Processo único ou Multi-Worker com Armazenamento Isolado/Compartilhado
├── Aplicação FastAPI / Starlette
└── chatflow-agent Runner
├── SQLiteSessionStore (Modo WAL) ou Armazenamento Distribuído Customizado
└── Despacho Concorrente de TarefasIntegração como Serviço ASGI com FastAPI
O chatflow-agent se integra de forma transparente ao FastAPI. Ao utilizar runner.run_async(), as requisições HTTP de entrada executam de maneira concorrente no event loop sem bloquear o servidor:
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from chatflow_agent import Agent, Runner
from chatflow_agent.core.memory import SQLiteSessionStore
class ChatRequest(BaseModel):
session_id: str
message: str
class ChatResponse(BaseModel):
content: str
agent: str
runner: Runner | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global runner
store = SQLiteSessionStore("production_sessions.db", default_max_turns=40)
agent = Agent(
name="ProductionBot",
model="gpt-4o",
instructions="Provide concise, accurate enterprise responses.",
)
runner = Runner(starting_agent=agent, session_store=store)
yield
app = FastAPI(title="ChatFlow Enterprise API", lifespan=lifespan)
@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(payload: ChatRequest):
if runner is None:
raise HTTPException(status_code=503, detail="Runner service initializing.")
try:
result = await runner.run_async(
session_id=payload.session_id,
user_message=payload.message,
)
return ChatResponse(
content=result.content,
agent=result.active_agent_name,
)
except TimeoutError:
raise HTTPException(status_code=504, detail="Inference turn timed out.")
except Exception as err:
raise HTTPException(status_code=500, detail="Internal agent execution failure.")
@app.get("/healthz")
async def health_check():
"""Health check endpoint for container orchestrators."""
return {"status": "ok", "service": "chatflow-agent"}Considerações de Concorrência no SQLite
O SQLiteSessionStore fornece persistência sem infraestrutura adicional e sem dependências de serviços externos. Ao utilizar o SQLite em produção, configure os seguintes parâmetros:
- Modo WAL (Write-Ahead Logging): O modo de journal padrão do SQLite bloqueia o arquivo completo durante gravações. O modo WAL permite leituras concorrentes enquanto uma gravação está ocorrendo.
- Descarga para Threads de Trabalho: No
v0.4.0rc1, ochatflow-agentdelega operações de banco de dados para threads secundárias via_run_blocking(), garantindo que a E/S em disco não bloqueie o loop de eventos asyncio. - Isolamento de Processos: O SQLite não suporta gravações concorrentes de alta frequência com segurança entre múltiplos processos independentes do sistema operacional (ex.:
gunicorn -w 8). Execute um único processo worker com alta concorrência assíncrona (uvicorn --workers 1) ou segmente as sessões por identificador de usuário.
Conteinerização com Docker
Um Dockerfile multi-stage otimiza o tamanho final da imagem e reforça a segurança executando com um usuário sem privilégios de root:
# Multi-stage Dockerfile para aplicações chatflow-agent
FROM python:3.11-slim AS builder
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt
# Imagem Final de Execução
FROM python:3.11-slim AS runner
WORKDIR /app
# Executar como usuário não-root
RUN addgroup --system appgroup && adduser --system --group appuser
COPY --from=builder /root/.local /home/appuser/.local
COPY --chown=appuser:appgroup . .
ENV PATH=/home/appuser/.local/bin:$PATH
ENV PYTHONUNBUFFERED=1
USER appuser
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]Configuração de Reverse Proxy (Nginx)
Respostas de modelos de linguagem e ciclos de ferramentas multiagente podem levar vários segundos para serem concluídos. Assegure-se de que o proxy reverso não encerre as conexões prematuramente:
server {
listen 80;
server_name agent.example.com;
location / {
proxy_pass http://127.0.0.1:8000;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection 'upgrade';
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# Timeouts generosos para turnos de inferência multiagente
proxy_connect_timeout 120s;
proxy_send_timeout 120s;
proxy_read_timeout 120s;
}
}Lista de Verificação de Prontidão para Produção
| Categoria | Item de Verificação | Configuração Recomendada |
|---|---|---|
| Redes | Timeout de Leitura do Proxy | timeout >= 120s |
| Resiliência | Timeout de Turno | Configurar turn_timeout=45.0 para evitar turnos travados |
| Persistência | Modo de Journal do SQLite | PRAGMA journal_mode=WAL; |
| Segurança | Gestão de Segredos | Injetados via variáveis de ambiente; nunca no código-fonte |
| Monitoramento | Verificações de Integridade | Endpoint /healthz respondendo com HTTP 200 |
| Modelo de Processo | Concorrência | 1 processo worker por contêiner com concorrência asyncio |