企业官网建设流程全解析

Agent如果只依赖LLM的内置知识，就像一个只靠记忆力的学生——知识有限且会过时。RAG（Retrieval-Augmented Generation）让Agent能从外部知识库中检索专业、实时的信息，大幅提升回答的准确性和深度。本章将从全流程解析到高级策略，系统讲解RAG技术。

6.1 RAG全流程解析：数据清洗、分块与索引构建

RAG的核心流程

离线阶段：文档 -> 清洗 -> 分块 -> Embedding -> 向量数据库 在线阶段：问题 -> Embedding -> 检索 -> 拼入提示词 -> LLM生成

数据清洗

原始文档通常格式混乱、包含噪音，清洗质量直接影响检索效果：

import re class DocumentCleaner: """文档清洗工具""" def clean(self, text: str) -> str: text = self._remove_boilerplate(text) text = self._normalize_whitespace(text) text = self._remove_special_chars(text) return text.strip() def _remove_boilerplate(self, text: str) -> str: """移除页眉页脚、版权声明等模板文本""" patterns = [ r"版权所有.*?保留一切权利", r"本文档仅供参考.*?不构成任何建议", r"第\s*\d+\s*页\s*/\s*\d+", ] for pattern in patterns: text = re.sub(pattern, "", text, flags=re.IGNORECASE) return text def _normalize_whitespace(self, text: str) -> str: """规范化空白字符""" text = re.sub(r"\n{3,}", "\n\n", text) # 多个换行压缩为两个 text = re.sub(r"[ \t]+", " ", text) # 多个空格压缩为一个 return text def _remove_special_chars(self, text: str) -> str: """移除不可见字符和乱码""" text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text) return text

文本分块策略

分块是RAG中最关键的环节。块太大则检索精度低，块太小则上下文不完整。

策略一：固定长度分块

from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每块500字符 chunk_overlap=50, # 块间重叠50字符 separators=["\n\n", "\n", "。", "！", "？", "；", " "], )

策略二：语义分块

按语义边界（段落、章节）分块，而非机械切割：

from langchain.text_splitter import MarkdownHeaderTextSplitter # 按Markdown标题层级分块 md_splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "h1"), ("##", "h2"), ("###", "h3"), ] ) chunks = md_splitter.split_text(markdown_doc) # 每个chunk会自动携带其所属的标题层级信息

策略三：父-子分块（Small-to-Big Retrieval）

检索时用小块（精度高），返回时用大块（上下文完整）：

from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore # 子块分块器（检索用） child_splitter = RecursiveCharacterTextSplitter(chunk_size=200) # 父块分块器（返回用） parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000) vectorstore = Chroma(embedding_function=embeddings) docstore = InMemoryStore() # 存储父块原文 retriever = ParentDocumentRetriever( vectorstore=vectorstore, docstore=docstore, child_splitter=child_splitter, parent_splitter=parent_splitter, )

索引构建

from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 从文档构建索引 vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./knowledge_base", collection_metadata={"hnsw:space": "cosine"} # 使用余弦相似度 )

6.2 高级检索策略：混合搜索与重排序技术

基础向量检索的局限

纯向量检索存在一个根本性问题：语义相似不等于答案相关。比如用户问"Python如何安装"，向量检索可能返回一篇标题为"安装Python的各种方法"但内容过时的文章，而错过一篇标题不太相关但内容精确的教程。

混合搜索：向量 + 关键词

混合搜索将向量检索（语义匹配）和BM25关键词检索（精确匹配）结合，取长补短：

from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever # 向量检索器 vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10}) # BM25关键词检索器 bm25_retriever = BM25Retriever.from_documents(chunks, k=10) # 混合检索器 ensemble_retriever = EnsembleRetriever( retrievers=[vector_retriever, bm25_retriever], weights=[0.5, 0.5] # 向量和关键词各占50%权重 ) results = ensemble_retriever.invoke("Python安装教程")

重排序（Re-ranking）

检索回来的文档按相似度排序，但相似度最高的不一定是最相关的。重排序模型能更精确地评估文档与查询的相关性：

from langchain.retrievers import ContextualCompressionRetriever from langchain_cohere import CohereRerank # 使用Cohere重排序模型 compressor = CohereRerank(model="rerank-v3.5", top_n=5) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=ensemble_retriever ) # 先检索大量候选，再重排序筛选最相关的5条 results = compression_retriever.invoke("如何优化RAG的检索效果？")

查询改写与扩展

