Fay 的大小模型协同工作逻辑

课程IDcourse-fay-big-small-model-collab作者Fay 开源社区版本1.0.0章节数6封面目录为什么 Fay 要搞大小模型协同模型工厂按 role 切换大小模型小模型当第一响应者闲聊判断与工具分流大模型后台执行循环ExecutionManager任务运行中新消息的意图分类后台完成回填小模型改写结果与全链路总结第1节 为什么 Fay 要搞大小模型协同欢迎来到 Fay 大小模型协同工作课程。单模型做数字人对话有一个绕不开的矛盾想让首句响应又快又便宜就得用小模型但小模型在工具规划、长链路推理、事实核查上能力有限换大模型又会让每一次你好都要等三五秒、花一笔不小的 token 费。Fay 的解法是让两种模型各干各擅长的事小模型当第一响应者立刻开口接住用户判断这是闲聊还是需要查工具大模型当后台执行者在独立线程里跑工具调用循环完成后把结果交还给小模型改写成最终回复。这套机制在 Fay 里由三块代码支撑system.conf 里分别配置 gpt_model_* 与 big_model_* 两组模型llm/nlp_cognitive_stream.py 的 question 函数做分流llm/execution_manager.py 跑后台执行循环。本课程会顺着这条链路把协同机制从入口到回填完整讲一遍。# system.conf - 两套模型并存 # 小模型扛首句响应、闲聊判断 gpt_model_engine qwen/qwen3-4b-2507 gpt_base_url https://... key_gpt_api_key sk-... # 大模型扛后台工具执行循环 big_model_engine MiniMax-M2.7 big_model_base_url https://api.minimax.chat/v1 big_model_api_key sk-cp-... # 大模型未配置时会自动降级为小模型 # 也就是说小配置可以单跑协同是可选增强第2节 模型工厂按 role 切换大小模型第二节看一眼 Fay 如何在一份代码里同时拿到大模型和小模型实例。关键函数在 llm/execution_manager.py 的 _get_llm_instance。它只有一个参数 role取值 small 或 big返回一个 ChatOpenAI 对象。当 role 是 big 时它优先读 system.conf 里的 big_model_engine、big_model_base_url、big_model_api_key。如果某个 base_url 或 api_key 为空会回落到小模型的 gpt_base_url、key_gpt_api_key也就是说你可以只换模型名不换服务商。关键的一行是如果 big_model_engine 本身没配会打印请求大模型但未配置降级为小模型然后按小模型参数构造。这让 Fay 对单模型用户依旧可用——协同是增强特性而不是硬依赖。另外一个细节是超时与重试大模型 timeout 120 秒、max_retries 2小模型 timeout 60 秒、max_retries 1。这背后的假设是——大模型要跑工具循环、响应时间本来就长小模型要扛首句响应失败了宁可快速失败也别让用户干等。# llm/execution_manager.py - 模型工厂 def _get_llm_instance(role: str small, streaming: bool True) - ChatOpenAI: cfg.load_config() if role big: if cfg.big_model_engine: actual_base_url cfg.big_model_base_url or cfg.gpt_base_url actual_api_key cfg.big_model_api_key or cfg.key_gpt_api_key return ChatOpenAI( modelcfg.big_model_engine, base_urlactual_base_url, api_keyactual_api_key, streamingstreaming, timeout120, max_retries2, ) # 无大模型配置 → 降级为小模型 util.log(1, 请求大模型但未配置 big_model_engine降级为小模型) # small / 降级路径 return ChatOpenAI( modelcfg.gpt_model_engine, base_urlcfg.gpt_base_url, api_keycfg.key_gpt_api_key, streamingstreaming, timeout60, max_retries1, )第3节 小模型当第一响应者闲聊判断与工具分流第三节讲小模型当第一响应者时它到底在判断什么。在 question 函数里如果没有正在执行的后台任务且注册了工具Fay 会调用 _call_planner_llm 走规划器。规划器的 system prompt 很特别——它不是一个通用助手而是一个闲聊判断器只输出两种合法 JSON是闲聊就 {action: finish, message: ...}不是闲聊就 {action: tool, keyword: ...}。关键约束写在 prompt 里小模型被明确告知你是第一响应者涉及事实信息的回答系统会在后台另起一个大模型自动核实。所以它被禁止在 message 里写我来查一下稍等我核实这种过渡语——过渡语由系统统一插入小模型写了只会和系统重复。流式阶段还有一个设计规划器边流边解析。只要早期 token 能判定是 tool 分支立刻触发 on_tool_detected 回调先推一句我来帮你查一下稍等…给用户然后把工具调用提交给大模型后台。这一下把大模型开始干活的感知延迟几乎压到零。还有一个兜底如果小模型把 finish 写得超过 80 字Fay 会认为它在硬编事实信息自动追加等等我再帮你核实一下…并把请求转交给大模型。# llm/nlp_cognitive_stream.py - 规划器分流节选 first_decision _call_planner_llm( plan_state, stream_callback_first_plan_stream_callback, # finish 边流边出 on_tool_detected_on_tool_detected, # tool 时提前插过渡语 ) if first_decision[action] tool: # kb_search 场景硬塞首工具给大模型 return _submit_tool_execution( {tool: kb_search, args: {query: search_query}}, show_plan_msgnot already_notified, ) # finish 分支的越界兜底finish 过长 → 自动转核实 finish_msg first_decision.get(message, ) if has_tools and len(finish_msg) 80 and len(content.strip()) 3: write_sentence(\n\n等等我再帮你核实一下…\n\n---\n) return _submit_tool_execution( {tool: None, args: {}}, show_plan_msgFalse, unverified_responsefinish_msg, # 带上小模型原回复供大模型对照 )第4节 大模型后台执行循环ExecutionManager第四节看大模型在后台到底怎么跑。llm/execution_manager.py 定义了一个 ExecutionManager 单例内部维护两张字典_states 存每个用户当前的 ExecutionState_threads 存对应的后台线程。submit 方法会先检查同一用户是否已有运行中的任务如果有就直接返回 False——也就是说每个用户同时只有一个大模型线程防止并发工具冲突。核心循环是 _big_model_execute。流程是这样的如果小模型在规划阶段硬塞了首工具比如 kb_search就以它作为起点否则让大模型自己规划首步。然后进入最多 30 步的执行循环——调用工具、记录结果、把整个已执行摘要塞给大模型让它决策下一步要么继续 tool要么 finish。这里有两个值得注意的防御机制。第一是防重复调用每次大模型返回一个新 tool 决策先扫整段 tool_results如果已经有同名同参且成功的调用就直接终止循环——避免模型在上下文压缩后重新发一遍同样的工具。第二是 cancel_flag 与 modify_request用户在任务执行期间想插话的话可以通过管理器传进来循环每一轮都会检查一次。循环结束后所有工具结果会被拼成 final_tool_context 字符串再加上大模型最后给出的 final_response_hint一并等小模型来消费。# llm/execution_manager.py - 后台执行循环节选 def _big_model_execute(state: ExecutionState): big_llm _get_llm_instance(big, streamingFalse) current_action state.first_plan # 可能由小模型预先指定 max_steps 30 while len(state.tool_results) max_steps: if state.cancel_flag: return # 用户取消 if state.modify_request: # 用户补充要求 state.audit_log.append(f收到修改指令: {state.modify_request}) state.modify_request None spec state.tool_registry.get(current_action[name]) success, output, error spec.executor(current_action[args], attempts) state.tool_results.append({call: current_action, success: success, ...}) # 大模型决策下一步 decision _extract_decision(big_llm.invoke(_build_execution_next_step_messages(state))) if decision[action] tool: # 防重复同名同参已成功 → 直接终止 if any(r[success] and r[call] decision_call for r in state.tool_results): break current_action {name: decision[tool], args: decision[args]} else: # finish state.final_response_hint decision.get(message, ) break第5节 任务运行中新消息的意图分类第五节看一个协同框架绕不开的难题后台任务还在跑用户又发了一条新消息该怎么处理。Fay 的答案在 question 函数的情况2分支。当检测到用户有 RUNNING 状态的 ExecutionState 时会用小模型跑一次轻量级意图分类输出五个标签之一。update_task 表示用户在补充当前任务的要求——比如原来问天气补一句只要今天的这时候调用 exec_mgr.modify 把指令塞给正在跑的大模型循环大模型下一轮决策会看到这条补充。query_progress 表示用户在问进度——好了没还要多久系统直接从 state.current_step 和 tool_results 长度读出当前进度报出来不打扰大模型。cancel_task 就是放弃任务直接把 cancel_flag 置 True大模型循环在下一轮检查时退出。new_task 最微妙——系统判断用户提的是一个独立新任务。这时候不会粗暴打断而是生成一句反问我正在帮你处理 X你是想加这个新需求还是改做新的交给用户确认。normal_chat 是和任务无关的闲聊小模型直接用 run_direct_llm 回复一句后台任务继续跑不受影响。这套分流让协同不只是两个模型轮流跑而是在用户看来对话始终流畅。# llm/nlp_cognitive_stream.py - question() 情况2节选 running_state exec_mgr.get_state(username) if running_state and running_state.status ExecutionStatus.RUNNING: intent _classify_intent_for_running_task(content, running_state) if intent update_task: exec_mgr.modify(username, content) # 补充要求塞入大模型循环 reply 好的我已经把你的补充要求传达给正在执行的任务了。 elif intent query_progress: reply (f正在执行「{running_state.original_request}」\n f当前进度{running_state.current_step} f已完成 {len(running_state.tool_results)} 步。) elif intent cancel_task: exec_mgr.cancel(username) # 置 cancel_flag reply 好的已取消正在执行的任务。 elif intent new_task: reply _build_new_task_confirm_reply(content, running_state) # 反问确认 else: # normal_chat run_direct_llm() # 小模型直接回复不打扰后台第6节 后台完成回填小模型改写结果与全链路总结最后一节看大模型跑完后结果怎么回到用户眼前并把整条协同链路串起来。大模型后台线程执行完毕时ExecutionManager 会触发 on_complete 回调——也就是 _auto_reply_after_execution。这个函数的目标是不重新进入 question 全流程避免重新加载记忆与历史导致上下文溢出用一个精简 prompt 让小模型基于工具结果生成最终回复并写回原始 conversation_id 的流。精简 prompt 里只保留三样东西用户原始请求、工具调用细节、工具执行结果。系统明确告诉小模型你已经说过我来帮你查一下稍等…不要再重复过渡语直接给答案。小模型流式输出按标点切句写入 stream_manager。一个很有意思的细节Fay 会把大模型的完整执行日志包在 … 里作为最终消息的前缀GUI 会把它折叠成思考过程面板。用户既看到干净的答案也能点开看大模型到底调了哪些工具、耗时多少。如果这是一次核实场景兜底分支传入了 unverified_response小模型的 prompt 还多一条规则比对工具结果和之前的未核实回复——一致就说核实了一下刚才说的没问题有矛盾才更正。避免为了显得勤快去修改其实正确的回答。整条协同链路串一下 第一用户消息进 question小模型规划器边流边判断闲聊直接流出非闲聊推过渡语并提交大模型。 第二ExecutionManager 在独立线程里跑大模型循环防重复、可取消、可修改、最多 30 步。 第三循环期间用户的新消息由小模型做意图分类根据 update/query/cancel/new/normal 五种分支分别处理。 第四循环结束_auto_reply_after_execution 用精简 prompt 让小模型改写成最终回复附执行日志 。 第五回复进入同一 conversation_id 的流记忆存入对话历史状态清空等待下一轮。一份小模型配置 一份大模型配置 一个执行管理器 Fay 在数字人场景下的低延迟、高质量对话。这就是 Fay 大小模型协同的全貌。# llm/nlp_cognitive_stream.py - _auto_reply_after_execution节选 def _auto_reply_after_execution(username, finished_exec_state): sm stream_manager.new_instance() conv_id finished_exec_state.conversation_id sm.set_current_conversation(username, conv_id) # 复用原会话 tool_context finished_exec_state.final_tool_context[:4000] hint finished_exec_state.final_response_hint[:500] compact_system f你已经告诉用户我来帮你查一下稍等…现在工具执行完毕。 请基于工具结果直接回答不要再重复过渡语。 **用户消息**: {finished_exec_state.original_request} **工具执行结果**: {tool_context} small_llm _get_llm_instance(small, streamingTrue) # 把大模型执行日志作为 think 前缀写回 think_tag fthink\n执行耗时: {elapsed}s共 {len(tool_results)} 步\n...\n/think\n _write(think_tag, force_firstTrue) for chunk in small_llm.stream([SystemMessage(contentcompact_system), HumanMessage(...)]): # 按标点切句流式写回同一 conversation_id ... exec_mgr.consume_result(username) # 清理状态 MyThread(targetremember_conversation_thread, args(username, original_request, full_text)).start() # 记忆归档 # 全链路协同矩阵 # ------------------------------------------------- # | 阶段 | 角色 | 模型 | 关键产出 | # ------------------------------------------------- # | 入口 | 规划器 | 小模型 | finish / tool | # | 执行 | 执行者 | 大模型 | tool_results | # | 插话 | 分类器 | 小模型 | intent 标签 | # | 回填 | 改写器 | 小模型 | 最终流式回复 | # -------------------------------------------------