AI Agent 30天速成｜Day10 笔记

修订说明基于Day9全工具化架构迭代保留Chroma向量库、统一工具网关、Redis记忆、熔断限流体系新增工具链批量编排支持顺序/并行多工具组合任务无需模型多轮ReAct循环实现异步后台任务队列长耗时知识库批量入库、大规模向量离线计算不阻塞HTTP接口增加全局接口Token鉴权、分布式会话锁适配多实例分布式部署场景新增Agent调用成本统计模块统计LLM/Embedding调用次数、Token消耗、计费估算今日总学习目标全天8h分配理论学习2.5h分层代码编写调试4h面试复盘背诵1.5h掌握工具编排语法定义串行/并行工具流水线实现一次性多工具批量执行基于Redis List实现轻量异步任务队列处理长耗时离线向量任务实现全局API鉴权、用户资源配额LLM/Embedding每日调用上限分布式锁解决多实例Redis会话并发读写错乱问题新增调用成本统计模块全链路采集Token消耗与接口调用量整合流水线编排、异步任务、鉴权配额、分布式锁升级生产级Agent服务一、核心理论教学笔记1 工具流水线编排Tool Pipeline1.1 单轮ReAct存在短板ReAct每轮只能执行单个工具多步骤任务需要多次LLM思考消耗大量Token与接口次数固定流程先检索再计算、批量入库重复循环效率极低。1.2 两种编排模式串行流水线依赖型上一个工具输出作为下一个工具输入顺序执行示例rag_search → calculator先查公式再代入计算并行流水线无依赖多个工具同时并发执行结果统一汇总示例同时执行多条rag_search多维度知识库检索1.3 流水线结构化定义通过Pydantic定义流水线JSON结构模型可一次性生成多工具任务网关自动解析批量调度仅消耗单次LLM思考。流水线内置变量传递前序工具输出可作为后续工具入参无需重复拼接Prompt。2 Redis异步任务队列轻量版无需RabbitMQ2.1 适用场景批量文档入库、海量文本Embedding、大规模知识库更新等耗时操作同步HTTP接口会超时。2.2 队列执行流程前端提交批量任务 → 接口校验权限配额 → 写入Redis任务队列返回task_id后台独立协程循环消费队列执行向量入库/Embedding批量工具任务进度、执行结果持久存入Redis Hash用户通过task_id轮询查询状态失败任务自动重试2次最终失败写入死信队列人工排查3 分布式配套能力3.1 全局API鉴权所有接口请求携带access_token服务内存维护token-角色-配额映射区分普通用户/管理员每日调用上限超量直接拦截。3.2 分布式会话锁多服务实例同时收到同一用户多轮提问并发读写Redis会话会造成消息错乱基于Redis SETNX实现会话互斥锁单会话同一时间仅允许一条对话执行。3.3 调用成本统计每条LLM、Embedding调用采集输入/输出Token存入Redis计数器按厂商单价估算消耗费用支持按用户、会话、日期维度统计用量。4 Day9→Day10 完整升级链路用户HTTP请求携带token鉴权 → 校验用户每日调用配额→ 获取分布式会话锁防止并发乱序→ 读取Redis分层记忆自动摘要/滑动窗口压缩→ 模型二选一简单任务单工具ReAct循环执行固定多步骤任务生成流水线编排网关批量并行/串行执行工具→ 短任务同步返回长批量任务推送Redis异步队列返回task_id→ 所有工具执行埋点限流、熔断、日志、Token消耗统计→ 工具结果回填上下文LLM生成回答脱敏处理→ 对话持久Redis释放分布式会话锁→ 后台协程持续消费异步队列处理离线向量批量任务5 生产级配额与限流规则访客仅基础闲聊禁用所有工具每日LLM 10次上限普通用户calculator/rag_search可用禁用vector_addLLM 200次/天Embedding 500次/天管理员全工具开放无每日硬上限仅全局令牌桶流量控制二、今日学习重点实现ToolPipeline流水线结构化定义网关自动解析串行/并行批量工具执行搭建Redis轻量异步任务队列实现长耗时向量任务后台异步处理开发全局Token鉴权、用户每日调用配额拦截逻辑Redis分布式锁解决多实例会话并发读写冲突全链路Token采集、用量统计、成本估算模块开发兼容原有Day9所有能力无破坏性改动平滑升级三、今日难点 解决方案难点1 流水线工具之间参数传递复杂前序结果无法自动传给下一个工具解决方案流水线结构定义输出变量key网关缓存每一步工具结果后续工具可通过${变量名}动态填充入参自动替换文本。难点2 多实例部署同一用户并发对话导致历史消息错乱解决方案执行对话前后加分布式Redis锁锁过期时间30s对话流程未完成时同会话新请求直接返回“对话处理中请稍后”。难点3 批量向量入库耗时过长同步接口60s超时解决方案耗时任务剥离至Redis异步队列接口仅生成任务ID立即返回后台协程单独消费前端轮询任务进度。难点4 用户无感知超额调用LLM/Embedding产生高额费用解决方案按用户角色设置每日调用配额Redis计数器实时累加达到上限直接拦截同时每日零点自动重置计数。难点5 异步任务大量失败堆积无法感知解决方案区分正常队列与死信队列失败重试2次仍异常的任务转入死信日志记录全部失败task_id提供查询接口查看失败详情。四、完整项目代码基于Day9扩展新增文件修改原有文件依赖新增pip install python-multipart项目目录新增文件day10_agent/├── .env # 新增鉴权、队列、配额配置├── middleware.py # 新增分布式锁、用量统计逻辑├── security.py # 新增token鉴权、用户配额校验├── pipeline.py # 【今日新增】工具流水线编排定义├── async_task.py # 【今日新增】Redis异步任务队列消费├── llm_client.py├── tool_gateway.py # 适配流水线批量工具执行├── memory_store.py├── agent_core.py # 支持流水线ReAct双模式└── main.py # 新增鉴权中间件、任务查询接口1 .env 新增配置项原有LLM/Redis/Chroma配置保留API鉴权ADMIN_TOKENadmin123456USER_TOKENuser654321GUEST_TOKENguest0000每日调用配额ADMIN_QUOTA_LLM9999ADMIN_QUOTA_EMB9999USER_QUOTA_LLM200USER_QUOTA_EMB500GUEST_QUOTA_LLM10GUEST_QUOTA_EMB0异步任务队列TASK_QUEUE_KEYagent:task:queueDEAD_QUEUE_KEYagent:task:deadTASK_EXPIRE86400分布式锁LOCK_PREFIXagent:session:lockLOCK_EXPIRE30Token计费单价元/千tokenPRICE_INPUT0.5PRICE_OUTPUT1.0PRICE_EMB0.22 pipeline.py全新文件工具流水线编排from pydantic import BaseModel, Fieldfrom typing import List, Dict, Optional, Unionfrom tool_gateway import gatewayclass PipelineStep(BaseModel):task_id: str Field(description“流水线内步骤唯一标识”)tool_name: str Field(description“执行工具名称”)args: Dict Field(description“工具入参支持${变量名}引用上一步输出”)depends_on: List[str] Field(default[], description“依赖步骤id空则并行执行”)output_var: str Field(description“本步骤输出存储变量名供后续步骤引用”)class ToolPipeline(BaseModel):run_mode: str Field(description“serial串行 / parallel并行”)steps: List[PipelineStep] Field(description“流水线步骤列表”)class PipelineRunner:definit(self):self.var_cache {} # 存储各步骤输出变量# 替换参数内${xxx}变量 def replace_var(self, raw_arg: Union[str, Dict]) - Union[str, Dict]: if isinstance(raw_arg, str): for k, v in self.var_cache.items(): raw_arg raw_arg.replace(f${{{k}}}, str(v)) return raw_arg if isinstance(raw_arg, dict): new_dict {} for k, val in raw_arg.items(): new_dict[k] self.replace_var(val) return new_dict return raw_arg async def run_single_step(self, step: PipelineStep, trace_id: str, user_role: str): parsed_args self.replace_var(step.args) res await gateway.run_tool(step.tool_name, parsed_args, trace_id, user_role) self.var_cache[step.output_var] res return {step.task_id: res} async def run_pipeline(self, pipeline: ToolPipeline, trace_id: str, user_role: str): all_results {} step_map {s.task_id: s for s in pipeline.steps} finished set() while len(finished) len(step_map): run_coros [] run_steps [] for sid, step in step_map.items(): if sid in finished: continue # 判断依赖是否全部完成 dep_ok all(d in finished for d in step.depends_on) if dep_ok: run_coros.append(self.run_single_step(step, trace_id, user_role)) run_steps.append(sid) if not run_coros: break batch_res await asyncio.gather(*run_coros) for item in batch_res: all_results.update(item) for s in run_steps: finished.add(s) return { pipeline_var: self.var_cache, step_result: all_results }pipeline_runner PipelineRunner()3 async_task.py全新文件异步任务队列import asyncioimport jsonimport uuidimport osfrom dotenv import load_dotenvfrom memory_store import memoryfrom pipeline import ToolPipeline, pipeline_runnerload_dotenv()TASK_QUEUE os.getenv(“TASK_QUEUE_KEY”)DEAD_QUEUE os.getenv(“DEAD_QUEUE_KEY”)TASK_EXPIRE int(os.getenv(“TASK_EXPIRE”))class AsyncTaskCenter:definit(self):self.redis memory.redisdef gen_task_id(self): return str(uuid.uuid4()) async def submit_task(self, task_type: str, pipeline_dict: dict, trace_id: str, user_role: str): tid self.gen_task_id() task_data json.dumps({ task_id: tid, task_type: task_type, pipeline: pipeline_dict, trace_id: trace_id, user_role: user_role, status: pending, retry_cnt: 0 }) # 写入队列 Hash存储详情 await self.redis.lpush(TASK_QUEUE, task_data) await self.redis.setex(ftask:info:{tid}, TASK_EXPIRE, task_data) return tid async def get_task_info(self, task_id: str): raw await self.redis.get(ftask:info:{task_id}) if not raw: return None return json.loads(raw) async def consumer_loop(self): print(异步任务消费协程启动成功) while True: # 阻塞读取队列 raw_task await self.redis.brpop(TASK_QUEUE, timeout2) if not raw_task: continue _, task_str raw_task task json.loads(task_str) tid task[task_id] try: task[status] running await self.redis.setex(ftask:info:{tid}, TASK_EXPIRE, json.dumps(task)) # 执行流水线 pipeline ToolPipeline(**task[pipeline]) res await pipeline_runner.run_pipeline(pipeline, task[trace_id], task[user_role]) task[status] success task[result] json.dumps(res, ensure_asciiFalse) except Exception as e: task[retry_cnt] 1 task[err_msg] str(e) if task[retry_cnt] 2: # 重新入队重试 await self.redis.lpush(TASK_QUEUE, json.dumps(task)) continue task[status] failed # 转入死信队列 await self.redis.rpush(DEAD_QUEUE, json.dumps(task)) await self.redis.setex(ftask:info:{tid}, TASK_EXPIRE, json.dumps(task))task_center AsyncTaskCenter()4 middleware.py 新增分布式锁、用量统计代码在原有代码末尾追加import osLOCK_PREFIX os.getenv(“LOCK_PREFIX”)LOCK_EXPIRE int(os.getenv(“LOCK_EXPIRE”))PRICE_INPUT float(os.getenv(“PRICE_INPUT”))PRICE_OUTPUT float(os.getenv(“PRICE_OUTPUT”))PRICE_EMB float(os.getenv(“PRICE_EMB”))分布式会话锁async def session_lock(redis, session_id: str):key f{LOCK_PREFIX}{session_id}ok await redis.set(key, “locked”, exLOCK_EXPIRE, nxTrue)return bool(ok)async def session_unlock(redis, session_id: str):key f{LOCK_PREFIX}{session_id}await redis.delete(key)Token用量统计class TokenStat:definit(self, redis):self.redis redisasync def add_llm_token(self, user_role: str, input_tok: int, output_tok: int): today time.strftime(%Y%m%d) key_llm_in fstat:{today}:{user_role}:llm_input key_llm_out fstat:{today}:{user_role}:llm_output await self.redis.incrby(key_llm_in, input_tok) await self.redis.incrby(key_llm_out, output_tok) async def add_emb_token(self, user_role: str, tok_num: int): today time.strftime(%Y%m%d) key_emb fstat:{today}:{user_role}:embedding await self.redis.incrby(key_emb, tok_num) async def get_cost(self, user_role: str): today time.strftime(%Y%m%d) in_tok int(await self.redis.get(fstat:{today}:{user_role}:llm_in) or 0) out_tok int(await self.redis.get(fstat:{today}:llm_out) or 0) emb_tok int(await self.redis.get(fstat:{today}:emb) or 0) cost (in_tok / 1000) * PRICE_INPUT (out_tok / 1000) * PRICE_OUTPUT (emb_tok / 1000) * PRICE_EMB return round(cost, 4)stat_client TokenStat(memory.redis)5 security.py 新增Token鉴权配额校验原有代码保留追加下方内容import osTOKEN_ROLE_MAP {os.getenv(“ADMIN_TOKEN”): “admin”,os.getenv(“USER_TOKEN”): “user”,os.getenv(“GUEST_TOKEN”): “guest”}QUOTA_CFG {“admin”: {“llm”: int(os.getenv(“ADMIN_QUOTA_LLM”)),“emb”: int(os.getenv(“ADMIN_QUOTA_EMB”))},“user”: {“llm”: int(os.getenv(“USER_QUOTA_LLM”)),“emb”: int(os.getenv(“USER_QUOTA_EMB”))},“guest”: {“llm”: int(os.getenv(“GUEST_QUOTA_LLM”)),“emb”: int(os.getenv(“GUEST_QUOTA_EMB”))}}Token鉴权def verify_token(token: str) - tuple[bool, str]:if token not in TOKEN_ROLE_MAP:return False, “”return True, TOKEN_ROLE_MAP[token]校验当日调用配额async def check_quota(redis, user_role: str, call_type: str) - bool:today time.strftime(“%Y%m%d”)limit QUOTA_CFG[user_role][call_type]key fquota:{today}:{user_role}:{call_type}current int(await redis.get(key) or 0)if current limit:return Falseawait redis.incr(key)return True6 agent_core.py 改造支持流水线编排原有导入追加from pipeline import ToolPipeline, pipeline_runnerfrom middleware import stat_clientclass ReActAgent:# 原有init、reflect函数不变async def run_chat(self, session_id: str, user_role: str, user_input: str, trace_id: str):history await memory.load_history(session_id)base_msg [{“role”:“system”, “content”:self.system_prompt}] historybase_msg.append({“role”:“user”, “content”:user_input})msg_list await memory.auto_compress(base_msg)# 新增判断生成流水线还是单轮ReActpipeline_prompt “”判断用户问题是否为固定多步骤任务是则输出ToolPipeline JSONrun_mode选serial/parallel简单单任务输出{“is_pipeline”:false}“”pipe_check_msg msg_list [{“role”:“user”, “content”:pipeline_prompt}]pipe_raw await llm_client.chat_sync(pipe_check_msg, temperature0.0)[“choices”][0][“message”][“content”]# 省略JSON解析逻辑区分两种执行分支if “is_pipeline” in pipe_raw and json.loads(pipe_raw)[“is_pipeline”] is False:# 原有ReAct循环逻辑不变执行后统计tokenloop_cnt 0tool_record {}input_tokens sum(len(m[“content”]) for m in msg_list)while loop_cnt self.max_loop:# 原有ReAct代码不变…output_tokens len(final_raw)await stat_client.add_llm_token(user_role, input_tokens, output_tokens)else:# 流水线分支pipe_data json.loads(pipe_raw)pipeline ToolPipeline(**pipe_data)pipeline_res await pipeline_runner.run_pipeline(pipeline, trace_id, user_role)concat_info json.dumps(pipeline_res[“step_result”], ensure_asciiFalse)final_prompt msg_list [{“role”:“user”, “content”:f结合流水线结果回答{concat_info}}]final_raw await llm_client.chat_sync(final_prompt, temperature0.1)[“choices”][0][“message”][“content”]input_tokens sum(len(m[“content”]) for m in final_prompt)output_tokens len(final_raw)await stat_client.add_llm_token(user_role, input_tokens, output_tokens)# 脱敏、持久化逻辑不变final_ans desensitize(final_raw)await memory.append(session_id, “user”, user_input)await memory.append(session_id, “assistant”, final_ans)log_client.write(trace_id, “INFO”, {“final_answer”: final_ans[:300]})return {“trace_id”: trace_id,“tool_record”: tool_record if “tool_record” in locals() else pipeline_res,“answer”: final_ans}react_agent ReActAgent()7 main.py 新增鉴权、任务查询、异步提交接口from fastapi import FastAPI, Query, Header, HTTPExceptionimport asynciofrom agent_core import react_agentfrom memory_store import memoryfrom middleware import global_bucket, create_trace_id, log_client, session_lock, session_unlockfrom security import input_verify, verify_token, check_quotafrom async_task import task_centerapp FastAPI(title“Day10 流水线Agent异步任务分布式锁鉴权配额”)全局启动事件新增异步消费协程app.on_event(“startup”)async def startup():await memory.connect()asyncio.create_task(task_center.consumer_loop())app.on_event(“shutdown”)async def shutdown():await memory.close()通用对话接口增加token鉴权app.get(“/agent/chat”)async def chat_api(Authorization: str Header(…, description“Bearer token”),session_id: str Query(…),user_role: str Query(default“user”),prompt: str Query(…)):# 1 Token鉴权token Authorization.replace(“Bearer “,””)auth_ok, real_role verify_token(token)if not auth_ok:raise HTTPException(status_code401, detail“非法访问令牌”)# 2 配额校验llm_quota_ok await check_quota(memory.redis, real_role, “llm”)if not llm_quota_ok:return {“trace_id”: create_trace_id(), “answer”: “今日LLM调用额度已耗尽请明日再试”}# 3 分布式会话锁lock_success await session_lock(memory.redis, session_id)if not lock_success:return {“trace_id”: create_trace_id(), “answer”: “当前会话正在处理对话请稍后重试”}try:trace_id create_trace_id()log_client.write(trace_id, “INFO”, {“session_id”: session_id, “input”: prompt, “role”: real_role})# 输入安全校验ok, safe_text await input_verify(prompt)if not ok:return {“trace_id”: trace_id, “answer”: safe_text}# 全局限流if not await global_b.get_token():log_client.write(trace_id, “WARN”, {“msg”: “触发全局限流”})return {“trace_id”: trace_id, “answer”: “服务繁忙请稍后重试”}# 主Agent执行result await react_agent.run_chat(session_id, real_role, safe_text, trace_id)return resultfinally:# 释放会话锁await session_unlock(memory.redis, session_id)提交异步批量流水线任务接口app.post(“/agent/task/submit”)async def submit_task(Authorization: str Header(…),session_id: str,task_type: str,pipeline: dict):token Authorization.replace(“Bearer “,””)auth_ok, real_role verify_token(token)if not auth_ok:raise HTTPException(401, “token无效”)# 仅管理员可提交批量入库任务if task_type “batch_embed” and real_role ! “admin”:raise HTTPException(403, “仅管理员可执行批量向量任务”)trace_id create_trace_id()tid await task_center.submit_task(task_type, pipeline, trace_id, real_role)return {“task_id”: tid, “trace_id”: trace_id, “msg”: “任务已进入后台队列”}查询异步任务进度app.get(“/agent/task/query”)async def query_task(task_id: str):info await task_center.get_task_info(task_id)if not info:return {“msg”: “任务不存在或已过期”}return infoifname “main”:import uvicornuvicorn.run(“main.py”, reloadTrue)五、今日实操练习任务配置.env内三类Token分别用admin/user/guest令牌调用接口验证权限隔离与每日配额拦截构造复合问题先检索RAG定义再计算(10020)*5观察模型自动生成串行流水线批量执行工具使用admin账号提交批量文档入库流水线获取task_id轮询/agent/task/query查看后台执行进度同一session_id并发发送两条提问验证分布式锁拦截返回会话繁忙提示连续调用接口耗尽user角色每日LLM配额观察配额拦截提示人为制造批量向量任务异常查看死信队列存储失败任务信息查看日志与Redis统计key验证LLM/Embedding Token消耗与费用估算正常六、配套完整面试题含标准答案基础问答1 Tool流水线和ReAct循环的核心区别各自适用场景标准答案ReAct单次只能调用一个工具多步骤任务需要多轮LLM思考消耗更多Token适合无固定流程、开放式未知问题。Tool流水线一次性定义串行/并行多工具步骤模型仅一次规划网关批量调度适合流程固定、多依赖/多并行查询的标准化任务批量知识库入库、固定计算检索组合优先使用流水线。2 为什么要用Redis实现异步任务不直接同步执行批量向量入库标准答案批量Embedding、大规模文档入库耗时可达数十秒HTTP接口默认超时会断开连接异步队列将耗时操作剥离后台前端立刻获取task_id轮询结果提升接口可用性同时失败任务支持自动重试不阻塞用户对话主线程。3 分布式会话锁解决什么问题实现原理标准答案多服务实例部署时同一用户同时发起多条对话请求并发读写Redis会话列表会造成消息错乱、上下文拼接异常。基于Redis SETNX原子命令加锁会话处理期间持有30s过期锁新同会话请求检测锁存在直接拒绝对话结束主动释放锁保证单会话串行执行。4 用户配额统计基于Redis什么数据结构零点如何重置每日用量标准答案使用Redis String计数器key拼接日期用户角色区分每日用量每日零点定时任务删除当日统计key自动重置次日计数设置过期时间24h自动清理过期历史用量数据。5 流水线变量传递实现逻辑是什么标准答案网关内置变量缓存字典每一步工具执行完成后将输出存入对应output_var解析后续步骤入参时正则匹配${变量名}自动替换为上一步工具输出无需模型重复拼接上下文参数。工程实操题1 流水线步骤存在循环依赖A依赖BB依赖A如何处理