今日总学习目标

1. 实现全工具化架构：向量入库、向量化、检索、计算均为可调用工具
2. 基于Chroma本地持久向量库完成知识库增删查，文档入库也走工具调用
3. 掌握工具嵌套执行逻辑，上层检索工具自动调用底层Embedding工具
4. 统一中间件体系对四类工具全覆盖，分层超时隔离Embedding/LLM请求
5. 独立运行整套ReAct Agent，无任何前置课程代码依赖
   每日时长分配（全天8h）

• 理论笔记阅读：2.5h
• 分层代码编写调试：4h
• 复盘面试背诵：1.5h

一、核心理论教学笔记

1 全工具化分层设计

底层基础工具（依赖Embedding接口）

text_embedding：批量文本转语义向量，所有向量操作唯一底层依赖

业务向量工具（嵌套底层embedding）

1. vector_add：新增知识库文档，自动向量化存入Chroma
2. rag_search：检索知识库，自动生成查询向量，Chroma相似度召回

数值工具

calculator：四则数学运算

所有工具全部注册到统一网关，权限、限流、熔断、日志一套逻辑复用

2 Chroma向量库优势（对比内存/FAISS）

1. 磁盘持久化：程序重启向量与文档不丢失
2. 原生元数据支持：可标记文档来源、分类，检索过滤
3. 开箱即用，无需手动维护向量id映射字典
4. 内置相似度排序，支持MMR去重检索
5. 同步API简单适配异步业务

3 完整工具化执行链路

用户提问

1. 接口生成全局TraceID，执行输入安全校验、令牌桶限流
2. 读取Redis会话记忆，自动滑动窗口+摘要压缩
3. ReAct循环推理：模型自主选择工具
   • 数学题 → calculator
   • 新增资料 → vector_add（内部调用text_embedding存入Chroma）
   • 查询知识 → rag_search（内部调用text_embedding生成查询向量，Chroma召回）
4. 网关统一拦截：权限校验 → 熔断判断 → 指数退避重试 → 独立超时执行
5. 工具结果回填对话上下文，反思判断信息是否充足
6. 汇总全部观测结果，LLM生成脱敏最终回答
7. 对话持久存入Redis，全链路日志落地

4 生产中间件全覆盖规则

• LLM全局60s超时；Embedding独立15s超时
• 令牌桶统一管控所有工具请求，削峰防429限流
• 单工具连续失败触发熔断，冷却后试探恢复
• 访客仅可闲聊；普通用户可用计算、检索；管理员可批量入库文档
• 每条工具调用携带TraceID写入日志，记录耗时、参数、异常

二、今日学习重点

1. 将文档入库、向量化、检索全部封装为可调用工具，实现纯工具驱动Agent
2. Chroma持久向量库接入项目，上层工具无感知读写向量库
3. 实现工具嵌套调用，上层检索/入库工具自动调用Embedding底层工具
4. 安全、限流、熔断、日志中间件对四类工具统一生效
5. 独立可运行完整项目，无Day1~Day8代码依赖

三、今日难点 & 解决方案

难点1：向量相关逻辑散落在代码各处，无法监控限流

解决方案：全部封装标准Function，统一网关调度，所有向量操作都经过中间件拦截

难点2：程序重启知识库丢失

解决方案：使用Chroma持久客户端，向量数据落地本地文件夹

难点3：Embedding高频调用频繁触发平台限流

解决方案：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       # LLM+Embedding异步客户端
├── tool_gateway.py     # 全量工具注册（Chroma向量工具+计算器）
├── memory_store.py     # Redis分层持久记忆
├── agent_core.py       # ReAct主智能体
└── main.py             # FastAPI入口

1 .env 配置文件

# LLM & Embedding
LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions
LLM_EMBED_URL=https://dashscope.aliyuncs.com/compatible-mode/v1/embeddings
LLM_API_KEY=你的API_KEY
# Redis
REDIS_URL=redis://127.0.0.1:6379
# 限流熔断
TOKEN_BUCKET_CAP=12
TOKEN_RATE=3
MAX_FAIL_TIMES=3
COOLDOWN_SECONDS=10
# 开关
ENABLE_SAFE_FILTER=true
ENABLE_LOG=true
# Chroma持久化路径
CHROMA_PERSIST_PATH=./chroma_kb
CHROMA_COLLECTION_NAME=agent_kb

2 middleware.py

import 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_retry=3):
    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", encoding="utf-8") as f:
            f.write(json.dumps(log_data, ensure_ascii=False) + "\n")

log_client = AgentLog()

def create_trace_id() -> str:
    return str(uuid.uuid4())

3 security.py

import re
import os

SENSITIVE_RULES = [
    re.compile(r"1[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_len=800) -> tuple[bool, str]:
    if len(raw_input) > max_len:
        return False, "输入过长，请精简提问"
    safe_text = escape_inject(raw_input)
    return True, safe_text