今日总学习目标实现全工具化架构向量入库、向量化、检索、计算均为可调用工具基于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 Embedding LLM_BASE_URLhttps://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions LLM_EMBED_URLhttps://dashscope.aliyuncs.com/compatible-mode/v1/embeddings LLM_API_KEY你的API_KEY # Redis REDIS_URLredis://127.0.0.1:6379 # 限流熔断 TOKEN_BUCKET_CAP12 TOKEN_RATE3 MAX_FAIL_TIMES3 COOLDOWN_SECONDS10 # 开关 ENABLE_SAFE_FILTERtrue ENABLE_LOGtrue # Chroma持久化路径 CHROMA_PERSIST_PATH./chroma_kb CHROMA_COLLECTION_NAMEagent_kb2 middleware.pyimport asyncio import time import json import os import uuid from dotenv import load_dotenv load_dotenv() # 令牌桶限流 class TokenBucket: def __init__(self, cap: int, rate: float): self.capacity cap self.rate rate self.token_num cap self.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 False global_bucket TokenBucket(int(os.getenv(TOKEN_BUCKET_CAP)), float(os.getenv(TOKEN_RATE))) # 熔断降级 class CircuitBreaker: def __init__(self, max_fail: int, cool_sec: int): self.max_fail max_fail self.cool cool_sec self.fail_count 0 self.state closed self.open_start 0 async 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.time breaker_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 1 for _ 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 e await asyncio.sleep(delay) delay min(delay * 2, 8) return await call_func() # 日志埋点 class AgentLog: def __init__(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 re import os SENSITIVE_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 text def desensitize(text: str) - str: for pat in SENSITIVE_RULES: text pat.sub(******, text) return text def 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 True if need_auth admin and user_role admin: return True return False async 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_text