AI 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.timeout) as session:
                resp = await session.post(self.conf["embedding_url"], json=payload, headers=headers)
                res_data = await resp.json()
                vecs = [item["embedding"] for item in res_data["data"]]
                return vecs

# 内存FAISS向量库封装
class FaissVectorStore:
    def __init__(self, vec_dim: int = 1536):
        self.index = faiss.IndexFlatL2(vec_dim)
        self.meta_map: Dict[int, Dict] = {}
        self.current_id = 0

    # 批量插入向量+原文
    def add_batch(self, vectors: List[List[float]], metas: List[Dict]):
        arr = np.array(vectors, dtype=np.float32)
        self.index.add(arr)
        for meta in metas:
            self.meta_map[self.current_id] = meta
            self.current_id += 1

    # 相似度检索TopK
    def search(self, query_vec: List[float], top_k: int = 3, score_threshold: float = 600) -> List[Dict]:
        arr = np.array([query_vec], dtype=np.float32)
        distances, ids = self.index.search(arr, top_k)
        results = []
        for dist, idx in zip(distances[0], ids[0]):
            if idx == -1:
                continue
            if dist > score_threshold:
                continue
            meta = self.meta_map.get(idx, {})
            meta["distance"] = float(dist)
            results.append(meta)
        return results

# RAG统一封装类
class RAGService:
    def __init__(self, model_name="qwen-turbo"):
        self.emb_client = AsyncEmbeddingClient(model_name)
        self.vec_store = FaissVectorStore(vec_dim=1536)

    # 文档入库：文本分片+向量化+存入向量库
    async def add_document(self, doc_text: str, source: str = "默认文档"):
        chunks = split_text_chunk(doc_text)
        if not chunks:
            return
        vecs = await self.emb_client.batch_embedding(chunks)
        metas = [{"text": c, "source": source} for c in chunks]
        self.vec_store.add_batch(vecs, metas)

    # 用户提问召回知识库片段
    async def retrieve(self, query: str, top_k=3):
        query_vec = await self.emb_client.batch_embedding([query])
        return self.vec_store.search(query_vec[0], top_k=top_k)

# 测试RAG
async def test_rag():
    rag = RAGService()
    # 模拟知识库文档
    doc = """AI Agent 是具备工具调用、记忆、规划能力的智能体。
    Agent可以读取私有知识库，结合计算器、数据库等外部工具完成复杂任务。
    RAG检索增强生成用于解决大模型知识滞后、幻觉问题，先检索文档再回答用户问题。
    Function Calling允许模型自主调用外部函数，完成数学计算、实时数据查询。"""
    await rag.add_document(doc, source="Agent基础文档")
    # 用户提问召回
    res = await rag.retrieve("RAG有什么作用")
    print("召回知识库片段：")
    for item in res:
        print(f"相似度距离：{item['distance']} 内容：{item['text']}")

if __name__ == "__main__":
    asyncio.run(test_rag())

2. RAG+工具调用混合LLM服务 llm_rag_agent.py

import asyncio
from pydantic import BaseModel, Field, ValidationError
import json
from rag_store import RAGService
# 复用Day2 LLM客户端、工具定义
from llm_client_v2 import AsyncLLMClientV2, TOOLS, TOOL_MAP, CalcToolParams

# 混合Agent整合类
class MixedRAGAgent:
    def __init__(self, model_name="qwen-turbo"):
        self.llm = AsyncLLMClientV2(model_name)
        self.rag = RAGService(model_name)
        self.system_base = """
你是智能AI助手，回答严格遵守以下规则：
1. 优先参考下方知识库检索内容作答，禁止编造不存在信息；
2. 如果知识库无相关内容，再判断是否需要调用计算器工具；
3. 数学计算必须调用calculator工具，禁止手动计算；
4. 无资料、无可用工具时，直接回复暂无相关信息。
【知识库参考资料】：
{rag_context}
"""

    # 完整问答链路：RAG召回 + 工具调用闭环
    async def chat(self, user_query: str):
        # 1. RAG召回相关文档
        retrieve_list = await self.rag.retrieve(user_query, top_k=3)
        rag_context = "\n".join([item["text"] for item in retrieve_list])
        # 2. 拼接系统提示词
        system_prompt = self.system_base.format(rag_context=rag_context)
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_query}
        ]
        # 3. 执行工具调用对话
        answer = await self.llm.chat_with_tools(messages, TOOLS)
        return {
            "answer": answer,
            "reference_doc_count": len(retrieve_list),
            "reference_content": rag_context
        }

# 测试混合Agent
async def test_mixed_agent():
    agent = MixedRAGAgent()
    # 预先写入知识库
    doc = """RAG全称检索增强生成，作用是接入私有文档减少模型幻觉。
    Function Calling是大模型调用外部工具的能力，支持计算器、接口查询。"""
    await agent.rag.add_document(doc, source="Agent知识库")

    # 测试1：知识库相关问题
    print("问题1：什么是RAG？")
    res1 = await agent.chat("什么是RAG？")
    print("回答：", res1["answer"])
    print("参考文档：", res1["reference_content"])

    # 测试2：数学计算，触发工具调用
    print("\n问题2：365*2等于多少？")
    res2 = await agent.chat("365*2等于多少？")
    print("回答：", res2["answer"])

if __name__ == "__main__":
    asyncio.run(test_mixed_agent())

3. FastAPI 一体化接口 main_rag.py

from fastapi import FastAPI, Query
import asyncio
from llm_rag_agent import MixedRAGAgent

app = FastAPI(title="Day3 RAG+Function Calling混合Agent")
agent = MixedRAGAgent()

