0Pricing
AI Agents · 课时

构建知识增强型智能体

端到端流程：实体链接 → 图谱查询 → 答案合成

构建知识增强型智能体 是 CoddyKit 上的免费 AI Agents 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

什么是知识增强型智能体

知识增强型智能体会利用知识库丰富其答案。当收到问题时，智能体会提取实体，在知识图谱中查询这些实体，通过向量搜索查找相关文档，并将所有上下文提供给 LLM，以生成丰富且有依据的答案。

完整的检索流水线

智能体流水线：1 接收问题 → 2 提取实体 → 3 查询图谱以获取实体上下文 → 4 通过向量搜索查找相关文档 → 5 合并所有上下文 → 6 由 LLM 生成答案。

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)

步骤 1：实体提取

从问题中提取命名实体。这些实体会成为图谱查询的锚点。使用 spaCy 提高速度；对于棘手的情况或领域特定实体，则使用 LLM。

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)

步骤 2：图谱查询

针对每个提取出的实体查询知识图谱，以获取其属性和关系。这些信息会为 LLM 提供它无法凭空生成的背景事实。

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)}"
    )

步骤 3：向量搜索

使用原始问题运行向量搜索，以查找知识库中语义相关性最高的文档。这些文档会为答案提供支持证据。

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')

步骤 4：合并上下文

将图谱上下文和向量文档汇总到一个结构清晰的上下文字符串中。顺序很重要：先放图谱事实（高精度），再放向量文档（覆盖范围广）。

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

步骤 5：生成 LLM 答案

将合并后的上下文作为系统消息或用户上下文传递给 LLM。要求它使用所提供的信息，并注明每条事实来自哪个来源。

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.content

完整的智能体编排器

编排器函数会将所有步骤串联起来。它接收一个问题，运行完整流水线，并返回包含答案及所用上下文的结构化结果。

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 ctx

处理空检索结果

当知识库中没有相关信息时，智能体应明确说明这一点，而不是产生幻觉。调用 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)

缓存检索结果

缓存实体查询结果和向量搜索结果，以避免针对相似问题重复执行成本高昂的 API 调用。请使用 TTL 定期刷新过时数据。

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')

日志记录与可观测性

记录每个检索步骤，以便诊断答案为何正确或不正确。记录找到的实体、检索到的文档数量、文档的相关性分数，以及最终答案。

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')

知识检查：知识增强代理

请检验您对构建知识增强代理的理解。

知识增强代理总结

知识增强代理将实体提取、图遍历和向量搜索组合成一个流水线，为 LLM 提供丰富且有事实依据的上下文。这样可以生成更准确、更少产生幻觉的回答，并以知识库中的真实数据作为支持。主要新增功能包括：缓存、对空检索结果的回退处理，以及用于可观测性的结构化日志记录。

常见问题解答

「构建知识增强型智能体」课时是免费的吗？

是的 — 「构建知识增强型智能体」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「构建知识增强型智能体」这节课中我会学到什么？

端到端流程：实体链接 → 图谱查询 → 答案合成 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。

「构建知识增强型智能体」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 知识图谱的实体提取
  2. 通过智能体工具查询 Neo4j
  3. 结合向量检索与图谱检索
  4. 构建知识增强型智能体
← 返回 AI Agents