用户的原始查询可能不够精确。通过LLM改写或扩展查询，可以提升检索召回率：

class QueryRewriter: def __init__(self, llm): self.llm = llm def expand_query(self, original_query: str) -> list[str]: """将原始查询扩展为多个子查询""" prompt = f""" 原始查询：{original_query} 请生成3个不同角度的子查询，帮助检索到更全面的信息： 1. 同义词替换版本 2. 更具体的细化版本 3. 更宽泛的上位版本 """ response = self.llm.invoke(prompt) return [original_query] + self._parse_queries(response.content) def hyde_query(self, original_query: str) -> str: """HyDE：让LLM先假设性回答，用回答来检索""" prompt = f"请回答以下问题（即使不确定也给出你的最佳猜测）：{original_query}" hypothetical_answer = self.llm.invoke(prompt).content return hypothetical_answer # 用假设性答案作为检索查询

6.3 知识图谱与Agent的结合：结构化数据的利用

向量检索的盲区

向量检索擅长语义匹配，但不擅长处理实体关系和结构化查询。比如"和马斯克共同创立PayPal的人还创立了哪些公司？"——这类多跳关系查询，向量检索几乎无法处理。

知识图谱增强RAG

from langchain_community.graphs import Neo4jGraph from langchain.chains import GraphCypherQAChain # 连接知识图谱 graph = Neo4jGraph( url="bolt://localhost:7687", username="neo4j", password="your_password" ) # 用LLM生成Cypher查询 chain = GraphCypherQAChain.from_llm( llm=ChatOpenAI(model="gpt-4o", temperature=0), graph=graph, verbose=True, ) result = chain.run("和马斯克共同创立PayPal的人还创立了哪些公司？") # LLM自动生成：MATCH (p:Person)-[:COFOUNDED]->(c:Company {name: 'PayPal'})<-[:COFOUNDED]-(other:Person) # MATCH (other)-[:COFOUNDED]->(other_c:Company) # RETURN other.name, collect(other_c.name)

GraphRAG：微软的图谱增强方案

微软的GraphRAG方案流程：

1. 抽取：从文档中抽取实体和关系
2. 构建：构建社区图谱，发现实体集群
3. 索引：为每个社区生成摘要
4. 检索：先定位相关社区，再在社区内检索

# 简化版实体抽取 def extract_entities_and_relations(text: str, llm) -> dict: prompt = f"""从以下文本中抽取实体和关系，以JSON格式返回： 文本：{text} 格式：{{ "entities": [{{"name": "实体名", "type": "类型"}}], "relations": [{{"source": "实体1", "target": "实体2", "relation": "关系"}}] }}""" response = llm.invoke(prompt) return json.loads(response.content)

6.4 解决RAG痛点：丢失中间内容与多跳推理问题

痛点一：Lost in the Middle

研究表明，LLM对上下文中间位置的信息注意力最弱。当检索返回多个文档片段时，排在中间的片段容易被忽略。

解决方案：文档重排

def relevance_ordered_placement(documents: list[str], strategy: str = "descending") -> list[str]: """按相关性重新排列文档，避免重要信息被放在中间""" # 方案1：递减排列 - 最相关的在前 if strategy == "descending": return documents # 保持检索排序 # 方案2：交替排列 - 最相关和次相关的交替放置 if strategy == "alternating": result = [] left, right = 0, len(documents) - 1 while left <= right: result.append(documents[left]) if left != right: result.append(documents[right]) left += 1 right -= 1 return result return documents

痛点二：多跳推理

"谁是美国第46任总统的妻子的出生地？"——这需要两跳推理：第46任总统是谁 -> 他的妻子是谁 -> 她的出生地。

解决方案：迭代检索

class MultiHopRetriever: def __init__(self, retriever, llm): self.retriever = retriever self.llm = llm def retrieve(self, query: str, max_hops: int = 3) -> list[str]: all_docs = [] current_query = query for hop in range(max_hops): docs = self.retriever.invoke(current_query) all_docs.extend(docs) # 判断是否需要继续检索 judge_prompt = f""" 原始问题：{query} 已检索到的信息：{[d.page_content[:200] for d in all_docs]} 问题是否已经可以被完整回答？ 如果否，请生成下一步需要检索的子问题。 """ response = self.llm.invoke(judge_prompt).content if "是" in response and "完整" in response: break # 提取下一步检索的子问题 current_query = response.split("子问题：")[-1].strip() if "子问题：" in response else query return all_docs

