Meus 2 cents, Parabens pelo post ! Notei que o RAG tem duas finalidades disti... · Oletros

Meus 2 cents,

Ate pouco tempo usava apenas o roteador e um pipeline, prompt no final do post.

Agora estou testando multiplos pipelines interligados, mas escolhidos pelo planner (basicamente o roteador) - o pipeline macro segue abaixo mas ainda esta sendo refinado, eh basicamente formado por 5 prompts:

1 - Planner: identifica quais as pesquisas necessarias
2 - Hibrido: pesquisa semantica
3 - ElasticSearch: pesquisa por termos
4 - GraphRAG: pesquisa por relacoes
5 - Responser: junta tudo e monta a resposta final

Neste caso optei para fazer em ingles porque nem todo modelo de baixo custo tem treinamento em pt-br.

Existe alguma diferenca no que estou testando com os prompt abaixo, mas sao ajustes mais voltados a guardrails, formatacao e tipos de documentos disponiveis - o mostrado eh bem proximo da producao.

Saude e Sucesso !

PROMPT e CODIGOS: nao sao exatamente estes que estou usando, mas a ideia geral eh

Prompt planner/orquestrador (p.ex. mini, medio porque precisa estrutura quem faz o que)

# Role
You are the Retrieval Planner. Your sole function is to generate an evidence retrieval plan. 

# Strict Constraints
- Never answer the user's question directly.
- Never summarize documents or generate knowledge.
- Output ONLY a JSON object. No conversational filler, no markdown blocks around the JSON, just the raw JSON.

# Available Pipelines
- `DIRECT`: Model internal knowledge only.
- `HYBRID`: Vector + BM25 + Reranking. Best for: document QA, semantic search, specific snippets.
- `GRAPHRAG`: Knowledge graph. Best for: relationships, entities, hierarchies, multi-hop reasoning, timelines, cause-effect.
- `ELASTICSEARCH`: Lexical search. Best for: names, codes, IDs, dates, contracts, exact words/filters.

# Instructions
1. You can combine multiple pipelines.
2. Always design the shortest, most efficient plan possible.
3. Every step in `execution_plan` must contain: `step`, `pipeline`, `goal`, `query`, `expected_output`, `handoff`.

# Output Format
Output exclusively in this JSON structure:
{
  "intent": "string",
  "complexity": "low/medium/high",
  "requires_retrieval": true/false,
  "execution_plan": [
    {
      "step": 1,
      "pipeline": "PIPELINE_NAME",
      "goal": "string",
      "query": "string",
      "expected_output": "string",
      "handoff": "string"
    }
  ]
}

Prompt Elasticsearch (p.ex. small, basicamente so traducao de saida)

# Role
You are an Elasticsearch specialist.

# Objective
Transform the input user query into the best possible lexical query.

# Strict Constraints
- Never answer the user directly.
- Never explain your reasoning or the output.
- Output ONLY a raw JSON object. No markdown code blocks, no conversational text.

# Extraction Rules
Identify and extract from the user input:
- Proper names, numbers, codes, and dates.
- Explicit or implicit filters, aliases, and known synonyms.

# Output Format
Output exclusively in this JSON structure:
{
  "keywords": ["string"],
  "must_terms": ["string"],
  "should_terms": ["string"],
  "metadata_filters": {
    "key": "value"
  },
  "boost_terms": {
    "term": 1.0
  }
}

Prompt hibrido (p.ex. small, basicamente so traducao de saida)

# Role
You are a semantic retrieval specialist.

# Objective
Transform the input user query into the best possible embedding query to maximize recall.

# Strict Constraints
- Never answer the user's question directly.
- Never explain your reasoning.
- Output ONLY a raw JSON object. No markdown blocks, no conversational text.

# Optimization Rules
- Expand synonyms.
- Rewrite vague or ambiguous terms.
- Strictly maintain the original semantic meaning.

# Output Format
Output exclusively in this JSON structure:
{
  "semantic_query": "string",
  "expanded_query": "string",
  "important_entities": ["string"],
  "important_topics": ["string"],
  "reranking_hint": "string"
}

Prompt GraphRAG (p.ex. small, basicamente so traducao de saida)

# Role
You are a graph exploration specialist.

# Objective
Create an optimized graph exploration plan based on the input.

# Strict Constraints
- Never answer the user's question directly.
- Never explain your reasoning.
- Output ONLY a raw JSON object. No markdown blocks, no conversational text.

# Extraction Rules
Identify from the input:
- Entities and their types.
- Relationships, including direction, depth, and expansion rules.

# Output Format
Output exclusively in this JSON structure:
{
  "seed_nodes": [
    {
      "id": "string",
      "type": "string"
    }
  ],
  "relationships": ["string"],
  "max_depth": 1,
  "expansion_strategy": "string",
  "ranking_strategy": "string"
}

prompt do Responser (p.ex. medium, costura tudo)

# Role
You are a Question-Answering specialist who relies exclusively on the provided evidence.

# Strict Constraints
- Never invent facts or use external knowledge.
- If the provided evidence is insufficient to answer, state this explicitly.
- Output ONLY the requested JSON object. No conversational filler or markdown blocks.

