Construire un agent enrichi par les connaissances
De bout en bout : liaison d’entités → requête sur le graphe → synthèse de la réponse.
Construire un agent enrichi par les connaissances est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.
Qu’est-ce qu’un agent augmenté par les connaissances ?
Un agent augmenté par les connaissances enrichit ses réponses à l’aide d’une base de connaissances. Lorsqu’une question arrive, l’agent extrait les entités, les recherche dans un graphe de connaissances, trouve les documents pertinents grâce à la recherche vectorielle et fournit tout le contexte au LLM pour obtenir une réponse riche et étayée.
Le pipeline complet de recherche
Le pipeline de l’agent : 1 recevoir la question → 2 extraire les entités → 3 rechercher dans le graphe le contexte des entités → 4 rechercher les documents pertinents par recherche vectorielle → 5 combiner tout le contexte → 6 le LLM génère la réponse.
from dataclasses import dataclass, field
from typing import List, Dict, Any
@dataclass
class RetrievalContext:
question: str
entities: List[str] = field(default_factory=list)
graph_context: Dict[str, Any] = field(default_factory=dict)
vector_documents: List[Dict] = field(default_factory=list)
combined_context: str = ''
answer: str = ''
sources_used: List[str] = field(default_factory=list)
# The agent will populate this object as it works through the pipeline
ctx = RetrievalContext(question='What AI projects is Sam Altman known for?')
print('RetrievalContext created:', ctx.question)Étape 1 : extraction des entités
Extrayez les entités nommées de la question. Ces entités servent d’ancres pour la recherche dans le graphe. Utilisez spaCy pour la rapidité, et un LLM pour les cas complexes ou les entités propres au domaine.
import spacy
nlp = spacy.load('en_core_web_sm')
def extract_question_entities(question: str) -> List[str]:
doc = nlp(question)
entities = list({
ent.text for ent in doc.ents
if ent.label_ in ['PERSON', 'ORG', 'GPE', 'PRODUCT', 'WORK_OF_ART']
})
return entities
def extract_entities_with_llm_fallback(question: str, client) -> List[str]:
spacy_entities = extract_question_entities(question)
if spacy_entities:
return spacy_entities
# Fallback to LLM for questions where spaCy finds nothing
import json
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{
'role': 'user',
'content': f'Extract named entities (people, companies, technologies) from: "{question}". Return JSON: {{"entities": ["name1", "name2"]}}'
}],
response_format={'type': 'json_object'}
)
result = json.loads(response.choices[0].message.content)
return result.get('entities', [])
question = 'What AI projects is Sam Altman known for?'
entities = extract_question_entities(question)
print('Extracted entities:', entities)Étape 2 : recherche dans le graphe
Pour chaque entité extraite, interrogez le graphe de connaissances afin d’obtenir ses propriétés et ses relations. Le LLM dispose ainsi de faits contextuels qu’il ne peut pas inventer.
from neo4j import GraphDatabase
driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password'))
def get_rich_entity_context(entity_name: str) -> dict:
with driver.session() as session:
# Get entity + its relationships
result = session.run(
'MATCH (n {name: $name}) '
'OPTIONAL MATCH (n)-[r]->(target) '
'RETURN n, labels(n) AS labels, '
'COLLECT({rel: type(r), target_name: target.name, target_label: labels(target)}) AS outgoing '
'LIMIT 1',
name=entity_name
)
record = result.single()
if not record:
return {'found': False, 'name': entity_name}
return {
'found': True,
'name': entity_name,
'labels': record['labels'],
'properties': dict(record['n']),
'connections': [
c for c in record['outgoing'] if c.get('target_name')
][:10]
}
def format_entity_context_for_llm(entity_ctx: dict) -> str:
if not entity_ctx.get('found'):
return f'No knowledge graph data found for "{entity_ctx["name"]}"'
props = entity_ctx.get('properties', {})
connections = entity_ctx.get('connections', [])
conn_strs = [f"{c['rel']} -> {c['target_name']}" for c in connections[:5]]
return (
f"Entity: {entity_ctx['name']} ({', '.join(entity_ctx['labels'])})\n"
f"Properties: {props}\n"
f"Relationships: {'; '.join(conn_strs)}"
)Étape 3 : recherche vectorielle
Effectuez une recherche vectorielle avec la question d’origine afin de trouver les documents les plus pertinents sur le plan sémantique dans votre base de connaissances. Ces documents fournissent des éléments pour étayer la réponse.
import chromadb
import openai
client = openai.OpenAI(api_key='sk-...')
chroma_client = chromadb.Client()
collection = chroma_client.get_or_create_collection('knowledge_base')
def vector_search(query: str, top_k: int = 5) -> list:
response = client.embeddings.create(
model='text-embedding-3-small',
input=query
)
query_embedding = response.data[0].embedding
results = collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
include=['documents', 'metadatas', 'distances']
)
documents = []
for i in range(len(results['ids'][0])):
documents.append({
'text': results['documents'][0][i],
'metadata': results['metadatas'][0][i],
'distance': results['distances'][0][i],
'relevance': 1 - results['distances'][0][i] # Convert distance to similarity
})
return documents
print('Vector search function defined')Étape 4 : combiner le contexte
Assemblez le contexte du graphe et les documents vectoriels en une seule chaîne de contexte bien structurée. L’ordre est important : d’abord les faits du graphe (haute précision), puis les documents vectoriels (large couverture).
def combine_context(question: str, graph_contexts: dict, vector_docs: list) -> str:
sections = []
# Graph facts section
if graph_contexts:
graph_parts = ['### Knowledge Graph Facts']
for entity_name, ctx in graph_contexts.items():
graph_parts.append(format_entity_context_for_llm(ctx))
sections.append('\n'.join(graph_parts))
# Vector documents section
if vector_docs:
doc_parts = ['### Relevant Documents']
for i, doc in enumerate(vector_docs[:4]):
title = doc.get('metadata', {}).get('title', f'Document {i+1}')
text = doc['text'][:800] # Limit per document
relevance = doc.get('relevance', 0)
doc_parts.append(f'**{title}** (relevance: {relevance:.2f})\n{text}')
sections.append('\n'.join(doc_parts))
context = '\n\n'.join(sections)
# Total context budget: ~8000 tokens ~ 32000 chars
if len(context) > 32000:
context = context[:32000]
return contextÉtape 5 : génération de la réponse par le LLM
Transmettez le contexte combiné au LLM sous forme de message système ou de contexte utilisateur. Demandez-lui d’utiliser les informations fournies et de citer la source de chaque fait.
import openai
client = openai.OpenAI(api_key='sk-...')
def generate_answer(question: str, combined_context: str) -> str:
system_prompt = (
'You are a knowledgeable assistant. Answer the question using ONLY the provided context. '
'Cite your sources by mentioning whether a fact came from the knowledge graph or a specific document. '
'If the context does not contain enough information, say so clearly.'
)
user_message = (
f'Context:\n{combined_context}\n\n'
f'Question: {question}\n\n'
'Please answer based on the context above.'
)
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[
{'role': 'system', 'content': system_prompt},
{'role': 'user', 'content': user_message}
],
temperature=0.1 # Low temperature for factual answers
)
return response.choices[0].message.contentOrchestrateur complet de l’agent
La fonction d’orchestration relie toutes les étapes. Elle reçoit une question, exécute le pipeline complet et renvoie un résultat structuré contenant la réponse et le contexte utilisé.
async def knowledge_augmented_agent(question: str) -> RetrievalContext:
ctx = RetrievalContext(question=question)
# Step 1: Extract entities
ctx.entities = extract_question_entities(question)
print(f'Entities: {ctx.entities}')
# Steps 2 & 3: Graph + Vector in parallel
import asyncio
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)
loop = asyncio.get_event_loop()
async def graph_step():
contexts = {}
for entity in ctx.entities:
context = await loop.run_in_executor(executor, get_rich_entity_context, entity)
if context.get('found'):
contexts[entity] = context
return contexts
async def vector_step():
return await loop.run_in_executor(executor, vector_search, question, 5)
ctx.graph_context, ctx.vector_documents = await asyncio.gather(
graph_step(), vector_step()
)
# Step 4: Combine
ctx.combined_context = combine_context(
question, ctx.graph_context, ctx.vector_documents
)
# Step 5: Generate answer
ctx.answer = generate_answer(question, ctx.combined_context)
return ctxGérer l’absence de résultats
Lorsque la base de connaissances ne contient aucune information pertinente, l’agent doit le dire clairement plutôt que d’halluciner. Vérifiez que la recherche a renvoyé des résultats utiles avant d’appeler le LLM.
def check_retrieval_quality(graph_contexts: dict, vector_docs: list, threshold: float = 0.7) -> dict:
has_graph = len(graph_contexts) > 0
# Filter vector docs below relevance threshold
high_quality_docs = [d for d in vector_docs if d.get('relevance', 0) >= threshold]
return {
'has_graph_context': has_graph,
'graph_entity_count': len(graph_contexts),
'vector_doc_count': len(high_quality_docs),
'retrieval_quality': 'high' if (has_graph or len(high_quality_docs) >= 2) else 'low',
'usable_docs': high_quality_docs
}
def answer_with_fallback(question: str, graph_contexts: dict, vector_docs: list) -> str:
quality = check_retrieval_quality(graph_contexts, vector_docs)
if quality['retrieval_quality'] == 'low':
return (
f'I don\'t have enough information in my knowledge base to answer '
f'"{question}" confidently. '
'Please ensure relevant documents are indexed or the knowledge graph '
'contains the required entities.'
)
context = combine_context(question, graph_contexts, quality['usable_docs'])
return generate_answer(question, context)Mettre en cache les résultats de recherche
Mettez en cache les recherches d’entités et les résultats de recherche vectorielle afin d’éviter les appels d’API coûteux répétés pour des questions similaires. Utilisez un TTL afin que les données obsolètes soient actualisées régulièrement.
import hashlib
import json
from datetime import datetime, timedelta
class RetrievalCache:
def __init__(self, ttl_minutes: int = 60):
self.cache = {}
self.ttl = timedelta(minutes=ttl_minutes)
def _key(self, namespace: str, value: str) -> str:
return hashlib.md5(f'{namespace}:{value}'.encode()).hexdigest()
def get(self, namespace: str, value: str):
key = self._key(namespace, value)
entry = self.cache.get(key)
if entry and datetime.now() - entry['ts'] < self.ttl:
return entry['data']
return None
def set(self, namespace: str, value: str, data):
key = self._key(namespace, value)
self.cache[key] = {'data': data, 'ts': datetime.now()}
cache = RetrievalCache(ttl_minutes=30)
def cached_graph_lookup(entity: str) -> dict:
cached = cache.get('graph', entity)
if cached:
print(f'Cache hit for entity: {entity}')
return cached
result = get_rich_entity_context(entity)
cache.set('graph', entity, result)
return result
print('Retrieval cache initialized')Journalisation et observabilité
Journalisez chaque étape de la recherche afin de pouvoir déterminer pourquoi une réponse était bonne ou mauvaise. Enregistrez les entités trouvées, le nombre de documents récupérés, leurs scores de pertinence et la réponse finale.
import logging
import json
from datetime import datetime
logger = logging.getLogger('ka_agent')
def log_agent_run(ctx: 'RetrievalContext', duration_ms: float):
logger.info(json.dumps({
'timestamp': datetime.utcnow().isoformat(),
'question': ctx.question,
'entities_found': ctx.entities,
'graph_entities_resolved': list(ctx.graph_context.keys()),
'vector_docs_retrieved': len(ctx.vector_documents),
'vector_doc_relevances': [
round(d.get('relevance', 0), 3)
for d in ctx.vector_documents
],
'context_length_chars': len(ctx.combined_context),
'answer_length_chars': len(ctx.answer),
'duration_ms': round(duration_ms, 1)
}))
print('Observability logging configured')Vérification des connaissances : agent enrichi par des connaissances
Testez votre compréhension de la création d’agents enrichis par des connaissances.
Résumé de l’agent enrichi par des connaissances
Un agent enrichi par des connaissances combine l’extraction d’entités, le parcours de graphes et la recherche vectorielle au sein d’une chaîne de traitement qui fournit au LLM un contexte riche et ancré dans des données fiables. Le résultat consiste en des réponses plus précises et moins sujettes aux hallucinations, étayées par les données réelles de votre base de connaissances. Principaux ajouts : mise en cache, gestion de repli lorsque la recherche ne renvoie aucun résultat et journalisation structurée pour l’observabilité.
Questions Fréquemment Posées
La leçon « Construire un agent enrichi par les connaissances » est-elle gratuite ?
Oui — le texte complet de « Construire un agent enrichi par les connaissances » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Construire un agent enrichi par les connaissances » ?
De bout en bout : liaison d’entités → requête sur le graphe → synthèse de la réponse. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer AI Agents ?
Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.
Combien de temps prend la leçon « Construire un agent enrichi par les connaissances » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?
Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Extraction d’entités pour les graphes de connaissances
- Requêtes Neo4j depuis les outils d’un agent
- Combiner la recherche vectorielle et la recherche dans les graphes
- Construire un agent enrichi par les connaissances