Agent 全日制30天速成｜Day3 教学笔记

今日总学习目标

1. 掌握向量嵌入（Embedding）原理、文本向量化、向量相似度计算
2. 搭建简易本地向量库（内存版FAISS），实现文本入库、检索、删除
3. 打通基础RAG检索链路：文档分片→向量化存储→用户提问召回片段→注入Prompt
4. 结合Day2工具调用，实现「RAG知识库检索 + Function Calling」混合Agent
   每日时长分配（全天8h）

• 理论笔记阅读+理解：2.5h
• 代码编写调试：4h
• 复盘+面试题背诵：1.5h

一、核心理论教学笔记

1. Embedding 向量嵌入基础

1.1 核心概念

Embedding（向量嵌入）：将自然语言文本转化为固定长度浮点数字向量，语义相近的文本，向量在空间距离更近。

• 用途：知识库检索、文档相似度匹配、分类聚类、对话记忆检索
• 兼容规范：国内通义千问、DeepSeek、OpenAI均提供统一Embedding接口，同样只替换base_url与key即可复用代码

1.2 相似度计算（RAG核心）

常用两种计算方式：

1. 余弦相似度 Cosine Similarity（向量检索标准）
   • 取值范围[-1,1]，越接近1代表语义越相似
   • 不受向量长度影响，长/短文本对比首选
2. 欧式距离：距离越小越相似，长文本向量不推荐

1.3 文本分片（Chunk）规则

大模型单次Embedding存在输入长度限制，长文档必须拆分片段：

• 固定长度分片：按字符/Token切割（简单易实现，缺点容易割裂完整语义）
• 重叠分片：每段chunk保留前后重叠文本，解决上下文割裂问题（工程首选）
• 分片长度建议：200~600汉字，兼顾检索精度与上下文完整性

1.4 Embedding 接口调用要点

1. 输入支持批量文本，单次批量提交降低接口调用次数、节省计费
2. 批量接口有单次文本条数上限，超过需做分批切割
3. Embedding无流式输出，均为同步返回向量数组
4. 向量化过程会消耗Token，批量任务需增加并发限流、重试机制

2. 向量数据库基础（FAISS内存版）

2.1 FAISS介绍

Meta开源轻量向量检索库，适合本地测试、小型知识库，无需部署服务：

• 支持向量添加、批量入库、相似度TopK检索
• 支持向量与原文映射存储（索引→文本元数据）
• 百万级以内向量检索速度极快，适合学习阶段使用

2.2 向量库存储两层结构

1. FAISS索引：仅存储浮点向量，用于快速相似度检索
2. 元数据映射字典：index_id -> {原文文本、来源、标题}，检索拿到id后反向取出原文

2.3 检索流程

用户提问 → 提问文本向量化 → FAISS检索TopN相似向量 → 通过id匹配原文片段 → 拼接进Prompt交给LLM

3. RAG检索增强生成完整链路（标准5步）

1. 文档加载与分块：读取知识库文档，切割为重叠Chunk片段
2. 文本向量化入库：调用Embedding接口，所有Chunk生成向量存入FAISS
3. 用户提问向量化：用户问题单独生成查询向量
4. 相似度召回：检索最相似的Top-K知识库片段
5. 上下文增强生成：把召回文档作为参考资料塞入System Prompt，限制模型仅基于参考内容回答，减少幻觉

3.4 RAG解决的核心问题

• 大模型知识截止时间问题，可接入实时业务文档、私有知识库
• 大幅降低模型幻觉，强制模型引用检索到的资料作答
• 减少超长上下文Token消耗，无需全量传入知识库，只召回相关片段

4. RAG + Function Calling 混合Agent逻辑

两条能力互补：

1. 私有文档、内部知识库查询 → 使用RAG向量检索
2. 实时数据、计算、外部接口查询 → 使用Function Calling工具调用
   执行流程：
   用户提问 → 先执行RAG召回知识库内容 → 将检索片段注入System Prompt → 模型判断是否需要调用工具 → 执行工具闭环 → 结合知识库+工具结果输出最终答案

二、今日学习重点

1. 封装兼容OpenAI标准的Embedding异步调用客户端
2. 实现重叠文本分片工具函数
3. 封装内存FAISS向量库，完成增、查基础操作
4. 搭建完整单轮RAG问答链路，限制模型仅基于检索内容作答
5. 融合Day2工具调用，实现RAG+Function Calling混合智能体

三、今日难点 & 解决方案

难点1：分片切割破坏完整语义，检索结果不相关

解决方案：

1. 开启重叠分片，设置50~100字重叠区间
2. 分片长度控制在300~500汉字，避免单块内容过短/过长
3. 检索后增加重排逻辑（简单版：过滤相似度低于阈值的片段）

难点2：向量检索返回无关文本，模型答非所问

解决方案：

1. 设置相似度阈值，低于阈值直接丢弃片段，不注入Prompt
2. Prompt强约束：没有相关资料时统一回复“暂无相关资料”，禁止编造内容
3. 优化Embedding输入，对知识库chunk做简单摘要再向量化

难点3：批量Embedding接口超限、请求频繁触发限流

解决方案：

1. 封装批量分批工具，按接口最大条数拆分文本列表
2. 使用信号量控制Embedding并发，捕获429限流指数退避重试
3. 向量本地缓存，重复文本无需重复调用Embedding接口

难点4：RAG与工具调用逻辑冲突，模型优先调用工具忽略知识库

解决方案：

1. System Prompt明确优先级：先参考知识库内容，无法解决再调用工具
2. 检索片段放在Prompt最靠前位置，提升权重
3. Few-shot示例演示“先读知识库，再判断是否需要工具”

四、完整练习代码（基于Day1、Day2代码扩展）

依赖安装

pip install faiss-cpu numpy aiohttp pydantic fastapi uvicorn

1. embedding向量客户端 + FAISS向量库 rag_store.py

import aiohttp
import asyncio
import faiss
import numpy as np
import json
from typing import List, Dict, Tuple

# 模型配置（沿用前两日）
MODEL_CONFIG = {
    "qwen-turbo": {
        "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
        "embedding_url": "https://dashscope.aliyuncs.com/compatible-mode/v1/embeddings",
        "api_key": "你的通义千问key"
    },
    "deepseek-chat": {
        "base_url": "https://api.deepseek.com/v1/chat/completions",
        "embedding_url": "https://api.deepseek.com/v1/embeddings",
        "api_key": "你的deepseek key"
    }
}

# 文本分片工具（重叠分片）
def split_text_chunk(text: str, chunk_size: int = 400, overlap: int = 80) -> List[str]:
    chunks = []
    start = 0
    text_len = len(text)
    while start < text_len:
        end = min(start + chunk_size, text_len)
        chunk = text[start:end]
        chunks.append(chunk.strip())
        start += chunk_size - overlap
    return chunks

# 异步Embedding客户端
class AsyncEmbeddingClient:
    def __init__(self, model_name: str = "qwen-turbo"):
        self.conf = MODEL_CONFIG[model_name]
        self.semaphore = asyncio.Semaphore(3)
        self.timeout = aiohttp.ClientTimeout(total=30)

    async def batch_embedding(self, texts: List[str]) -> List[List[float]]:
        headers = {
            "Authorization": f"Bearer {self.conf['api_key']}",
            "Content-Type": "application/json"
        }
        payload = {
            "input": texts,
            "model": "text-embedding-v1"
        }
        async with self.semaphore:
            async with aiohttp.ClientSession(timeout=self.t