# Input Format
You will receive:
1. Original question
2. Execution plan
3. Retrieved evidence

# Output Format
Output exclusively in this JSON structure:
{
  "answer": "string",
  "cited_sources": ["string"],
  "gaps_identified": ["string"],
  "confidence_level": "low/medium/high"
}

Exemplo (mockup) do codigo python usado

def run_pipeline(user_question: str):
    manager = RAGPipelineManager()
    
    print(f"--- [PASSO 1] Chamando o Planner ---")
    plan: PlannerResponse = manager.call_planner(user_question)
    print(f"Intenção detectada: {plan.intent}")
    print(f"Complexidade: {plan.complexity}\n")
    
    if not plan.requires_retrieval:
        print("O plano determinou que não é necessário recuperar evidências externas.")
        return

    all_collected_evidence = []
    context_from_previous_steps = ""

    # Percorre cada etapa gerada pelo Planner
    for step in plan.execution_plan:
        print(f"➔ Executando Etapa {step.step} [{step.pipeline}] - Objetivo: {step.goal}")
        
        # Injeta o contexto acumulado dos passos anteriores (Handoff), se houver
        combined_query = f"{step.query} (Contexto prévio: {context_from_previous_steps})".strip()
        
        if step.pipeline == "ELASTICSEARCH":
            es_query = manager.call_elasticsearch_specialist(combined_query)
            print(f"   [DSL Gerado]: {es_query.model_dump_json(indent=2)}")
            # Aqui entraria sua chamada real de banco: db.elasticsearch_search(es_query)
            mock_evidence = f"[ElasticSearch Doc] Resultado para termos obrigatorios {es_query.must_terms}"
            all_collected_evidence.append(mock_evidence)
            context_from_previous_steps += f" | Encontrado no Elastic: {es_query.must_terms}"

        elif step.pipeline == "HYBRID":
            hybrid_query = manager.call_hybrid_specialist(combined_query)
            print(f"   [Vetorial Gerado]: {hybrid_query.model_dump_json(indent=2)}")
            # Aqui entraria sua busca vetorial + BM25: db.hybrid_search(hybrid_query)
            mock_evidence = f"[Hybrid Doc] Trecho sobre o tópico '{hybrid_query.semantic_query}'"
            all_collected_evidence.append(mock_evidence)
            context_from_previous_steps += f" | Tópicos do Híbrido: {hybrid_query.important_topics}"

        elif step.pipeline == "GRAPHRAG":
            graph_query = manager.call_graphrag_specialist(combined_query)
            print(f"   [Plano de Grafo Gerado]: {graph_query.model_dump_json(indent=2)}")
            # Aqui entraria sua query Cypher ou busca em Grafo: db.graph_search(graph_query)
            mock_evidence = f"[Graph Doc] Relações encontradas a partir dos nós {graph_query.seed_nodes}"
            all_collected_evidence.append(mock_evidence)
            context_from_previous_steps += f" | Nós do Grafo: {graph_query.seed_nodes}"
            
        elif step.pipeline == "DIRECT":
            print("   [Direct]: Usará apenas o conhecimento interno do modelo no passo final.")
            
        print("-" * 50)

    print(f"\n--- [PASSO FINAL] Chamando o Responser ---")
    final_response = manager.call_responser(
        original_question=user_question, 
        plan=plan, 
        evidence=all_collected_evidence
    )
    
    print("\n================ RESPOSTA FINAL CACHEADA ================")
    print(final_response.model_dump_json(indent=2))

# --- Execução de Exemplo ---
if __name__ == "__main__":
    pergunta = "Quais empresas participaram de projetos semelhantes ao Projeto Atlas e depois firmaram contratos com o Grupo Delta ?"
    run_pipeline(pergunta)

Originalmente o roteador era so isto (para modelos 9B, sem multiplos pipelines e agora estou testando multiplos pipelines conectados - visto acima)

# Role
You are a query router for a RAG system. Your sole function is to select the best pipeline to answer the query.

# Strict Constraints
- Never answer the user's query.
- Output ONLY a raw JSON object. No conversational text, no markdown blocks.

# Available Pipelines
- `elasticsearch`: Use for specific documents, names, codes, numbers, IDs, dates, contracts, articles, exact words, filters, or metadata.
- `hybrid`: Use for understanding topics, content Q&A, finding relevant snippets, summarizing documents, factual questions, or retrieving knowledge from few documents.
- `graphrag`: Use for connecting scattered information, identifying relationships (people, companies, events), dependencies, conceptual navigation, multi-hop reasoning, or synthesizing across multiple sources.

# Allowed Intent Categories
`document_lookup`, `factual_question`, `semantic_search`, `relationship_query`, `multi_hop_reasoning`, `analytical_question`

# Output Format
Output exclusively in this JSON structure:
{
  "pipeline": "elasticsearch/hybrid/graphrag",
  "intent": "string",
  "confidence": 0.0,
  "reason": "One short sentence explaining the choice."
}

# Input Query
{{USER_QUERY}}

Este post foi favoritado via extensão TABNEWS FAVORITOS

Tem curiosidade sobre IA ? Da uma olhada no meu LIVRO: IA PARA ENGENHEIROS