系列终篇。前六篇拆解了工程地基和四个 Agent今天看它们如何被组装成一个完整系统Orchestrator的三种执行模式与懒加载机制、简历审查 模拟面试的求职全链路 Pipeline、以及统一入口unified_chat.py的五类零 Token 规则拦截 LLM 意图路由 SSE 流式分发。读完这篇你会对多智能体系统怎么搭有完整的认识。一、Orchestrator统一请求入口1.1 三种执行模式classExecutionMode(str,Enum):SINGLEsingle# 单 Agent 直达最常见PIPELINEpipeline# 多 Agent 串联如求职全链路CLARIFYclarify# 澄清对话意图不明需追问classAgentRequest(BaseModel):student_id:strtenant_id:strtenant_defaultsession_id:stragent_type:AgentType user_message:strcontext:dict[str,Any]{}pipeline_mode:boolFalsepropertydefthread_id(self)-str:LangGraph Checkpointer 的线程 IDreturnfstudent_{self.student_id}_session_{self.session_id}统一入参AgentRequest 统一出参AgentResponsesuccess/content/structured/fallback_used/error_msg——不管哪个 Agent进出格式都一样上层调用方API 层完全不用关心具体是哪个 Agent 在处理。1.2 懒加载需要时才构建图classOrchestrator:def__init__(self):self._agent_graphs:dict[AgentType,Any]{}def_get_agent_graph(self,agent_type:AgentType)-Any:ifagent_typenotinself._agent_graphs:ifagent_typeAgentType.QA:frombackend.agents.qa.graphimportbuild_qa_graph self._agent_graphs[agent_type]build_qa_graph()elifagent_typeAgentType.EXAM:frombackend.agents.exam.graphimportbuild_exam_graph self._agent_graphs[agent_type]build_exam_graph()# ... resume / interview 同理returnself._agent_graphs[agent_type]懒加载的意义四个 LangGraph 图如果启动时全部构建会白白占用内存和时间尤其 QA 图要初始化检索相关配置。首次用到哪个才import build之后缓存复用。注意 import 放在函数内部——这也是延迟 import模式顺便避免了循环依赖。1.3 单 Agent 执行套三层兜底asyncdef_run_single_agent(self,request:AgentRequest)-AgentResponse:graphself._get_agent_graph(request.agent_type)initial_state{messages:[HumanMessage(contentrequest.user_message)],student_id:request.student_id,tenant_id:request.tenant_id,session_id:request.session_id,**request.context,# 附加上下文平铺进 State}config{configurable:{thread_id:request.thread_id},callbacks:[langfuse_handler],# 可观测性回调metadata:{langfuse_user_id:request.student_id,...},}with_retry(agent_typerequest.agent_type.value)# 三层兜底asyncdef_invoke():returnawaitgraph.ainvoke(initial_state,configconfig)result_stateawait_invoke()# 取最后一条消息作为 AI 回答last_messageresult_state[messages][-1]contentlast_message.textifhasattr(last_message,text)\elsestr(last_message.content)returnAgentResponse(successTrue,agent_typerequest.agent_type,contentcontent,structuredresult_state.get(structured_output),fallback_usedresult_state.get(fallback_used,False))注意**request.context展开——附加上下文如文件路径、简历结果直接平铺进 LangGraph 的 State和图内定义的字段无缝融合。二、Pipeline多 Agent 串联2.1 Pipeline 定义self._pipelines:dict[str,list[AgentType]]{job_preparation:[AgentType.RESUME,AgentType.INTERVIEW],# 求职全链路先审简历再面试}2.2 顺序执行 上下文传递asyncdef_run_pipeline(self,request:AgentRequest)-PipelineResult:pipeline_keyrequest.context.get(pipeline_key,job_preparation)ifpipeline_keynotinself._pipelines:raisePipelineError(f未知 Pipeline 类型:{pipeline_key})agent_sequenceself._pipelines[pipeline_key]resultPipelineResult()current_contextdict(request.context)# 累积上下文foridx,agent_typeinenumerate(agent_sequence):step_requestAgentRequest(student_idrequest.student_id,tenant_idrequest.tenant_id,session_idf{request.session_id}_step{idx},# 避免检查点串台agent_typeagent_type,user_messagerequest.user_message,contextcurrent_context,# 带上累积上下文pipeline_modeFalse)# 单步内不再触发 Pipelinestep_responseawaitself._run_single_agent(step_request)result.steps.append(step_response)ifnotstep_response.success:break# 失败保留已完成的成果终止后续# ★ 上下文传递前序结构化输出注入后序 contextifstep_response.structured:current_context[f{agent_type.value}_result]step_response.structuredifagent_typeAgentType.RESUME:# 简历 → 面试衔接review_idstep_response.structured.get(review_id)ifreview_id:current_context[resume_review_id]review_id scorestep_response.structured.get(weighted_score,0)ifscore60:# ★ 评分门槛简历不合格不进面试breakreturnresultPipeline 的四个工程细节session_id 加_step{idx}后缀——每一步用独立的 checkpoint避免步骤间状态串台失败保留前序成果——第一步简历审查成功、第二步面试失败结果不丢整体算部分成功上下文传递——前序的structured_output以{agent_type}_result为键注入累积 context供后续 Agent 使用业务门槛——简历加权分 60 分直接终止 Pipeline不进入面试。多 Agent 串联不是简单的跑完就完而是有业务规则的。2.3 结果聚合_aggregate_pipeline把各步结果聚合成统一的AgentResponsecontent是各步文本用---分隔线拼接structured是step_1/step_2...的明细字典fallback_used用any()聚合——只要任一步走了降级整体标记降级。三、统一入口unified_chat.py前端统一 AI 助手只有一个接口POST /api/v1/unified/stream内部完成拦截 → 路由 → 分发 → 流式全流程。3.1 五类零 Token 规则拦截在调 LLM 之前先用规则精确匹配五类社交/元场景直接返回模板零 Token 消耗_HELLO_KEYWORDSfrozenset([你好,您好,hi,hello,hey,...])_THANKS_KEYWORDSfrozenset([谢谢,感谢,辛苦了,...])_BYE_KEYWORDSfrozenset([再见,拜拜,bye,...])_IDENTITY_REre.compile(r(你|您)(是谁|叫什么|...)|介绍.{0,4}(你自己|自己|一下),re.IGNORECASE)_CAPABILITY_REre.compile(r(你|您)(能|可以|会).{0,6}(做|帮|干)|怎么(用|使用)...,re.IGNORECASE)def_pre_filter(text:str)-str|None:t_lower_STRIP_TAIL_RE.sub(,text.strip().lower())# 去末尾标点ift_lowerin_HELLO_KEYWORDS:return_REPLY_HELLOift_lowerin_THANKS_KEYWORDS:return_REPLY_THANKSift_lowerin_BYE_KEYWORDS:return_REPLY_BYEif_IDENTITY_RE.search(text):return_REPLY_IDENTITYif_CAPABILITY_RE.search(text):return_REPLY_CAPABILITYreturnNone为什么值得做这层拦截用户天天说你好、“谢谢”、“你是谁”每次都让 LLM 回答纯属浪费钱和时间。规则拦截把最高频的场景用零成本解决把算力留给真正需要 LLM 的问题。3.2 LLM 意图路由规则没命中才调 LLM 做跨 Agent 路由判断_ROUTE_PROMPT判断用户需求应路由到哪个功能。 可选功能 - qa : 技术知识问答用户直接提问不涉及文件上传 - exam : 试卷/作业批改需上传 Word 答卷... - resume : 简历审查需上传 PDF 简历... - interview : 模拟面试用户提到面试练习面试... - multi_agent : 综合求职准备同时提到简历 面试... - clarify : 意图不明确需要追问 严格按以下 JSON 格式返回不要有其他内容 {{label: 功能名, reason: 一句话说明判断依据}} 用户输入{message}asyncdef_llm_route(message:str)-_RouteResult:llmget_llm(intent,temperature0)respawaitllm.ainvoke([HumanMessage(content_ROUTE_PROMPT.format(messagemessage))])parsedjson.loads(resp.text.strip())labelparsed.get(label,qa).strip().lower()reasonparsed.get(reason,LLM 路由判断)iflabelnotin_VALID_LABELS:# 非法 label → 降级 qalabelqareturn_RouteResult(labellabel,agent_type_LABEL_TO_AGENT[label],execution_mode_LABEL_TO_MODE[label],confidence0.85,# 固定值仅供前端展示reasonreason)路由失败也降级到qa绝不阻断 SSE 流——这是整个项目永远有响应哲学的一贯体现。3.3 SSE 事件分发router.post(/stream)asyncdefunified_chat_stream(req:UnifiedChatRequest,current_userDepends(get_current_user)):asyncdefevent_generator():# Step 0规则前置拦截pre_reply_pre_filter(req.message)ifpre_replyisnotNone:yield_sse({type:token,content:pre_reply})yield_sse({type:done})return# Step 1LLM 路由判断decisionawait_llm_route(req.message)# Step 2推送路由决策卡片yield_sse({type:routing_decision,agent_type:decision.agent_type.value,agent_display:_AGENT_DISPLAY.get(decision.agent_type,),confidence:round(decision.confidence,4),reason:decision.reason,execution_mode:decision.execution_mode.value})# Step 3按 label 分发iflabelqa:asyncforeventin_stream_qa_agent(req,current_user):yieldevent# 透传 QA 的流式事件eliflabelin(exam,resume,interview):yield_sse({type:guidance,...})# 引导跳转对应功能页eliflabelmulti_agent:yield_sse({type:pipeline_plan,steps:[...],...})# 求职全链路计划else:# clarifyyield_sse({type:guidance,message:您的问题我还不太确定应该用哪个功能...})yield_sse({type:done})returnEventSourceResponse(event_generator())SSE 事件类型设计前端据此渲染不同 UI事件作用前端表现routing_decision路由结果显示已转接到 XX卡片progressQA 节点进度检索知识库…提示token回答流式片段逐字渲染guidance引导跳转显示引导卡片跳转按钮pipeline_plan求职全链路计划两步计划卡片metaQA 元数据“引用了 N 篇”done流结束收尾error异常错误提示3.4 QA 流式astream_events 逐事件解析asyncdef_stream_qa_agent(req,current_user):graphorchestrator._get_agent_graph(AgentType.QA)initial_state{...}asyncforeventingraph.astream_events(initial_state,configconfig,versionv2):evtevent[event]nodeevent.get(metadata,{}).get(langgraph_node,)# ① 节点开始且是进度表里的节点 → 推 progressifevton_chain_startandnodein_PROGRESS_LABELS:yield_sse({type:progress,stage:_PROGRESS_LABELS[node]})# ② 生成节点出 token → 推 token回答逐字elifevton_chat_model_streamandnodein_GENERATE_NODES:chunkevent[data].get(chunk)ifchunkandchunk.content:yield_sse({type:token,content:chunk.content})# ③ 生成节点结束 → 收集 answer_mode / sources / confidenceelifevton_chain_endandnodein_GENERATE_NODES:outputevent[data].get(output,{})...astream_events(versionv2)把图内部的一切事件节点开始、模型 token 流、节点结束暴露出来API 层按需筛选转发——LangGraph 的流式能力是 SSE 逐字输出的基础。四、MCP Server给 Agent 装工具项目把知识库检索和 Web 搜索封装成MCP Server挂载在 FastAPI 子应用上# main.py 里把 MCP 变成 ASGI 子应用挂载_kb_appkb_mcp.streamable_http_app()_ws_appws_mcp.streamable_http_app()app.mount(/mcp/kb,_kb_app)# 知识库检索工具app.mount(/mcp/search,_ws_app)# 联网搜索工具QA Agent 的低置信度场景会通过 MCP 调 Web 搜索兜底知识库检索也可以通过 MCP 暴露给外部工具链。这是Agent 能力标准化的方向——工具和 Agent 解耦可复用、可扩展。五、完整请求链路回顾以用户输入帮我准备求职为例走一遍全链路用户输入 帮我准备求职 → 规则拦截不命中五类模板 → LLM 路由label multi_agentexecution_mode pipeline → 推送 routing_decision 事件 pipeline_plan 事件两步计划卡片 → 前端引导用户先传简历 → 调 /resume 接口 → Resume Agent 六维评分 → 前端引导用户开始面试 → 调 /interview 接口 → Interview Agent 五阶段陪练 →如果走后端 PipelineOrchestrator 自动串联 RESUME → INTERVIEW 简历 weighted_score 60 直接终止 → 面试结束 → 生成五维度报告 → 前端展示六、系列总结七篇学到了什么篇主题核心知识点一项目概览架构分层、四大 Agent、技术栈二工程地基pydantic-settings、LLM 工厂、三层兜底三RAG 问答三层意图分类、Hybrid 检索、HyDE、Multi-Query四简历审查直线图、run_in_executor、六维并行、Think Tool五试卷批改三轨并行、限流、interrupt/resume、决策合并六模拟面试五阶段状态机、质量标签、简历联动七系统集成编排器、Pipeline、规则拦截、SSE 分发贯穿七篇的三条工程主线能用规则就不用模型客观题规则批、阶段推进代码写死、问候语模板拦截——LLM 只做真正需要语义理解的事永远有响应三层兜底、单轨失败不拖累其他轨、路由失败降级 qa、单题失败标记人工复核——任何环节挂了用户都不会拿到空响应AI 与人协作HitL 教师终审、低置信度问题入队、needs_review 标记——AI 做初稿人做终审不确定性明确转移。如果你把这个系列从头到尾看完并动手把代码跑起来相信你对生产级多智能体系统的理解会上一个台阶。有任何问题欢迎评论区交流项目源码仅供教学参考未经授权不得用于商业目的。