Agent 全日制30天速成｜Day9 笔记

全部能力工具化文档入库、文本Embedding、知识库检索、数学计算全部注册为标准Function工具ReAct自主调用无硬编码业务逻辑向量存储替换为 Chroma持久向量库彻底移除内存列表/FAISS支持磁盘持久、元数据过滤工具嵌套执行rag_search、vector_add 底层自动调用 text_embedding 工具生成语义向量整套代码完全独立不依赖之前代码限流/熔断/权限/日志对所有工具统一生效所有向量生成统一调用厂商Embedding接口不使用本地字符假向量今日总学习目标实现全工具化架构向量入库、向量化、检索、计算均为可调用工具基于Chroma本地持久向量库完成知识库增删查文档入库也走工具调用掌握工具嵌套执行逻辑上层检索工具自动调用底层Embedding工具统一中间件体系对四类工具全覆盖分层超时隔离Embedding/LLM请求独立运行整套ReAct Agent无任何前置课程代码依赖每日时长分配全天8h理论笔记阅读2.5h分层代码编写调试4h复盘面试背诵1.5h一、核心理论教学笔记1 全工具化分层设计底层基础工具依赖Embedding接口text_embedding批量文本转语义向量所有向量操作唯一底层依赖业务向量工具嵌套底层embeddingvector_add新增知识库文档自动向量化存入Chromarag_search检索知识库自动生成查询向量Chroma相似度召回数值工具calculator四则数学运算所有工具全部注册到统一网关权限、限流、熔断、日志一套逻辑复用2 Chroma向量库优势对比内存/FAISS磁盘持久化程序重启向量与文档不丢失原生元数据支持可标记文档来源、分类检索过滤开箱即用无需手动维护向量id映射字典内置相似度排序支持MMR去重检索同步API简单适配异步业务3 完整工具化执行链路用户提问接口生成全局TraceID执行输入安全校验、令牌桶限流读取Redis会话记忆自动滑动窗口摘要压缩ReAct循环推理模型自主选择工具数学题 → calculator新增资料 → vector_add内部调用text_embedding存入Chroma查询知识 → rag_search内部调用text_embedding生成查询向量Chroma召回网关统一拦截权限校验 → 熔断判断 → 指数退避重试 → 独立超时执行工具结果回填对话上下文反思判断信息是否充足汇总全部观测结果LLM生成脱敏最终回答对话持久存入Redis全链路日志落地4 生产中间件全覆盖规则LLM全局60s超时Embedding独立15s超时令牌桶统一管控所有工具请求削峰防429限流单工具连续失败触发熔断冷却后试探恢复访客仅可闲聊普通用户可用计算、检索管理员可批量入库文档每条工具调用携带TraceID写入日志记录耗时、参数、异常二、今日学习重点将文档入库、向量化、检索全部封装为可调用工具实现纯工具驱动AgentChroma持久向量库接入项目上层工具无感知读写向量库实现工具嵌套调用上层检索/入库工具自动调用Embedding底层工具安全、限流、熔断、日志中间件对四类工具统一生效独立可运行完整项目无Day1~Day8代码依赖三、今日难点 解决方案难点1向量相关逻辑散落在代码各处无法监控限流解决方案全部封装标准Function统一网关调度所有向量操作都经过中间件拦截难点2程序重启知识库丢失解决方案使用Chroma持久客户端向量数据落地本地文件夹难点3Embedding高频调用频繁触发平台限流解决方案Embedding工具独立超时令牌桶管控QPS熔断拦截连续失败请求难点4模型不会自主执行文档入库/检索解决方案System提示词完整列出全部工具名称与用途搭配少样本示例四、完整可运行代码依赖安装pip install aiohttp pydantic fastapi uvicorn aioredis python-dotenv chromadb numpy项目目录day9_full_tool_agent/├── .env # 环境配置├── middleware.py # 限流/熔断/重试/日志├── security.py # 注入防护、敏感脱敏、权限├── llm_client.py # LLMEmbedding异步客户端├── tool_gateway.py # 全量工具注册Chroma向量工具计算器├── memory_store.py # Redis分层持久记忆├── agent_core.py # ReAct主智能体└── main.py # FastAPI入口1 .env 配置文件LLM EmbeddingLLM_BASE_URLhttps://dashscope.aliyuncs.com/compatible-mode/v1/chat/completionsLLM_EMBED_URLhttps://dashscope.aliyuncs.com/compatible-mode/v1/embeddingsLLM_API_KEY你的API_KEYRedisREDIS_URLredis://127.0.0.1:6379限流熔断TOKEN_BUCKET_CAP12TOKEN_RATE3MAX_FAIL_TIMES3COOLDOWN_SECONDS10开关ENABLE_SAFE_FILTERtrueENABLE_LOGtrueChroma持久化路径CHROMA_PERSIST_PATH./chroma_kbCHROMA_COLLECTION_NAMEagent_kb2 middleware.pyimport asyncioimport timeimport jsonimport osimport uuidfrom dotenv import load_dotenvload_dotenv()令牌桶限流class TokenBucket:definit(self, cap: int, rate: float):self.capacity capself.rate rateself.token_num capself.last_refill_time time.time()def refill(self): now time.time() delta now - self.last_refill_time add_tokens delta * self.rate self.token_num min(self.capacity, self.token_num add_tokens) self.last_refill_time now async def get_token(self) - bool: self.refill() if self.token_num 1: self.token_num - 1 return True return Falseglobal_bucket TokenBucket(int(os.getenv(“TOKEN_BUCKET_CAP”)), float(os.getenv(“TOKEN_RATE”)))熔断降级class CircuitBreaker:definit(self, max_fail: int, cool_sec: int):self.max_fail max_failself.cool cool_secself.fail_count 0self.state “closed”self.open_start 0async def can_run(self) - bool: now time.time() if self.state open: if now - self.open_start self.cool: self.state half_open return True return False return True def success(self): self.fail_count 0 self.state closed def fail(self): self.fail_count 1 if self.fail_count self.max_fail and self.state ! open: self.state open self.open_start time.timebreaker_map {}def get_breaker(tool_name: str):if tool_name not in breaker_map:breaker_map[tool_name] CircuitBreaker(int(os.getenv(“MAX_FAIL_TIMES”)), int(os.getenv(“COOLDOWN_SECONDS”)))return breaker_map[tool_name]指数退避重试async def backoff_retry(call_func, max_retry3):delay 1for _ in range(max_retry):try:return await call_func()except Exception as e:err_str str(e)if “401” in err_str or “参数非法” in err_str or “权限不足” in err_str:raise eawait asyncio.sleep(delay)delay min(delay * 2, 8)return await call_func()日志埋点class AgentLog:definit(self):self.log_file “./agent_trace.log”self.switch os.getenv(“ENABLE_LOG”) “true”def write(self, trace_id: str, level: str, content: dict): if not self.switch: return log_data { time: time.strftime(%Y-%m-%d %H:%M:%S), trace_id: trace_id, level: level, **content } with open(self.log_file, a, encodingutf-8) as f: f.write(json.dumps(log_data, ensure_asciiFalse) \n)log_client AgentLog()def create_trace_id() - str:return str(uuid.uuid4())3 security.pyimport reimport osSENSITIVE_RULES [re.compile(r1[3-9]\d{9}“),re.compile(r”\d{17}[\dXx]“),re.compile(r”\d{16,19}“)]INJECT_MARKS [”“, ”“”‘, ’‘’]工具权限分级TOOL_AUTH {“calculator”: “user”,“text_embedding”: “user”,“rag_search”: “user”,“vector_add”: “admin”}def escape_inject(text: str) - str:for mark in INJECT_MARKS:text text.replace(mark, mark[:-1])return textdef desensitize(text: str) - str:for pat in SENSITIVE_RULES:text pat.sub(“******”, text)return textdef check_tool_auth(tool_name: str, user_role: str) - bool:need_auth TOOL_AUTH.get(tool_name, “admin”)if need_auth “user” and user_role in [“user”, “admin”]:return Trueif need_auth “admin” and user_role “admin”:return Truereturn Falseasync def input_verify(raw_input: str, max_len800) - tuple[bool, str]:if len(raw_input) max_len:return False, “输入过长请精简提问”safe_text escape_inject(raw_input)return True, safe_text4 llm_client.pyimport aiohttpimport asyncioimport jsonimport refrom typing import List, Dict, AsyncGeneratorfrom dotenv import load_dotenvimport osload_dotenv()class AsyncLLMClient:definit(self):self.base_url os.getenv(“LLM_BASE_URL”)self.embed_url os.getenv(“LLM_EMBED_URL”)self.api_key os.getenv(“LLM_API_KEY”)self.headers {“Authorization”: fBearer {self.api_key},“Content-Type”: “application/json”}self.timeout_llm aiohttp.ClientTimeout(total60)self.timeout_emb aiohttp.ClientTimeout(total15)async def chat_sync(self, messages: List[Dict], temperature0.1, toolsNone): payload { model: qwen-turbo, messages: messages, temperature: temperature, stream: False } if tools: payload[tools] tools async with aiohttp.ClientSession(timeoutself.timeout_llm) as session: resp await session.post(self.base_url, jsonpayload, headersself.headers) return await resp.json() async def chat_stream(self, messages: List[Dict], temperature0.1): payload { model: qwen-turbo, messages: messages, temperature: temperature, stream: True } async with aiohttp.ClientSession(timeoutself.timeout_llm) as session: async with session.post(self.base_url, jsonpayload, headersself.headers) as resp: buffer async for chunk in resp.content.iter_chunked(1024): buffer chunk.decode(utf-8) while data: in buffer: idx buffer.find(data:) end buffer.find(\n\n, idx) if end -1: break block buffer[idx5:end].strip() buffer buffer[end2:] if block [DONE]: return try: item json.loads(block) delta item[choices][0][delta].get(content, ) if delta: yield delta except: continue async def batch_embedding(self, text_list: List[str]) - List[List[float]]: payload { model: text-embedding-v1, input: text_list } async with aiohttp.ClientSession(timeoutself.timeout_emb) as session: resp await session.post(self.embed_url, jsonpayload, headersself.headers) res_data await resp.json() vecs [i[embedding] for i in res_data[data]] return vecs async def chat_json(self, messages: List[Dict], schema): prompt_ext f仅输出标准JSON无多余文字JSON规范{schema.model_json_schema()} new_msg messages.copy() new_msg[-1][content] prompt_ext raw await self.chat_sync(new_msg, temperature0.0)[choices][0][message][content] match re.search(r\{.*\}, raw, re.S) if not match: raw await self.chat_sync(new_msg, 0.0)[choices][0][message][content] match re.search(r\{.*\}, raw, re.S) return schema.model_validate_json(match.group())llm_client AsyncLLMClient()5 tool_gateway.py核心全向量操作工具化Chromaimport asyncioimport chromadbimport osfrom dotenv import load_dotenvimport numpy as npfrom pydantic import BaseModel, Fieldfrom typing import List, Dictfrom llm_client import llm_clientfrom middleware import get_breaker, backoff_retry, log_clientfrom security import check_tool_authload_dotenv()Chroma初始化CHROMA_PATH os.getenv(“CHROMA_PERSIST_PATH”)COLL_NAME os.getenv(“CHROMA_COLLECTION_NAME”)chroma_client chromadb.PersistentClient(pathCHROMA_PATH)coll chroma_client.get_or_create_collection(nameCOLL_NAME, metadata{“hnsw:space”: “cosine”}) 全部工具参数模型 class CalcArgs(BaseModel):num1: float Field(description“第一个数字”)num2: float Field(description“第二个数字”)op: str Field(description“运算符 ±*/”)class EmbeddingArgs(BaseModel):text_list: List[str] Field(description“待向量化文本数组”)class VectorAddArgs(BaseModel):text: str Field(description“入库知识库文本”)source: str Field(default“default”, description“文档来源标签”)class RagSearchArgs(BaseModel):query: str Field(description“检索问题”)top_k: int Field(default3) 底层工具函数 async def calculator(num1, num2, op):try:match op:case “”: res num1 num2case “-”: res num1 - num2case “*”: res num1 * num2case “/”:if num2 0:return “计算失败除数不能为0”res num1 / num2case _: return f不支持运算符{op}return f计算结果{num1}{op}{res}except Exception as e:return f计算异常{str(e)}底层向量工具调用Embedding接口async def text_embedding(text_list: List[str]):return await llm_client.batch_embedding(text_list)入库工具嵌套embedding存入Chromaasync def vector_add(text: str, source: str):vecs await text_embedding([text])import uuiddoc_id str(uuid.uuid4())coll.add(ids[doc_id],embeddingsvecs,documents[text],metadatas[{“source”: source}])return f文档入库成功来源{source}检索工具嵌套embedding查询Chromaasync def rag_search(query: str, top_k: int):q_vec await text_embedding([query])res coll.query(query_embeddingsq_vec,n_resultstop_k)docs res[“documents”][0]if not docs:return “知识库未匹配到相关内容”return “\n”.join([f文档片段{d} for d in docs]) 统一工具网关 class ToolGateway:definit(self):self.tool_registry {“calculator”: {“model”: CalcArgs, “func”: calculator},“text_embedding”: {“model”: EmbeddingArgs, “func”: text_embedding},“vector_add”: {“model”: VectorAddArgs, “func”: vector_add},“rag_search”: {“model”: RagSearchArgs, “func”: rag_search}}