Orquestração de Pipelines de IA

Construa fluxos de trabalho de IA confiáveis e observáveis — desde a ingestão de documentos até a inferência em lote e pipelines de treinamento de modelos.

Quando Usar Esta Habilidade

Use esta habilidade quando:

• Agendar ingestão e reindexação recorrentes de documentos RAG
• Orquestrar fluxos de trabalho de processamento LLM em lote com várias etapas
• Executar trabalhos noturnos de avaliação e ajuste fino de modelos
• Construir pipelines ETL que alimentam modelos de IA
• Gerenciar dependências entre preparação de dados e serviço de modelos

Contrato de Execução

Estes são fragmentos de design de aplicação: adaptadores, conjuntos de dados, arquivos de modelo, segredos, esquemas de vetor e manipuladores de implantação não são empacotados. Fixe as versões suportadas de frameworks/provedores, selecione o alvo e as permissões reais, e autorize agendamentos, cobranças do provedor e gravações separadamente. Use primeiro "fixtures" sintéticas; preserve dados privados, valide fontes buscadas e persista checkpoints versionados para evitar repetir efeitos colaterais.

Seleção de Ferramentas

Prefect — Início Rápido

pip install prefect prefect-kubernetes

# Start Prefect server (or use Prefect Cloud)
prefect server start

# In another terminal
prefect worker start --pool default-agent-pool

Prefect: Esboço de Design de Ingestão RAG (não executável)

from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash
from datetime import timedelta
import hashlib

@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=24))
def fetch_documents(source_url: str) -> list[dict]:
    """Fetch documents from source; cached to avoid re-fetching."""
    logger = get_run_logger()
    logger.info(f"Fetching from {source_url}")
    # ... fetch logic
    return documents

@task(retries=3, retry_delay_seconds=30)
def chunk_and_embed(documents: list[dict]) -> list[dict]:
    """Chunk documents and generate embeddings with retry on failure."""
    from sentence_transformers import SentenceTransformer
    model = SentenceTransformer("BAAI/bge-large-en-v1.5")
    chunks = []
    for doc in documents:
        doc_chunks = chunk_text(doc["content"])
        embeddings = model.encode(doc_chunks, batch_size=64)
        for chunk, emb in zip(doc_chunks, embeddings):
            chunks.append({"text": chunk, "embedding": emb.tolist(),
                           "source": doc["url"], "doc_hash": doc["hash"]})
    return chunks

@task(retries=2)
def upsert_to_vector_store(chunks: list[dict]) -> int:
    """Upsert embeddings to Qdrant, skip unchanged documents."""
    from qdrant_client import QdrantClient
    client = QdrantClient("http://qdrant:6333")
    client.upsert(collection_name="knowledge-base", points=[...])
    return len(chunks)

@flow(name="rag-ingestion", log_prints=True)
def rag_ingestion_pipeline(sources: list[str]):
    """Full RAG ingestion flow — runs daily."""
    logger = get_run_logger()
    total = 0
    for source in sources:
        docs = fetch_documents(source)
        chunks = chunk_and_embed(docs)
        count = upsert_to_vector_store(chunks)
        total += count
        logger.info(f"Ingested {count} chunks from {source}")
    logger.info(f"Pipeline complete: {total} total chunks indexed")

if __name__ == "__main__":
    rag_ingestion_pipeline.serve(
        name="daily-rag-ingestion",
        cron="0 2 * * *",          # 2 AM daily
        parameters={"sources": ["https://docs.myapp.com", "https://api.myapp.com/kb"]},
    )

Prefect: Pipeline de Inferência LLM em Lote

from prefect import flow, task
from prefect.concurrency.asyncio import concurrency
import asyncio
from openai import AsyncOpenAI

@task(retries=0)  # Uma nova tentativa em lote pode repetir chamadas faturáveis bem-sucedidas
async def process_batch(items: list[dict], model: str) -> list[dict]:
    """Processa um lote de itens através de LLM com proteção de limite de taxa."""
    client = AsyncOpenAI(timeout=60.0, max_retries=0)
    async with concurrency("openai-api", occupy=len(items), timeout_seconds=30, strict=True):  # Slots concorrentes, não requisições por minuto
        tasks = [
            client.chat.completions.create(
                model=model,

…