企业官网建设流程全解析

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. 如果知识库文档极大，全量向量化耗时过长，如何做增量更新？

面试题参考答案