RAG Avancé et Agentic RAG avec Agno

from agno.agent import Agent
from agno.knowledge.embedder.cohere import CohereEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.knowledge.reranker.cohere import CohereReranker
from agno.models.openai import OpenAIResponses
from agno.vectordb.lancedb import LanceDb, SearchType

knowledge = Knowledge(
    vector_db=LanceDb(
        uri="tmp/lancedb",
        table_name="docs_reranked",
        search_type=SearchType.hybrid,
        embedder=CohereEmbedder(id="embed-v4.0"),
        reranker=CohereReranker(model="rerank-v3.5"),
    ),
)

# Charger des documents
knowledge.insert(url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf")

agent = Agent(
    model=OpenAIResponses(id="gpt-4o"),
    knowledge=knowledge,
    instructions="Utilise ta base de connaissances. Cite tes sources.",
    markdown=True,
)

agent.print_response("Quelle est la différence entre Pad Thai et Pad See Ew ?", stream=True)

from agno.agent import Agent
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.models.openai import OpenAIResponses
from agno.tools.knowledge import KnowledgeTools
from agno.vectordb.lancedb import LanceDb, SearchType

# Knowledge Base
knowledge = Knowledge(
    vector_db=LanceDb(
        uri="tmp/lancedb",
        table_name="docs_agentic",
        search_type=SearchType.hybrid,
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)

knowledge.insert(path="./documentation/")

# KnowledgeTools : l''agent décide quand chercher
knowledge_tools = KnowledgeTools(
    knowledge=knowledge,
    think=True,        # L''agent réfléchit avant de chercher
    search=True,       # Peut faire des recherches
    analyze=True,      # Évalue la pertinence des résultats
    add_few_shot=True, # Exemples d''utilisation dans le prompt
)

agent = Agent(
    model=OpenAIResponses(id="gpt-4o"),
    tools=[knowledge_tools],
    instructions=[
        "Tu es un assistant technique.",
        "Utilise tes outils de recherche quand l''utilisateur pose une question factuelle.",
        "Pour les salutations ou questions triviales, réponds directement.",
    ],
    markdown=True,
)

# L''agent ne cherche PAS pour ça :
agent.print_response("Bonjour !", stream=True)

# L''agent CHERCHE pour ça :
agent.print_response("Comment configurer le module d''authentification ?", stream=True)

import asyncio
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.models.openai import OpenAIResponses
from agno.vectordb.lancedb import LanceDb, SearchType

# Knowledge Base
knowledge = Knowledge(
    vector_db=LanceDb(
        uri="tmp/lancedb",
        table_name="production_docs",
        search_type=SearchType.hybrid,
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)

# Charger les documents de manière asynchrone
asyncio.run(knowledge.ainsert(
    name="Documentation",
    url="https://docs.agno.com/llms-full.txt"
))

# Agent avec mémoire persistante
agent = Agent(
    name="Assistant Production",
    model=OpenAIResponses(id="gpt-4o"),
    knowledge=knowledge,
    instructions=[
        "Cherche dans ta base de connaissances avant de répondre.",
        "Si tu ne trouves pas l''information, dis-le clairement.",
        "Réponds en français.",
    ],
    # Mémoire persistante via SQLite
    db=SqliteDb(
        session_table="sessions",
        db_file="tmp/agents.db",
    ),
    add_history_to_context=True,  # Inclut l''historique dans le contexte
    markdown=True,
)

# L''agent se souvient des conversations précédentes
agent.print_response("Je travaille sur le module d''auth", stream=True)
# ... plus tard ...
agent.print_response("Quels sont les paramètres de configuration ?", stream=True)
# → L''agent sait qu''on parle du module d''auth

from agno.vectordb.pgvector import PgVector, SearchType

knowledge = Knowledge(
    vector_db=PgVector(
        db_url="postgresql://user:pass@localhost:5432/mydb",
        table_name="documents",
        search_type=SearchType.hybrid,
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)

# Installation pour PgVector
pip install agno openai pgvector psycopg pypdf

# Créer un dataset de test
test_cases = [
    {
        "question": "Comment configurer l''authentification ?",
        "expected_docs": ["auth-config.md", "security-guide.md"],
        "expected_answer_contains": ["JWT", "token", "middleware"],
    },
    {
        "question": "Quels sont les prérequis système ?",
        "expected_docs": ["requirements.md"],
        "expected_answer_contains": ["Python 3.8", "PostgreSQL"],
    },
]

# Tester chaque cas
for case in test_cases:
    response = agent.run(case["question"])
    answer = response.content

    # Vérifier que les mots-clés attendus sont présents
    found = [kw for kw in case["expected_answer_contains"] if kw.lower() in answer.lower()]
    missing = [kw for kw in case["expected_answer_contains"] if kw.lower() not in answer.lower()]

    print(f"Question : {case['question']}")
    print(f"Mots-clés trouvés : {found}")
    print(f"Mots-clés manquants : {missing}")
    print(f"Score : {len(found)}/{len(case['expected_answer_contains'])}")
    print("---")