一、今日总学习目标全天8h分配时长拆分理论学习2.5h分文件代码编写、单元调试4h面试题复盘、流程梳理背诵1.5h学习目标清晰认知单一体Agent的短板掌握中控调度垂直子Agent分治设计思想独立分层实现全套底层基础设施每模块解耦、可单独复用实现消息总线完成多Agent并行/串行任务调度、结果统一存储搭建完整业务链路会话记忆→RAG知识库检索→工具函数调用→ReAct反思校验→多智能体汇总输出掌握工程分层规范学会拆分项目文件提升代码可维护性二、核心理论教学笔记单一体Agent的致命缺陷提示词臃肿计算、检索、规划、记忆全部塞入同一个系统prompt模型极易混淆能力边界推理不稳定复杂多步骤任务容易漏执行步骤、选错工具、大量生成幻觉上下文持续膨胀无分层隔离长对话快速Token超限出现对话失忆维护成本极高新增工具/知识库需要大面积修改主Agent代码耦合严重多智能体分层架构标准工业分层四层完整结构自上而下接口层main.pyFastAPI提供SSE流式、多智能体对话接口接收用户请求调度总线层multi_agent.py消息总线统一存储各子Agent执行结果通过flow_id隔离单次工作流Dispatcher中控调度Agent解析用户需求自动拆分任务清单ReAct反思模块循环校验当前信息是否充足不足自动补充检索/计算垂直子AgentRetrieve检索Agent、Calc计算Agent职责单一互不干扰基础能力层LLM客户端llm_client.py兼容OpenAI标准同步/流式输出、结构化JSON强解析RAG向量检索rag_store.py文本重叠分块、简易余弦相似度向量检索工具函数tool_func.py标准化计算器工具Pydantic参数校验持久记忆层memory_store.py双层压缩会话记忆滑动窗口摘要压缩多会话隔离3. 任务执行两种模式并行执行无依赖任务同时运行同时检索知识库数学计算大幅提升响应速度串行执行存在数据依赖时分步执行必须先拿到检索结果再整合输出回答4. 全链路标准执行流程用户输入提问 → 接口接收读取独立会话历史自动执行Token预估、双层压缩裁剪中控Dispatcher拆解任务生成检索/计算任务列表消息总线分发任务多子Agent并行执行收集所有工具/检索结果存入总线ReAct反思判断信息是否充足不足自动补充多轮检索汇总全部结果交给大模型生成通顺完整回答本轮问答持久存入会话内存自动裁剪防止Token爆炸5. 各模块核心能力简介模块文件 核心功能llm_client.py 异步并发LLM请求、SSE流式分片解析、强制结构化JSON输出、失败重试rag_store.py 文本重叠分块、简易字符向量、余弦相似度检索、知识库批量入库memory_store.py 多会话隔离、Token估算、滑动窗口裁剪、历史对话摘要压缩tool_func.py 标准化计算器工具、Pydantic参数校验、运算异常捕获multi_agent.py 消息总线、中控任务拆解、检索/计算子Agent、ReAct反思循环main.py FastAPI服务、SSE流式接口、多智能体对话业务接口三、今日开发难点 落地解决方案难点1整套底层组件全部重新实现容易遗漏流式、JSON校验等细节解决方案模块化分步开发写完一个文件单独测试运行全部单元调试通过后再串联整体流程。难点2多Agent并行执行任务结果错乱、工作流数据互相污染解决方案每个对话流程分配唯一flow_id消息总线以flow_id为分区存储结果不同对话数据完全隔离。难点3模型拆解任务时JSON格式错乱调度器无法分发任务解决方案任务规划强制temperature0消除随机性Prompt严格约束输出格式正则提取JSON片段Pydantic模型二次校验解析失败自动重试一次。难点4多轮长对话持续Token溢出出现上下文截断、失忆解决方案双层压缩策略第一层滑动窗口永久保留system人设仅留存最近N轮原始对话第二层Token仍超限时将早期历史生成精简摘要替换原始长对话难点5ReAct无限循环重复调用工具浪费API额度解决方案设置最大反思轮次默认3轮达到上限强制停止工具调用基于已有信息汇总回答。四、分文件完整项目代码项目目录day8_multi_agent/├── llm_client.py # 异步LLM、SSE流式、结构化输出├── rag_store.py # 文本分块、简易余弦向量RAG检索├── memory_store.py # 分层会话记忆、双层压缩逻辑├── tool_func.py # 计算器工具、参数校验├── multi_agent.py # 消息总线、中控调度、ReAct、子Agent└── main.py # FastAPI接口入口依赖安装命令pip install aiohttp pydantic fastapi uvicorn numpyllm_client.pyimport asyncioimport aiohttpimport reimport jsonfrom typing import List, Dict, AsyncGeneratorfrom pydantic import BaseModel模型密钥配置LLM_CONFIG {“qwen-turbo”: {“base_url”: “https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions”,“api_key”: “你的通义千问API Key”}}结构化输出通用父类class LLMBaseSchema(BaseModel):passclass AsyncLLM:definit(self, model_name: str “qwen-turbo”):self.conf LLM_CONFIG[model_name]self.semaphore asyncio.Semaphore(5)self.timeout aiohttp.ClientTimeout(total60)# SSE流式分片解析缓冲区 async def _stream_parse(self, resp) - AsyncGenerator[str, None]: buffer async for chunk in resp.content.iter_chunked(1024): buffer chunk.decode(utf-8) while data: in buffer: idx buffer.find(data:) end_idx buffer.find(\n\n, idx) if end_idx -1: break block buffer[idx5:end_idx].strip() buffer buffer[end_idx2:] if block [DONE]: return try: data json.loads(block) delta data[choices][0][delta].get(content, ) if delta: yield delta except Exception: continue # 一次性同步完整返回 async def chat_sync(self, messages: List[Dict], temperature0.1) - str: payload { model: qwen-turbo, messages: messages, temperature: temperature, stream: False } headers { Authorization: fBearer {self.conf[api_key]}, Content-Type: application/json } async with self.semaphore: async with aiohttp.ClientSession(timeoutself.timeout) as session: async with session.post(self.conf[base_url], jsonpayload, headersheaders) as resp: res await resp.json() return res[choices][0][message][content] # 流式打字机输出 async def chat_stream(self, messages: List[Dict], temperature0.1): payload { model: qwen-turbo, messages: messages, temperature: temperature, stream: True } headers { Authorization: fBearer {self.conf[api_key]}, Content-Type: application/json } async with self.semaphore: async with aiohttp.ClientSession(timeoutself.timeout) as session: async with session.post(self.conf[base_url], jsonpayload, headersheaders) as resp: async for text in self._stream_parse(resp): yield text # 强制返回标准JSON带重试兜底 async def chat_struct(self, messages: List[Dict], schema: type[LLMBaseSchema]) - LLMBaseSchema: extend_prompt f仅输出标准JSON禁止任何解释、markdownJSON规范{schema.model_json_schema()} new_msg messages.copy() new_msg[-1][content] extend_prompt raw await self.chat_sync(new_msg, temperature0.0) match re.search(r\{.*\}, raw, re.S) if not match: raw await self.chat_sync(new_msg, 0.0) match re.search(r\{.*\}, raw, re.S) return schema.model_validate_json(match.group())全局单例客户端llm_client AsyncLLM()2. rag_store.pyimport numpy as npfrom typing import List, Dict文本重叠分片def split_chunk(text: str, chunk_size400, overlap80) - List[str]:chunks []start 0text_len len(text)while start text_len:end min(start chunk_size, text_len)chunks.append(text[start:end].strip())start chunk_size - overlapreturn chunks简易文本向量化def text_to_vec(text: str) - np.ndarray:vec np.array([ord© for c in text[:256]], dtypenp.float32)vec np.pad(vec, (0, 256 - len(vec)), mode“constant”)norm np.linalg.norm(vec)if norm 0:vec vec / normreturn vec余弦相似度计算def cosine_similarity(v1: np.ndarray, v2: np.ndarray) - float:return float(np.dot(v1, v2))简易内存RAG库class SimpleRAGStore:definit(self):self.doc_meta: List[Dict] []self.vec_cache: List[np.ndarray] []# 批量入库文档 async def add_document(self, text: str, source知识库): chunks split_chunk(text) for chunk in chunks: vec text_to_vec(chunk) self.vec_cache.append(vec) self.doc_meta.append({text: chunk, source: source}) # 相似度检索过滤低相关片段 async def search(self, query: str, top_k3, threshold0.6) - List[Dict]: q_vec text_to_vec(query) score_list [] for idx, vec in enumerate(self.vec_cache): score cosine_similarity(q_vec, vec) score_list.append((score, idx)) score_list.sort(reverseTrue) result [] for score, idx in score_list[:top_k]: if score threshold: result.append(self.doc_meta[idx]) return result全局向量库实例rag_store SimpleRAGStore()预加载测试知识库import asyncioasyncio.run(rag_store.add_document(“多智能体使用中控调度拆分任务RAG检索私有知识库Function Calling执行数学工具计算”))3. memory_store.pyfrom typing import List, Dictfrom llm_client import llm_clientclass ChatMemory:definit(self):self.session_map: Dict[str, List[Dict]] {}self.token_threshold 1800self.keep_raw_round 3# 简易Token估算 def estimate_token(self, msg_list: List[Dict]) - int: total 0 for msg in msg_list: total len(msg.get(content, )) * 2 return total # 滑动窗口裁剪保留system与最近N轮对话 def slide_trim(self, msg_list: List[Dict]) - List[Dict]: sys_msg None other_msg [] for m in msg_list: if m[role] system: sys_msg m else: other_msg.append(m) new_other other_msg[-self.keep_raw_round:] if sys_msg: return [sys_msg] new_other return new_other # 旧对话生成摘要压缩 async def compress_summary(self, msg_list: List[Dict]) - List[Dict]: sys_msg None history [] for m in msg_list: if m[role] system: sys_msg m else: history.append(m) if len(history) self.keep_raw_round: return msg_list old_part history[:-self.keep_raw_round] recent_part history[-self.keep_raw_round:] old_text \n.join([f{m[role]}:{m[content]} for m in old_part]) sum_prompt [{role: user, content: f精简对话摘要保留数字与关键业务信息{old_text}}] summary await llm_client.chat_sync(sum_prompt, temperature0.0) new_msg [] if sys_msg: new_msg.append(sys_msg) new_msg.append({role: system, content: f历史对话摘要{summary}}) new_msg.extend(recent_part) return new_msg # 自动双层压缩入口 async def auto_compress(self, msg_list: List[Dict]) - List[Dict]: if self.estimate_token(msg_list) self.token_threshold: return msg_list trim_res self.slide_trim(msg_list) if self.estimate_token(trim_res) self.token_threshold: return trim_res return await self.compress_summary(msg_list) # 读取会话历史 def load_history(self, session_id: str) - List[Dict]: return self.session_map.get(session_id, []) # 追加单条对话 def append_msg(self, session_id: str, role: str, content: str): if session_id not in self.session_map: self.session_map[session_id] [] self.session_map.append({role: role, content: content})全局内存会话实例memory ChatMemory()4. tool_func.pyfrom pydantic import BaseModel, Field计算器入参约束模型class CalcParams(BaseModel):num1: float Field(description“第一个运算数字”)num2: float Field(description“第二个运算数字”)op: str Field(description“运算符仅支持 - * /”)异步计算工具async def calculator_tool(params: CalcParams) - str:try:match params.op:case “”:res params.num1 params.num2case “-”:res params.num1 - params.num2case “*”:res params.num1 * params.num2case “/”:if params.num2 0:return “工具错误除数不能为0”res params.num1 / params.num2case _:return f不支持运算符{params.op}return f计算结果{params.num1}{params.op}{params.num2}{res}except Exception as e:return f计算异常{str(e)}5. multi_agent.pyimport refrom typing import List, Dictfrom pydantic import BaseModel, Fieldfrom llm_client import llm_client, LLMBaseSchemafrom rag_store import rag_storefrom memory_store import memoryfrom tool_func import calculator_tool, CalcParams单任务结构class TaskItem(BaseModel):agent_name: str Field(description“子Agent名称 retrieve/calc”)task_content: str Field(description“任务执行入参”)任务列表Schemaclass TaskPlan(LLMBaseSchema):task_list: List[TaskItem]消息总线隔离工作流存储各子Agent输出class MessageBus:definit(self):self.flow_storage: Dict[str, Dict[str, str]] {}def save_result(self, flow_id: str, agent: str, content: str): if flow_id not in self.flow_storage: self.flow_storage[flow_id] {} self.flow_storage[flow_id][agent] content def get_all_results(self, flow_id: str) - Dict[str, str]: return self.flow_storage.get(flow_id, {})bus MessageBus()知识库检索专用子Agentclass RetrieveAgent:async def run(self, query: str) - str:docs await rag_store.search(query)if not docs:return “知识库未查询到相关内容”text “\n”.join([item[“text”] for item in docs])return f【知识库检索】{text}数学计算专用子Agentclass CalcAgent:async def run(self, expr: str) - str:prompt [{“role”: “user”, “content”: f提取算式数字与运算符仅输出JSON{{num1,num2,op}}算式{expr}“}]raw await llm_client.chat_sync(prompt, temperature0.0)match re.search(r”{.*}, raw, re.S)params CalcParams.model_validate_json(match.group())return await calculator_tool(params)ReAct反思判断信息是否充足async def reflect_judge(question: str, info: Dict[str, str]) - bool:info_text “\n”.join([f{k}:{v} for k, v in info.items()])prompt f根据已有信息判断是否足够回答用户问题仅输出true/false。问题{question} 已有资料{info_text}res await llm_client.chat_sync([{“role”: “user”, “content”: prompt}], 0.0)return “true” in res.lower()中控调度核心Agentclass DispatcherAgent:# 拆解用户问题生成任务清单async def parse_task_plan(self, user_query: str) - TaskPlan:prompt_text f“”拆解用户问题生成任务仅输出JSONtask_list可选agent_nameretrieve/calc用户问题{user_query}“”msg [{“role”: “user”, “content”: prompt_text}]return await llm_client.chat_struct(msg, TaskPlan)# 完整工作流执行入口 async def run_workflow(self, session_id: str, user_query: str): flow_id fflow_{session_id} # 读取历史并自动压缩 history memory.load_history(session_id) base_msg [{role: system, content: 结合知识库与计算结果完整回答用户问题}] history base_msg.append({role: user, content: user_query}) compress_msg await memory.auto_compress(base_msg) # 第一步并行执行所有拆解任务 plan await self.parse_task_plan(user_query) retrieve_agent RetrieveAgent() calc_agent CalcAgent() task_coros [] task_meta [] for task in plan.task_list: task_meta.append(task) if task.agent_name retrieve: task_coros.append(retrieve_agent.run(task.task_content)) elif task.agent_name calc: task_coros.append(calc_agent.run(task.task_content)) task_outputs await asyncio.gather(*task_coros) for idx, t in enumerate(task_meta): bus.save_result(flow_id, t.agent_name, task_outputs[idx]) # 第二步ReAct反思循环最多3轮补充检索 max_loop 3 loop_cnt 1 all_info bus.get_all_results(flow_id) enough await reflect_judge(user_query, all_info) while not enough and loop_cnt max_loop: new_plan await self.parse_task_plan(f补充检索信息{user_query}) new_coros [] new_meta [] for t in new_plan.task_list: if t.agent_name retrieve: new_meta.append(t) new_coros.append(retrieve_agent.run(t.task_content)) new_out await asyncio.gather(*new_coros) for idx, t in enumerate(new_meta): bus.save_result(f{flow}_loop{loop_cnt}, t.agent_name, new_out[idx]) all_info bus.get_all_results(flow_id) enough await reflect_judge(user_query, all_info) loop_cnt 1 # 第三步汇总全部信息生成最终回答 concat_info \n.join([f{k}:{v} for k, v in all_info.items()]) final_prompt compress_msg [{role: user, content: f结合下面资料完整回答{concat_info}用户问题{user_query}}] final_ans await llm_client.chat_sync(final_prompt, temperature0.1) # 保存本轮对话至内存 memory.append_msg(session_id, user, user_query) memory.append_msg(session_id, assistant, final_ans) return { flow_id: flow_id, task_info: all_info, answer: final_ans }全局调度单例dispatcher DispatcherAgent()6. main.pyfrom fastapi import FastAPI, Queryfrom fastapi.responses import StreamingResponsefrom llm_client import llm_clientfrom multi_agent import dispatcherapp FastAPI(title“Day8 分文件多智能体全链路项目”)SSE流式生成器async def stream_generator(prompt: str):yield “data: 开始生成\n\n”async for chunk in llm_client.chat_stream([{“role”: “user”, “content”: prompt}]):yield fdata: {chunk}\n\nyield “data: [DONE]\n\n”基础流式对话接口app.get(“/chat/stream”)async def stream_chat(prompt: str):return StreamingResponse(stream_generator(prompt), media_type“text/event-stream”)完整多智能体业务接口app.get(“/agent/chat”)async def agent_chat(session_id: str Query(…, description“会话唯一标识”),prompt: str Query(…, description“用户提问内容”)):return await dispatcher.run_workflow(session_id, prompt)ifname “main”:import uvicornuvicorn.run(“main:app”, reloadTrue)五、今日实操练习任务新建独立文件夹按目录创建6个py文件填入对应代码修改LLM配置中的API密钥启动服务uvicorn main:app --reload访问127.0.0.1:8000/docs基础流式测试调用/chat/stream验证SSE打字机效果、分片解析无乱码复合多智能体测试接口/agent/chat传入任意session_id提问计算(10020)*6介绍多智能体RAG架构连续多轮同一session提问验证滑动窗口摘要双层压缩逻辑修改rag_store知识库文本验证检索结果同步更新构造除零计算测试工具异常捕获、错误信息正常返回构造信息不足的提问观察ReAct自动补充检索逻辑六、配套高频面试题基础问答单一体Agent和多智能体分治架构优缺点、适用场景消息总线在多Agent系统中的作用是什么flow_id设计意义并行任务、串行任务分别适用什么业务场景分层记忆两种压缩方案滑动窗口/摘要区别与取舍ReAct反思循环的实现逻辑为什么要限制最大循环轮次工程实操题如何保证模型稳定输出标准任务JSON防止调度解析失败SSE流式分片出现截断、残缺如何解决多轮对话持续Token超限双层压缩实现思路多Agent并行执行如何统一收集、隔离各工作流结果拓展思考题内存会话如何升级Redis分布式持久化支持多服务实例简易内存向量库如何替换Chroma/Milvus实现海量持久知识库如何给多智能体增加并发限流、工具熔断、指数退避重试七、面试题标准答案基础问答单Agent开发简单、代码集中缺点提示词臃肿、推理不稳定、维护困难适合简单闲聊。多智能体各模块职责单一、推理稳定、可独立迭代扩容适合多步骤工具知识库复杂业务。统一存储每个流程下所有子Agent执行结果flow_id区分单次对话工作流避免不同用户、不同轮次数据互相污染。无依赖查询/计算使用并行提升响应速度前置结果作为输入的任务使用串行。滑动窗口性能高直接删除早期历史但会丢失信息摘要通过LLM压缩旧对话保留关键数据额外消耗一次模型调用。长对话采用双层兜底策略。将当前全部检索、计算结果交给模型结构化判断信息是否充足限制最大轮次防止无限循环调用工具节省API额度。工程实操题temperature设为0消除随机性Prompt强制JSON规范正则提取大括号内容Pydantic模型校验解析失败自动重新规划一次。设置缓冲区拼接不完整分片仅处理完整data:行捕获单条JSON解析异常不中断整条流式输出。先滑动窗口保留system与最近N轮原始对话Token仍超标时将早期历史生成摘要替换原始长文本永久保留人设信息。每条对话分配唯一flow_id消息总线以flow_id作为key分区存储调度完成后统一读取对应分区结果。拓展思考题将内存字典替换aioredissession_id作为Redis键列表存储对话消息设置过期自动清理会话。上层检索接口不变底层向量计算、入库逻辑替换为Chroma持久化集合做到业务代码无感知切换。全局异步信号量控制最大并行任务数工具连续多次失败触发熔断捕获429限流异常实现指数退避重试。八、学习总结Day8完整搭建分层多智能体项目所有底层组件从零独立开发无任何前序代码依赖模块化拆分清晰适合分步练习复盘。