Agent的学习记录--5

来源Github的学习指南Agent Learning Hub的学习：

Stage 2——RAG

RAG（检索增强）全流程：chunk -> Embed -> Retrieve -> Answer with Citations

引入库和定义文献来源：

import os
import sys
import math
from typing import List, Dict, Any, Tuple

from config import get_llm, get_embeddings
from langchain_core.documents import Document
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_text_splitters import RecursiveCharacterTextSplitter

PAPERS_DIR = os.path.join(os.path.dirname(__file__), "sample_data", "papers")

Chunk

其涉及到文本切分的策略：递归切分 vs 粗暴切分。掌握 chunk_size和chunk_overlap。为了方便上手，我们先只读取txt格式的文章

由于在一个目录里可能会有多篇txt，又或者在实战中我们可能会有多个数据库查询的现象，我们需要列出所有符合格式要求的文件（对应Line 1-22）：

【TODO (学员实操) 2.1】:

实例化 LangChain 的 RecursiveCharacterTextSplitter：

chunk_size: 分块目标字符数 (使用函数入参 chunk_size)

chunk_overlap: 相邻块之间的重叠字符数 (使用函数入参 chunk_overlap)

separators: 优先级切分符列表，建议: ["\n\n", "\n", "。", "！", "？", " ", ""]

调用 splitter.split_documents(raw_docs) 进行切分。

遍历切分出来的每一个 doc，在其 metadata 中补充 "chunk_id": f"{doc.metadata['source']}_{idx}"。

返回切分后的 Document 列表。

我们接着上面进行：

def load_and_chunk_documents(chunk_size: int = 300, chunk_overlap: int = 50) -> List[Document]:
    """
    读取 sample_data/papers/ 下的所有 txt 论文，
    使用递归字符分块器将其切分为带有元数据 (source, chunk_id) 的 Document 列表。
    """
    raw_docs: List[Document] = []
    
    # 1. 扫描读取本地论文文件
    if not os.path.exists(PAPERS_DIR):
        raise FileNotFoundError(f"目录不存在: {PAPERS_DIR}")
        
    for filename in sorted(os.listdir(PAPERS_DIR)):
        if filename.endswith(".txt"):
            filepath = os.path.join(PAPERS_DIR, filename)
            with open(filepath, "r", encoding="utf-8") as f:
                content = f.read()
                raw_docs.append(Document(
                    page_content=content,
                    metadata={"source": filename, "total_len": len(content)}
                ))
    
    print(f"[Chunking] 成功加载 {len(raw_docs)} 篇原始学术论文。")
    
    # 以下为TODO环节
    delimiter = ["\n\n", "\n", "。", "！", "？", " ", ""]
    splitter = RecursiveCharacterTextSplitter(
        chunk_size = chunk_size,
        chunk_overlap = chunk_overlap, 
        separators=delimiter)
    split_docs = splitter.split_documents(raw_docs) # 分割对象为Document
    for idx, docs in enumerate(split_docs):
        source = docs.metadata.get("source", "docs")
        docs.metadata["chunk_id"] = f"{source}_{idx}"

    return split_docs

在这里一开始的时候，我没有使用enumerate, 我一开始使用了

for idx, docs in range(len(split_docs)), split_docs:

这样子是错误的，因为这样子 range(len(split_docs)), split_docs 会被 Python 解释为一个二元元组 ，导致报错

当我们想同时获得索引和元素的时候，使用内置函数 enumerate()：其中返回的list里第一个元素是索引，第二个元素是序列对象。另外 .append() 方法只适用于列表：

复习添加元素的方法：

对象	方法
list	.append()
set	.add() / .update()
dict	dict[key] = value
pandas dataframe	row: df.append(new_row)column: df['col'] = values
turble	NULL
str	NULL

Embed and retrieve

计算余弦相似度：

余弦相似度的公式为：

cos_{sim} = \frac{\vec{A} \cdot\vec{B}}{||\vec{A}||*||\vec{B}||}

def cosine_similarity(vec_a: List[float], vec_b:list[float]) -> float:
    dot_product = sum(a * b for a, b in zip(vec_a, vec_b)) #zip()方法为取向量a和向量b相同索引的元素
    norm_a = math.sqrt(sum(a * a for a in vec_a))
    norm_b = math.sqrt(sum(b * b for b in vec_b))
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot_product / (norm_a * norm_b)

我们发现我们输入的是要浮点数，但我们的数据都是文本数据，这时候需要使用Embed进行转换：

我们在config.py里建立了一个Embedding的函数，实际上，可以使用开源的Embedding模型的本地部署或者商业Embedding的API：实际代码我个人认为不需要理解，只需要知道这个功能有什么用即可。

