AI Agent 30天速成|Day3 笔记

发布时间:2026/7/5 3:16:53

AI Agent 30天速成|Day3 笔记 掌握向量嵌入Embedding原理、文本向量化、向量相似度计算搭建简易本地向量库内存版FAISS实现文本入库、检索、删除打通基础RAG检索链路文档分片→向量化存储→用户提问召回片段→注入Prompt结合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核心常用两种计算方式余弦相似度 Cosine Similarity向量检索标准取值范围[-1,1]越接近1代表语义越相似不受向量长度影响长/短文本对比首选欧式距离距离越小越相似长文本向量不推荐1.3 文本分片Chunk规则大模型单次Embedding存在输入长度限制长文档必须拆分片段固定长度分片按字符/Token切割简单易实现缺点容易割裂完整语义重叠分片每段chunk保留前后重叠文本解决上下文割裂问题工程首选分片长度建议200~600汉字兼顾检索精度与上下文完整性1.4 Embedding 接口调用要点输入支持批量文本单次批量提交降低接口调用次数、节省计费批量接口有单次文本条数上限超过需做分批切割Embedding无流式输出均为同步返回向量数组向量化过程会消耗Token批量任务需增加并发限流、重试机制2. 向量数据库基础FAISS内存版2.1 FAISS介绍Meta开源轻量向量检索库适合本地测试、小型知识库无需部署服务支持向量添加、批量入库、相似度TopK检索支持向量与原文映射存储索引→文本元数据百万级以内向量检索速度极快适合学习阶段使用2.2 向量库存储两层结构FAISS索引仅存储浮点向量用于快速相似度检索元数据映射字典index_id - {原文文本、来源、标题}检索拿到id后反向取出原文2.3 检索流程用户提问 → 提问文本向量化 → FAISS检索TopN相似向量 → 通过id匹配原文片段 → 拼接进Prompt交给LLM3. RAG检索增强生成完整链路标准5步文档加载与分块读取知识库文档切割为重叠Chunk片段文本向量化入库调用Embedding接口所有Chunk生成向量存入FAISS用户提问向量化用户问题单独生成查询向量相似度召回检索最相似的Top-K知识库片段上下文增强生成把召回文档作为参考资料塞入System Prompt限制模型仅基于参考内容回答减少幻觉3.4 RAG解决的核心问题大模型知识截止时间问题可接入实时业务文档、私有知识库大幅降低模型幻觉强制模型引用检索到的资料作答减少超长上下文Token消耗无需全量传入知识库只召回相关片段4. RAG Function Calling 混合Agent逻辑两条能力互补私有文档、内部知识库查询 → 使用RAG向量检索实时数据、计算、外部接口查询 → 使用Function Calling工具调用执行流程用户提问 → 先执行RAG召回知识库内容 → 将检索片段注入System Prompt → 模型判断是否需要调用工具 → 执行工具闭环 → 结合知识库工具结果输出最终答案二、今日学习重点封装兼容OpenAI标准的Embedding异步调用客户端实现重叠文本分片工具函数封装内存FAISS向量库完成增、查基础操作搭建完整单轮RAG问答链路限制模型仅基于检索内容作答融合Day2工具调用实现RAGFunction Calling混合智能体三、今日难点 解决方案难点1分片切割破坏完整语义检索结果不相关解决方案开启重叠分片设置50~100字重叠区间分片长度控制在300~500汉字避免单块内容过短/过长检索后增加重排逻辑简单版过滤相似度低于阈值的片段难点2向量检索返回无关文本模型答非所问解决方案设置相似度阈值低于阈值直接丢弃片段不注入PromptPrompt强约束没有相关资料时统一回复“暂无相关资料”禁止编造内容优化Embedding输入对知识库chunk做简单摘要再向量化难点3批量Embedding接口超限、请求频繁触发限流解决方案封装批量分批工具按接口最大条数拆分文本列表使用信号量控制Embedding并发捕获429限流指数退避重试向量本地缓存重复文本无需重复调用Embedding接口难点4RAG与工具调用逻辑冲突模型优先调用工具忽略知识库解决方案System Prompt明确优先级先参考知识库内容无法解决再调用工具检索片段放在Prompt最靠前位置提升权重Few-shot示例演示“先读知识库再判断是否需要工具”四、完整练习代码基于Day1、Day2代码扩展依赖安装pip install faiss-cpu numpy aiohttp pydantic fastapi uvicorn1. embedding向量客户端 FAISS向量库 rag_store.pyimport 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(total30) async def batch_embedding(self, texts: List[str]) - List[List[float]]: headers { Authorization: fBearer {self.conf[api_key]}, Content-Type: application/json } payload { input: texts, model: text-embedding-v1 } async with self.semaphore: async with aiohttp.ClientSession(timeoutself.timeout) as session: resp await session.post(self.conf[embedding_url], jsonpayload, headersheaders) 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, dtypenp.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], dtypenp.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_nameqwen-turbo): self.emb_client AsyncEmbeddingClient(model_name) self.vec_store FaissVectorStore(vec_dim1536) # 文档入库文本分片向量化存入向量库 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_k3): query_vec await self.emb_client.batch_embedding([query]) return self.vec_store.search(query_vec[0], top_ktop_k) # 测试RAG async def test_rag(): rag RAGService() # 模拟知识库文档 doc AI Agent 是具备工具调用、记忆、规划能力的智能体。 Agent可以读取私有知识库结合计算器、数据库等外部工具完成复杂任务。 RAG检索增强生成用于解决大模型知识滞后、幻觉问题先检索文档再回答用户问题。 Function Calling允许模型自主调用外部函数完成数学计算、实时数据查询。 await rag.add_document(doc, sourceAgent基础文档) # 用户提问召回 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.pyimport 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_nameqwen-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_k3) rag_context \n.join([item[text] for item in retrieve_list]) # 2. 拼接系统提示词 system_prompt self.system_base.format(rag_contextrag_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, sourceAgent知识库) # 测试1知识库相关问题 print(问题1什么是RAG) res1 await agent.chat(什么是RAG) print(回答, res1[answer]) print(参考文档, res1[reference_content]) # 测试2数学计算触发工具调用 print(\n问题2365*2等于多少) res2 await agent.chat(365*2等于多少) print(回答, res2[answer]) if __name__ __main__: asyncio.run(test_mixed_agent())3. FastAPI 一体化接口 main_rag.pyfrom fastapi import FastAPI, Query import asyncio from llm_rag_agent import MixedRAGAgent app FastAPI(titleDay3 RAGFunction 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, sourceDay3学习文档) 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, reloadTrue)五、今日必做练习任务填入有效API Key运行rag_store.py测试文档分片、向量入库、相似度检索修改分片chunk_size、overlap参数观察召回内容完整性变化调整检索score_threshold阈值观察无关片段过滤效果运行混合Agent代码分别测试知识库问题、数学计算问题验证RAG与工具调用双逻辑启动FastAPI连续提问查看接口返回的参考文档片段新增一段无关知识库文本测试模型不会引用无关内容六、今日配套面试题RAG入门必考基础问答什么是Embedding向量嵌入余弦相似度在RAG中的作用RAG完整执行流程分为哪几步RAG解决了大模型哪些痛点文本分块为什么需要设置重叠分片固定长度分片有什么缺陷简述FAISS的作用内存向量库的优缺点RAG Function Calling混合架构的执行顺序和设计思路工程实操题检索出无关文本如何优化召回效果写出至少3种方案批量Embedding调用时如何处理接口批量上限、并发限流问题如何通过Prompt约束模型只能基于检索到的知识库内容回答避免幻觉内存FAISS重启后向量丢失线上生产环境如何持久化向量数据拓展思考题进阶Top-K检索只取固定条数如何实现结果重排提升检索精度海量百万级知识库本地FAISS无法承载有哪些商用向量数据库可选如果知识库文档极大全量向量化耗时过长如何做增量更新面试题参考答案

相关新闻