痛点三：信息冲突

不同来源的信息可能互相矛盾：

def resolve_conflicts(documents: list[str], query: str, llm) -> str: """处理信息冲突""" conflict_prompt = f""" 问题：{query} 检索到的信息（可能有冲突）： {chr(10).join(f'来源{i+1}：{d}' for i, d in enumerate(documents))} 请分析： 1. 哪些信息之间存在冲突？ 2. 每个冲突的可能原因是什么（时效性、来源可靠性等）？ 3. 你倾向采纳哪个版本？为什么？ """ return llm.invoke(conflict_prompt).content

6.5 知识库动态更新：增量索引与版本管理策略

为什么需要动态更新？

知识库不是一成不变的。产品文档会更新、政策法规会调整、技术方案会演进。一个过时的知识库比没有知识库更危险——它会给出错误信息。

增量索引

class IncrementalIndexer: def __init__(self, vectorstore, embeddings): self.vectorstore = vectorstore self.embeddings = embeddings self.doc_hashes: dict[str, str] = {} # doc_id -> content_hash def _content_hash(self, content: str) -> str: import hashlib return hashlib.md5(content.encode()).hexdigest() def update(self, documents: list) -> dict: """增量更新索引""" added, updated, unchanged = 0, 0, 0 for doc in documents: doc_id = doc.metadata.get("doc_id", str(id(doc))) new_hash = self._content_hash(doc.page_content) if doc_id not in self.doc_hashes: # 新文档 self.vectorstore.add_documents([doc]) self.doc_hashes[doc_id] = new_hash added += 1 elif self.doc_hashes[doc_id] != new_hash: # 文档有变更，删除旧版本再添加新版本 self.vectorstore.delete(ids=[doc_id]) self.vectorstore.add_documents([doc]) self.doc_hashes[doc_id] = new_hash updated += 1 else: unchanged += 1 return {"added": added, "updated": updated, "unchanged": unchanged}

自动更新触发机制

import hashlib from datetime import datetime class AutoUpdater: """监控源文件变化，自动触发索引更新""" def __init__(self, indexer: IncrementalIndexer, source_dir: str): self.indexer = indexer self.source_dir = source_dir self.file_hashes: dict[str, str] = {} def scan_and_update(self) -> dict: """扫描源目录，检测变化并更新""" changes = {"added": [], "modified": [], "deleted": []} # 扫描当前文件 current_files = {} for root, dirs, files in os.walk(self.source_dir): for f in files: if f.endswith(('.md', '.txt', '.pdf', '.docx')): filepath = os.path.join(root, f) with open(filepath, 'rb') as file: file_hash = hashlib.md5(file.read()).hexdigest() current_files[filepath] = file_hash # 检测新增和修改 for filepath, file_hash in current_files.items(): if filepath not in self.file_hashes: changes["added"].append(filepath) elif self.file_hashes[filepath] != file_hash: changes["modified"].append(filepath) # 检测删除 for filepath in self.file_hashes: if filepath not in current_files: changes["deleted"].append(filepath) # 执行更新 # ... (将变更文件加载、分块、写入索引) self.file_hashes = current_files return changes

本章小结

学习资源推荐

如果你想更深入地学习大模型，以下是一些非常有价值的学习资源，这些资源将帮助你从不同角度学习大模型，提升你的实践能力。

一、全套AGI大模型学习路线

AI大模型时代的学习之旅：从基础到前沿，掌握人工智能的核心技能！​

因篇幅有限，仅展示部分资料，需要点击文章最下方名片即可前往获取

二、640套AI大模型报告合集

这套包含640份报告的合集，涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师，还是对AI大模型感兴趣的爱好者，这套报告合集都将为您提供宝贵的信息和启示

​因篇幅有限，仅展示部分资料，需要点击文章最下方名片即可前往获取

三、AI大模型经典PDF籍

随着人工智能技术的飞速发展，AI大模型已经成为了当今科技领域的一大热点。这些大型预训练模型，如GPT-3、BERT、XLNet等，以其强大的语言理解和生成能力，正在改变我们对人工智能的认识。 那以下这些PDF籍就是非常不错的学习资源。

因篇幅有限，仅展示部分资料，需要点击文章最下方名片即可前往获取

四、AI大模型商业化落地方案

作为普通人，入局大模型时代需要持续学习和实践，不断提高自己的技能和认知水平，同时也需要有责任感和伦理意识，为人工智能的健康发展贡献力量。