class PurePythonTFIDFEmbeddings(Embeddings):
    
    def __init__(self, vector_dim: int = 128):
        self.vector_dim = vector_dim

    def _text_to_vector(self, text: str) -> List[float]:
        tokens = [t.strip().lower() for t in text.replace("\n", " ").split() if t.strip()]
        # 支持字符级 n-gram，以便对中文和短句具有出色的语义重合敏感度
        char_ngrams = [text[i:i+2].lower() for i in range(len(text)-1)]
        all_features = tokens + char_ngrams
        
        vec = [0.0] * self.vector_dim
        for feat in all_features:
            h = abs(hash(feat)) % self.vector_dim
            vec[h] += 1.0
            
        norm = math.sqrt(sum(v * v for v in vec))
        if norm > 0:
            vec = [v / norm for v in vec]
        return vec

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        return [self._text_to_vector(t) for t in texts]

    def embed_query(self, text: str) -> List[float]:
        return self._text_to_vector(text)

def get_embeddings() -> Embeddings:
    """获取向量嵌入模型实例（优先使用轻量级教学嵌入器，保障 100% 离线与兼容性）"""
    return PurePythonTFIDFEmbeddings(vector_dim=256)

另外，Embedding在Agent架构里充当了三大角色：

长期记忆，当比如上下文窗口过多需要压缩时，就可以将过往的记忆存入向量库，每次根据当前提问检索出最相关的几条前情提要。
工具召回 (Tool Retrieval)：如果有大量工具API的接口，直接放进去会挤爆上下文且严重干扰模型判断。此时通过 Embedding 先检索出候选的 Top-5 工具交由 Agent 选择。
本地知识增强：针对几十本专业书籍、代码仓、PDF 文档。Agent 不需要重新训练微调模型，只需检索出最相关的几页进行阅读回答。

在实际上也有些场景我们不需要使用Embedding：

比如现在百万级超长上下文（Long-Context LLMs）时代
符号化精确搜索优于语义相关性的场景
纯流程与多工具协作型 Agent

在学习阶段，我先使用这个AI生成的简陋版Embedding

回到正文编写代码：

class SimpleVectorRetriever:
    def __init__(self, document: List[Document]):
        self.documents = document
        self.embeddings = get_embeddings()
        
        text = [doc.page_content for doc documents]
        self.doc_vectors = self.embeddings.embed_documents(text)
        
    def retrieve(self, query: str, top_k: int = 3, score_threshold: float = 0.1) -> list[Tuple[Document, float]]:
        query_vectors = self.embeddings.embed_query(query)
        related_doc = []
        for doc, doc_vec in zip(self.documents, self.doc_vectors):
            score = cosine_similarit(doc_vec, query_vectors)
            if score >= score_threshold:
                related_doc.append((doc, score))
        related_doc.sort(key=lambda x: x[1], reverse = True)
        return related_doc[:top_k]

Answer with Citations

在这一步，我们想要能够直接生成将检索到的上下文组装为带编号的参考资料，严格要求模型基于资料回答并标记 [Doc X] 引用来源。我们这里最后要输出的是一串字符串

def generate_answer_with_citations(query: str, retrieved_docs_with_scores: List[Tuple[Document, float]]) -> str:
    llm = get_llm(temperature=0.0)
    formatted_context_list = []
    for idx, (doc, score) in enumated(retrieved_docs_with_scores, 1) #index starts with 1
    	source = doc.metabase.get("source", "Unknown resource")
        chunk_id = doc.metabase.get("chunk_id", f"chunk_{idx}")
        formatted_context_list.append(
        	f"--- [Doc {idx}] (来源文件: {source} | 相似度: {score:.3f}) ---\n{doc.page_content}"
        )
    context_str = "\n\n".join(formatted_context_list)
    
    messages = [
        SystemMessage(content=(
            "你是一个学术检索专家。你必须完全基于提供的【参考资料】回答问题，严禁使用外部未验证的知识。\n"
            "要求：\n"
            "1. 关键事实和结论必须注明引用来源，如 [Doc 1] 或 [Doc 2]。\n"
            "2. 如果参考资料中没有相关信息，请明确回答'提供的参考资料中未提及相关信息'，严禁编造！\n"
            "3. 回答结尾附带【引用来源列表】，列出每个 [Doc X] 对应的文件名。"
        )),
        HumanMessage(content=f"【参考资料】:\n{context_str}\n\n【用户问题】:\n{query}")
    ]
    response = llm.invoke(messages)
    return response.content

在 LangChain 中，最规范、也最不容易引起模型误解的写法是采用 SystemMessage（约束指令）+ HumanMessage（动态资料与问题） 分离的模式

在实际应用里：当我们发现RAG效果不好的时候，一方面是大模型本身的问题，但更多的是我们策略的问题，我们首先考虑三点因素：

分块策略选错了（Chunk 太大引入噪音，或太小把关键句腰斩了）；
相似度阈值设低了（导致无关的废话被灌给模型）；
Prompt 缺乏强约束与引用校验

Agent学习记录-5

Agent的学习记录--5

Stage 2——RAG

Chunk

Embed and retrieve

Answer with Citations