# 预加载知识库文档
@app.on_event("startup")
async def load_knowledge():
    doc = """
AI Agent 30天速成课程Day3学习内容：
1. Embedding向量嵌入将文本转为数字向量，用于相似度检索；
2. FAISS本地向量库存储向量，实现私有知识库检索；
3. RAG检索增强生成解决大模型幻觉、知识滞后问题；
4. RAG结合Function Calling，同时支持知识库查询与外部工具调用；
5. 文本分片采用重叠切割，避免语义断裂。
"""
    await agent.rag.add_document(doc, source="Day3学习文档")

@app.get("/chat/rag")
async def chat_rag(prompt: str = Query(..., description="用户提问")):
    result = await agent.chat(prompt)
    return result

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main_rag:app", reload=True)

五、今日必做练习任务

1. 填入有效API Key，运行rag_store.py，测试文档分片、向量入库、相似度检索
2. 修改分片chunk_size、overlap参数，观察召回内容完整性变化
3. 调整检索score_threshold阈值，观察无关片段过滤效果
4. 运行混合Agent代码，分别测试知识库问题、数学计算问题，验证RAG与工具调用双逻辑
5. 启动FastAPI，连续提问，查看接口返回的参考文档片段
6. 新增一段无关知识库文本，测试模型不会引用无关内容

六、今日配套面试题（RAG入门必考）

基础问答

1. 什么是Embedding向量嵌入？余弦相似度在RAG中的作用？
2. RAG完整执行流程分为哪几步？RAG解决了大模型哪些痛点？
3. 文本分块为什么需要设置重叠分片？固定长度分片有什么缺陷？
4. 简述FAISS的作用，内存向量库的优缺点？
5. RAG + Function Calling混合架构的执行顺序和设计思路？

工程实操题

1. 检索出无关文本，如何优化召回效果？写出至少3种方案
2. 批量Embedding调用时，如何处理接口批量上限、并发限流问题？
3. 如何通过Prompt约束模型只能基于检索到的知识库内容回答，避免幻觉？
4. 内存FAISS重启后向量丢失，线上生产环境如何持久化向量数据？

拓展思考题（进阶）

1. Top-K检索只取固定条数，如何实现结果重排提升检索精度？
2. 海量百万级知识库，本地FAISS无法承载，有哪些商用向量数据库可选？
3. 如果知识库文档极大，全量向量化耗时过长，如何做增量更新？

面试题参考答案

基础问答

1. Embedding与余弦相似度
   Embedding把文本映射为低维浮点向量，语义相近文本向量空间距离更近；余弦相似度衡量两个向量夹角，取值0~1，数值越高代表语义越相似，是RAG检索排序标准。

2. RAG流程与解决痛点
   流程：文档分块→向量化入库→提问向量召回片段→片段注入Prompt生成回答。
   解决痛点：模型知识截止、私有业务数据无法输入、模型自由编造内容（幻觉）、超长上下文Token超限。

3. 重叠分片作用
   固定长度分片会把完整语义句子拆分成两段，导致检索缺失上下文；重叠分片每段保留前后一小段文本，保证跨分片语义完整，提升检索准确率。

4. FAISS优缺点
   FAISS是开源向量检索库，快速计算向量相似度。
   优点：本地运行无需部署服务、检索速度快、上手简单；缺点：仅内存存储，重启丢失数据，不支持海量数据持久化、复杂过滤条件。

5. 混合架构执行顺序
   先执行RAG召回私有知识库内容，将检索片段注入系统提示词；再交由模型判断是否需要调用外部工具；工具执行结果回填上下文，模型结合知识库+工具数据给出最终回答。

工程实操题

1. 优化无关检索片段方案

• 设置相似度阈值，过滤距离过大低相关文本；
• 采用重叠分片提升Chunk语义完整性；
• Prompt强约束禁止使用知识库外信息；
• 增加检索重排，过滤弱相关内容；
• 优化Embedding输入，对长文档做摘要再向量化。

2. 批量Embedding限流与分批
   按接口最大单次文本数量切割列表，分批提交；使用信号量限制并发；捕获429限流异常，指数退避重试；缓存重复文本向量，减少重复调用。

3. 约束模型减少幻觉
   System Prompt明确：无相关资料统一回复无信息，禁止编造；将检索片段放在Prompt最前，提升权重；配合temperature=0降低随机性；Few-shot示例演示正确回答方式。

4. 向量持久化方案
   生产不使用内存FAISS；选用支持持久化的向量库：Chroma、Milvus、Qdrant、PGVector；向量落地磁盘，支持重启加载、增量更新。

拓展思考题

1. 检索重排
   基础粗召回（FAISS TopK）→ 重排模型（Cross-Encoder）对召回片段与问题打分，过滤低分片段，大幅提升精准度。
2. 线上向量数据库
   开源：Milvus、Qdrant、Chroma、PGVector；云厂商托管：阿里云DashVector、腾讯云向量库、Pinecone。
3. 知识库增量更新
   记录文档唯一ID，仅对新增/修改文档分片向量化；增量插入向量库，不重复处理全量历史文档；维护文档-向量映射关系，支持删除旧向量。

学习总结

Day3完成AI Agent三大核心能力之一：RAG检索增强，打通「私有知识库读取」链路，同时融合Day2的Function Calling实现混合智能体。
向量嵌入、文本分片、相似度检索是RAG底层核心，面试高频提问；内存FAISS适合本地学习，为后续持久化向量库、高级重排、多模态RAG打下基础。
至此底层工具链全部成型：异步LLM客户端、SSE流式、Function Calling、多轮会话、RAG知识库，完整支撑简易Agent开发。