DeepCode Workflows 的 User-in-Loop 交互处理器Hook 机制、插件式注册与前后端事件桥接实战指南【免费下载链接】DeepCodeDeepCode: Open Agentic Coding (Agent Harness Loop Engineering Multi-Agent Orchestration)项目地址: https://gitcode.com/GitHub_Trending/deepc/DeepCode导读workflows/interactions是 DeepCode 中一套插件式的用户交互系统它允许需求澄清、计划审批等人工介入环节像中间件一样无侵入地插入工作流执行链路而无需修改核心工作流代码。本文以 workflows/interactions/USAGE.md 为骨架结合base.py、integration.py、requirement_analysis.py、plan_review.py等源码与tests/test_workflow_interactions.py测试用例完整讲解交互点的定义、Handler 生命周期、注册表优先级调度、与 WorkflowService 的桥接方式以及前后端事件契约并给出可直接落地的自定义插件与配置示例。读完本文你将能够在自己的工作流中新增需求提问计划审批实施前确认等任意交互关卡并理解其与 App Server 传输层解耦的设计原因。核心概念把用户交互做成工作流中的 Hook 点从流水线到挂载点插件式交互的设计意图在常规 Agent 工作流中生成计划 → 生成代码是一条单向流水线用户只能在最开始提交需求、在最后验收结果。DeepCode 的交互系统把这条流水线拆成若干个 Hook Point允许在每个阶段前后挂上一个 Handler由 Handler 决定是否发起一次用户交互工作流执行: [Phase 1] ──▶ [Hook Point] ──▶ [Phase 2] ──▶ [Hook Point] ──▶ [Phase 3] │ │ ▼ ▼ [Handler A] [Handler B] 需求分析 计划确认这条示意图在 workflows/interactions/USAGE.md 中给出。其设计哲学在 workflows/interactions/base.py 的模块注释中写得很明确Handler 被注册到特定的工作流交互点interaction points每个 Handler基于上下文自行决定是否触发should_triggerHandler 是可选的可通过配置启用/禁用工作流只需在交互点调用await interactions.run_hook(...)。也就是说交互逻辑与业务逻辑完全分离工作流代码不知道也不关心哪个 Handler 会触发、用户会怎么回答它只负责在正确的时机抛出一个 Hook 点。注意workflows/interactions/__init__.py首行注释特别强调User-in-loop interaction handlers. These are not Agent Plugins.——这是一套独立于 Agent 插件体系core/plugins的交互基础设施不要在阅读时把两者混淆。InteractionPoint交互点的枚举定义交互点由InteractionPoint枚举定义位于 workflows/interactions/base.py。命名规则为BEFORE_*阶段开始前与AFTER_*阶段完成后交互点定位说明BEFORE_PLANNING生成实施计划之前可用于需求澄清AFTER_PLANNING计划生成之后、实施之前可用于计划审批BEFORE_RESEARCH_ANALYSIS论文分析之前Paper-to-Code 流水线专用AFTER_RESEARCH_ANALYSIS论文分析之后Paper-to-Code 流水线专用AFTER_CODE_PLANNING代码计划生成之后Paper-to-Code 流水线专用BEFORE_IMPLEMENTATION代码生成开始之前通用关卡AFTER_IMPLEMENTATION代码生成之后通用关卡从源码结构看交互点被划分为三类Chat Planning 流水线钩子前两个、Paper-to-Code 流水线钩子中间三个、通用钩子后两个说明该机制既服务于对话式规划也服务于论文转代码Paper2Code流程。快速开始三步把交互插件接入现有工作流第 1 步在 WorkflowService 中添加集成支持workflows/interactions/USAGE.md 给出的接入方式是创建一个WorkflowInteractionIntegration实例然后在流水线的关键位置调用run_hook# workflows/interactions/integration.py 中的 WorkflowInteractionIntegration from workflows.interactions.integration import WorkflowInteractionIntegration from workflows.interactions import InteractionPoint class WorkflowService: def __init__(self): self._tasks {} self._subscribers {} # 添加这一行集成器会自动把交互回调接到注册表上 self._interaction_integration WorkflowInteractionIntegration(self) async def execute_chat_planning(self, task_id, requirements, enable_indexingFalse): # 添加插件支持 (仅需3行代码) # 1. 创建上下文自动携带 task_id 与时间戳 context self._interaction_integration.create_context( task_idtask_id, user_inputrequirements, enable_indexingenable_indexing, ) # 2. 运行 BEFORE_PLANNING 插件 (需求分析) context await self._interaction_integration.run_hook( InteractionPoint.BEFORE_PLANNING, context ) # 检查是否被取消 if context.get(workflow_cancelled): return {status: cancelled, reason: context.get(cancel_reason)} # 使用可能被增强的需求 requirements context.get(requirements, requirements) # 原有的计划生成代码 planning_result await run_chat_planning_agent(requirements, logger) # 添加计划确认插件 context[planning_result] planning_result context await self._interaction_integration.run_hook( InteractionPoint.AFTER_PLANNING, context ) if context.get(workflow_cancelled): return {status: cancelled, reason: context.get(cancel_reason)} # 使用可能被修改的计划 planning_result context.get(planning_result, planning_result) # 继续原有的代码实现流程 ...WorkflowInteractionIntegration的完整实现在 workflows/interactions/integration.py。需要理解的关键点create_context(task_id, **kwargs)返回{task_id: task_id, timestamp: UTC ISO 时间, **kwargs}即所有交互插件共享的工作流上下文容器见 integration.pyrun_hook(hook_point, context)从上下文中取出task_id并委托给InteractionRegistry.run_hook是交互执行的唯一入口见 integration.py构造WorkflowInteractionIntegration(self)时集成器会调用self._registry.set_interaction_callback(self._handle_interaction)把请求交互 → 等待响应的回调自动挂到注册表上见 integration.py。integration.py的模块注释还给出了最小改造范式在每个 Hook 点只加一行context await interactions.run_hook(...)随后用context.get(requirements, user_input)等取值方式接受 Handler 可能做出的修改见 integration.py。第 2 步提供用户响应 API当交互请求发出后工作流会进入等待状态任务状态变为waiting_for_input。用户侧提交响应的入口是submit_responseworkflows/interactions/USAGE.md 展示了一个典型的 FastAPI 路由写法# workflows.py (API routes) router.post(/respond/{task_id}) async def respond_to_interaction(task_id: str, response: InteractionResponseRequest): 用户提交交互响应 success workflow_service._interaction_integration.submit_response( task_idtask_id, actionresponse.action, dataresponse.data, skippedresponse.skipped, ) if not success: raise HTTPException(status_code404, detailNo pending interaction) return {status: ok}submit_response的实现位于 integration.py它在一个task_id - asyncio.Future的待处理交互表中查找对应 Future若存在且未完成则构造InteractionResponse(action, data, skipped)并通过future.set_result(response)唤醒等待中的工作流协程若不存在待处理交互则返回False路由层据此抛出 404。与之配套的还有三个状态管理方法见 integration.pyhas_pending_interaction(task_id)查询某个任务是否存在未决交互cancel_interaction(task_id)任务被取消时调用取消对应 Future 并清理记录等待超时时_handle_interaction会返回InteractionResponse(actiontimeout, skippedTrue)并自动清理见 integration.py。第 3 步前端处理interaction_required事件交互请求并不是通过 HTTP 响应直接返回的而是由_handle_interaction调用self._workflow_service._broadcast(...)广播一条结构化事件见 integration.py。前端在流式通道中订阅该事件即可workflows/interactions/USAGE.md 给出 TypeScript 侧的处理骨架// useStreaming.ts case interaction_required: // 显示交互面板 setInteraction({ type: message.interaction_type, title: message.title, description: message.description, data: message.data, options: message.options, }); break;配置与扩展启用/禁用插件与创建自定义 Handler通过默认注册表启停内置 Handlerworkflows/interactions暴露了一个进程级默认注册表可通过get_default_registry()获取。内置的两个 Handler——RequirementAnalysisHandler需求分析与PlanReviewHandler计划确认——会在首次调用时被自动注册见 base.pyfrom workflows.interactions import get_default_registry registry get_default_registry() # 禁用需求分析插件 registry.disable(requirement_analysis) # 启用计划确认插件 registry.enable(plan_review)InteractionRegistry提供的方法在 base.py 中实现方法作用register(handler)把 Handler 挂到其hook_point并按priority升序排序unregister(name)按名字移除 Handlerenable(name)/disable(name)动态启用/禁用某个 Handlerset_interaction_callback(cb)设置请求交互 → 取回响应的回调get_handlers(hook_point)获取某交互点上的 Handler 列表run_hook(hook_point, context, task_id)按优先级执行某交互点上所有已启用 Handlertests/test_workflow_interactions.py的test_interaction_registry_lifecycle_has_no_plugin_semantics用例验证了完整的生命周期注册 → 触发无回调时自动 skip→ 禁用后不再执行 → 重新启用 → 注销见 tests/test_workflow_interactions.py。run_hook 的执行语义优先级、超时与容错run_hook是整套机制的心脏实现于 base.py其执行语义值得逐条拆解按优先级顺序执行同一交互点上的 Handler 按priority升序排列数值越小越先执行默认priority 100禁用即跳过handler.enabled False时直接跳过should_trigger决定是否触发返回False的 Handler 不产生交互有回调 有 task_idasyncio.wait_for(callback(task_id, interaction), timeoutinteraction.timeout_seconds)等待用户响应响应skippedTrue走on_skip否则走process_response超时走on_timeout无回调非必需交互requiredFalse自动走on_skip必需交互requiredTrue则抛出RuntimeError防止静默吞掉关键关卡异常隔离单个 Handler 抛错只会记录error日志并继续执行后续 Handler不影响其余交互点。内置 Handler 深度剖析RequirementAnalysisHandlerAI 引导的需求澄清RequirementAnalysisHandler挂在BEFORE_PLANNING优先级 10最先执行实现在 workflows/interactions/requirement_analysis.py。其流程为用户在计划生成前提交初始需求Handler 通过RequirementAnalysisAgent.generate_guiding_questions生成 1-3 个针对性问题功能、技术、性能、UI、部署等维度用户回答问题或直接跳过若提交了答案调用agent.summarize_detailed_requirements生成增强版需求文档增强后的需求通过上下文键requirements传递到计划阶段。should_trigger的判定条件见 requirement_analysis.py上下文中未设置skip_requirement_analysis尚未处理过requirements_enhanced为假存在初始输入且长度 ≥ 10 字符。process_response处理用户答案后写入context[requirements]与context[user_input]并标记requirements_enhancedTrue见 requirement_analysis.pyon_skip/on_timeout则只标记已处理、不修改需求见 requirement_analysis.py。底层RequirementAnalysisAgent实现在 workflows/agents/requirement_analysis_agent.py通过core.compat.Agentattach_workflow_llm(phaseplanning)接入 LLM用较低温度问题生成 0.5、需求总结 0.3换取更稳定的结构化 JSON 输出。PlanReviewHandler带修订轮次的计划审批PlanReviewHandler挂在AFTER_PLANNING实现在 workflows/interactions/plan_review.py。用户可以对生成的 YAML 实施计划执行四种动作动作行为confirm批准计划设置plan_approvedTrue进入代码生成modify携带feedback反馈调用revise_plan_with_feedback让 AI 修订计划受max_modification_rounds限制默认 3 轮replace/edit用户直接提供新计划文本通过validate_plan_text校验后替换cancel设置workflow_cancelledTrue与cancel_reason工作流据此提前返回should_trigger会跳过skip_plan_reviewTrue或已批准plan_approvedTrue的上下文并且只有在存在有效计划上下文中的implementation_plan/planning_result或initial_plan_path指向的文件时才触发见 plan_review.py。on_skip/on_timeout都执行自动批准语义并打上plan_auto_approved标记保证无人值守场景下流水线不会被卡死见 plan_review.py。计划修订与审批的持久化、版本化逻辑由 workflows/plan_review_runtime.py 承担这是计划审批的完整运行时计划校验validate_plan_text检查file_structure、implementation_components、validation_approach、environment_setup、implementation_strategy等必需节见 plan_review_runtime.py修订闭环revise_plan_with_feedback用PlanRevisionAgent以温度 0.1 生成修订计划失败时携带上一次校验错误重试最多 2 次见 plan_review_runtime.py版本归档每次修订都把计划保存到plan_versions/initial_plan.vNN.label.txt并把事件追加到plan_review_history.jsonl见 plan_review_runtime.py审批门禁run_plan_review_gate循环生成请求 → 等待决策 → 处理动作支持最多max_rounds 4次交互超限后自动批准当前最新有效计划见 plan_review_runtime.py。应用层事件契约与传输层解耦的 JSON 结构interaction_required工作流发出的交互请求workflows/interactions/USAGE.md 规定WorkflowService 应将interaction_required作为结构化事件交给应用层事件槽。这一广播动作在_handle_interaction中真实执行见 integration.py{ type: interaction_required, task_id: xxx, interaction_type: requirement_questions, title: Lets clarify your requirements, description: Answer these questions..., data: { questions: [...] }, options: { submit: Submit Answers, skip: Skip }, timestamp: 2024-01-01T00:00:00Z }字段说明对照 base.py 的InteractionRequest数据结构interaction_type是交互类型标识data承载交互专属数据问题列表、计划文本、校验结果等options是可用动作按钮映射required表示是否可跳过timeout_seconds表示等待响应超时默认 300 秒PlanReviewHandler覆盖为 600 秒。广播的同时任务状态被置为waiting_for_input并挂上pending_interaction见 integration.py。用户响应结构callback 注入不绑定传输层用户响应通过注入的 callback 返回不绑定 HTTP 或 WebSocket transportworkflows/interactions/USAGE.md{ action: submit, data: { answers: { q1: Answer 1, q2: Answer 2 } }, skipped: false }这对应InteractionResponse的三个字段见 base.pyaction为动作标识如confirm/modify/submitdata为响应数据skipped标记用户是否选择跳过。由于工作流侧只依赖asyncio.Future等待响应、通过submit_response注入结果交互机制本身与 HTTP/WebSocket 完全解耦——这正是 workflows/interactions/USAGE.md 强调的App Server 会在后续阶段把这两个结构映射到版本化 JSON-RPC notification 和approval/respond/workflow/respond方法而 workflow 插件自身不得依赖传输层。事实上core/application/workflow_service.py中已存在面向多进程场景的interaction_id持久化交互等待器_interaction_lock、_InteractionWaiter、checkpoint 中的interaction字段等见 core/application/workflow_service.py说明该事件契约正沿可跨进程续接的方向演进。创建自定义交互插件实现一个 InteractionHandlerworkflows/interactions/USAGE.md 给出了完整的自定义插件模板。继承InteractionHandler需要实现三个抽象方法见 base.pyfrom workflows.interactions import InteractionHandler, InteractionPoint, InteractionRequest class MyCustomHandler(InteractionHandler): name my_custom_handler description My custom interaction hook_point InteractionPoint.BEFORE_IMPLEMENTATION priority 50 async def should_trigger(self, context): return context.get(enable_my_handler, True) async def create_interaction(self, context): return InteractionRequest( interaction_typecustom_interaction, titleCustom Check, descriptionPlease confirm..., data{key: value}, options{yes: Confirm, no: Cancel}, ) async def process_response(self, response, context): if response.action yes: context[custom_confirmed] True else: context[workflow_cancelled] True return context # 注册插件 registry.register(MyCustomHandler())三个钩子的职责划分should_trigger(context) - bool基于上下文决定是否发起交互例如读取context.get(skip_xxx)或检查前置产物是否存在create_interaction(context) - InteractionRequest构造发给用户的交互请求若想控制响应时限可覆盖timeout_seconds如计划审批设为 600 秒若不允许跳过设置requiredTrueprocess_response(response, context) - context处理用户响应并返回更新后的上下文on_skip与on_timeout可按需覆盖以提供默认行为基类默认把超时当跳过处理见 base.py。装饰器式零侵入接入除了在 WorkflowService 内部显式调用run_hookintegration.py还提供create_interaction_wrapper工厂函数可把既有工作流函数包进交互点而不改动其代码见 integration.pyexecute_planning_with_interactions create_interaction_wrapper( execute_planning, # 原始函数 before_hooks[InteractionPoint.BEFORE_PLANNING], after_hooks[InteractionPoint.AFTER_PLANNING], integrationinteraction_integration, )包装器会在调用原始函数前后依次执行 before/after 钩子任一环节出现workflow_cancelled即提前返回{status: cancelled, reason: ...}适合在不动核心流水线源码的前提下为旧流程快速叠加交互能力。交互点速查表workflows/interactions/USAGE.md 给出了交互点与默认插件的对照表结合源码整理如下Hook Point位置默认插件优先级BEFORE_PLANNING生成计划前RequirementAnalysisHandler10AFTER_PLANNING计划生成后PlanReviewHandler10BEFORE_IMPLEMENTATION代码生成前(无)—AFTER_IMPLEMENTATION代码生成后(无)—BEFORE_RESEARCH_ANALYSIS论文分析前(无)—AFTER_RESEARCH_ANALYSIS论文分析后(无)—AFTER_CODE_PLANNING代码计划生成后(无)—默认注册表仅自动注册需求分析与计划确认两个 Handler见 base.py其余交互点预留给业务方自行注册自定义插件。另可参考 workflows/interactions/init.py 了解包的对外导出以及 workflows/interactions/base.py 中get_default_registry(auto_registerFalse)参数在规避循环导入时的用法。优势总结workflows/interactions/USAGE.md 列出的五大优势均有对应的源码支撑无侵入—— 工作流只需在 Hook 点调用run_hook核心逻辑一行不改create_interaction_wrapper甚至能让旧函数零改动接入可插拔——InteractionRegistry.enable/disable/unregister支持运行时动态启停与移除base.py可扩展—— 新增一个交互点只需在InteractionPoint枚举中加一个成员再继承InteractionHandler实现三个钩子可配置—— 可通过上下文开关如skip_plan_review、skip_requirement_analysis、构造参数如PlanReviewHandler(config{max_modification_rounds: 5})以及注册表启停来控制行为解耦合—— 交互事件以 JSON 结构广播、响应经 callback 注入InteractionRequest/InteractionResponse数据结构与传输层完全隔离App Server 可自由选择 JSON-RPC notification 或approval/respond、workflow/respond方法承载而插件侧无需任何改动。延伸阅读完整使用指南workflows/interactions/USAGE.md基类与注册表实现workflows/interactions/base.py工作流集成桥workflows/interactions/integration.py需求分析插件workflows/interactions/requirement_analysis.py计划审批插件workflows/interactions/plan_review.py计划修订与审批运行时workflows/plan_review_runtime.py生命周期测试tests/test_workflow_interactions.py【免费下载链接】DeepCodeDeepCode: Open Agentic Coding (Agent Harness Loop Engineering Multi-Agent Orchestration)项目地址: https://gitcode.com/GitHub_Trending/deepc/DeepCode创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 DeepCode Workflows 的 User-in-Loop 交互处理器：Hook 机制、插件式注册与前后端事件桥接实战指南
DeepCode Workflows 的 User-in-Loop 交互处理器Hook 机制、插件式注册与前后端事件桥接实战指南【免费下载链接】DeepCodeDeepCode: Open Agentic Coding (Agent Harness Loop Engineering Multi-Agent Orchestration)项目地址: https://gitcode.com/GitHub_Trending/deepc/DeepCode导读workflows/interactions是 DeepCode 中一套插件式的用户交互系统它允许需求澄清、计划审批等人工介入环节像中间件一样无侵入地插入工作流执行链路而无需修改核心工作流代码。本文以 workflows/interactions/USAGE.md 为骨架结合base.py、integration.py、requirement_analysis.py、plan_review.py等源码与tests/test_workflow_interactions.py测试用例完整讲解交互点的定义、Handler 生命周期、注册表优先级调度、与 WorkflowService 的桥接方式以及前后端事件契约并给出可直接落地的自定义插件与配置示例。读完本文你将能够在自己的工作流中新增需求提问计划审批实施前确认等任意交互关卡并理解其与 App Server 传输层解耦的设计原因。核心概念把用户交互做成工作流中的 Hook 点从流水线到挂载点插件式交互的设计意图在常规 Agent 工作流中生成计划 → 生成代码是一条单向流水线用户只能在最开始提交需求、在最后验收结果。DeepCode 的交互系统把这条流水线拆成若干个 Hook Point允许在每个阶段前后挂上一个 Handler由 Handler 决定是否发起一次用户交互工作流执行: [Phase 1] ──▶ [Hook Point] ──▶ [Phase 2] ──▶ [Hook Point] ──▶ [Phase 3] │ │ ▼ ▼ [Handler A] [Handler B] 需求分析 计划确认这条示意图在 workflows/interactions/USAGE.md 中给出。其设计哲学在 workflows/interactions/base.py 的模块注释中写得很明确Handler 被注册到特定的工作流交互点interaction points每个 Handler基于上下文自行决定是否触发should_triggerHandler 是可选的可通过配置启用/禁用工作流只需在交互点调用await interactions.run_hook(...)。也就是说交互逻辑与业务逻辑完全分离工作流代码不知道也不关心哪个 Handler 会触发、用户会怎么回答它只负责在正确的时机抛出一个 Hook 点。注意workflows/interactions/__init__.py首行注释特别强调User-in-loop interaction handlers. These are not Agent Plugins.——这是一套独立于 Agent 插件体系core/plugins的交互基础设施不要在阅读时把两者混淆。InteractionPoint交互点的枚举定义交互点由InteractionPoint枚举定义位于 workflows/interactions/base.py。命名规则为BEFORE_*阶段开始前与AFTER_*阶段完成后交互点定位说明BEFORE_PLANNING生成实施计划之前可用于需求澄清AFTER_PLANNING计划生成之后、实施之前可用于计划审批BEFORE_RESEARCH_ANALYSIS论文分析之前Paper-to-Code 流水线专用AFTER_RESEARCH_ANALYSIS论文分析之后Paper-to-Code 流水线专用AFTER_CODE_PLANNING代码计划生成之后Paper-to-Code 流水线专用BEFORE_IMPLEMENTATION代码生成开始之前通用关卡AFTER_IMPLEMENTATION代码生成之后通用关卡从源码结构看交互点被划分为三类Chat Planning 流水线钩子前两个、Paper-to-Code 流水线钩子中间三个、通用钩子后两个说明该机制既服务于对话式规划也服务于论文转代码Paper2Code流程。快速开始三步把交互插件接入现有工作流第 1 步在 WorkflowService 中添加集成支持workflows/interactions/USAGE.md 给出的接入方式是创建一个WorkflowInteractionIntegration实例然后在流水线的关键位置调用run_hook# workflows/interactions/integration.py 中的 WorkflowInteractionIntegration from workflows.interactions.integration import WorkflowInteractionIntegration from workflows.interactions import InteractionPoint class WorkflowService: def __init__(self): self._tasks {} self._subscribers {} # 添加这一行集成器会自动把交互回调接到注册表上 self._interaction_integration WorkflowInteractionIntegration(self) async def execute_chat_planning(self, task_id, requirements, enable_indexingFalse): # 添加插件支持 (仅需3行代码) # 1. 创建上下文自动携带 task_id 与时间戳 context self._interaction_integration.create_context( task_idtask_id, user_inputrequirements, enable_indexingenable_indexing, ) # 2. 运行 BEFORE_PLANNING 插件 (需求分析) context await self._interaction_integration.run_hook( InteractionPoint.BEFORE_PLANNING, context ) # 检查是否被取消 if context.get(workflow_cancelled): return {status: cancelled, reason: context.get(cancel_reason)} # 使用可能被增强的需求 requirements context.get(requirements, requirements) # 原有的计划生成代码 planning_result await run_chat_planning_agent(requirements, logger) # 添加计划确认插件 context[planning_result] planning_result context await self._interaction_integration.run_hook( InteractionPoint.AFTER_PLANNING, context ) if context.get(workflow_cancelled): return {status: cancelled, reason: context.get(cancel_reason)} # 使用可能被修改的计划 planning_result context.get(planning_result, planning_result) # 继续原有的代码实现流程 ...WorkflowInteractionIntegration的完整实现在 workflows/interactions/integration.py。需要理解的关键点create_context(task_id, **kwargs)返回{task_id: task_id, timestamp: UTC ISO 时间, **kwargs}即所有交互插件共享的工作流上下文容器见 integration.pyrun_hook(hook_point, context)从上下文中取出task_id并委托给InteractionRegistry.run_hook是交互执行的唯一入口见 integration.py构造WorkflowInteractionIntegration(self)时集成器会调用self._registry.set_interaction_callback(self._handle_interaction)把请求交互 → 等待响应的回调自动挂到注册表上见 integration.py。integration.py的模块注释还给出了最小改造范式在每个 Hook 点只加一行context await interactions.run_hook(...)随后用context.get(requirements, user_input)等取值方式接受 Handler 可能做出的修改见 integration.py。第 2 步提供用户响应 API当交互请求发出后工作流会进入等待状态任务状态变为waiting_for_input。用户侧提交响应的入口是submit_responseworkflows/interactions/USAGE.md 展示了一个典型的 FastAPI 路由写法# workflows.py (API routes) router.post(/respond/{task_id}) async def respond_to_interaction(task_id: str, response: InteractionResponseRequest): 用户提交交互响应 success workflow_service._interaction_integration.submit_response( task_idtask_id, actionresponse.action, dataresponse.data, skippedresponse.skipped, ) if not success: raise HTTPException(status_code404, detailNo pending interaction) return {status: ok}submit_response的实现位于 integration.py它在一个task_id - asyncio.Future的待处理交互表中查找对应 Future若存在且未完成则构造InteractionResponse(action, data, skipped)并通过future.set_result(response)唤醒等待中的工作流协程若不存在待处理交互则返回False路由层据此抛出 404。与之配套的还有三个状态管理方法见 integration.pyhas_pending_interaction(task_id)查询某个任务是否存在未决交互cancel_interaction(task_id)任务被取消时调用取消对应 Future 并清理记录等待超时时_handle_interaction会返回InteractionResponse(actiontimeout, skippedTrue)并自动清理见 integration.py。第 3 步前端处理interaction_required事件交互请求并不是通过 HTTP 响应直接返回的而是由_handle_interaction调用self._workflow_service._broadcast(...)广播一条结构化事件见 integration.py。前端在流式通道中订阅该事件即可workflows/interactions/USAGE.md 给出 TypeScript 侧的处理骨架// useStreaming.ts case interaction_required: // 显示交互面板 setInteraction({ type: message.interaction_type, title: message.title, description: message.description, data: message.data, options: message.options, }); break;配置与扩展启用/禁用插件与创建自定义 Handler通过默认注册表启停内置 Handlerworkflows/interactions暴露了一个进程级默认注册表可通过get_default_registry()获取。内置的两个 Handler——RequirementAnalysisHandler需求分析与PlanReviewHandler计划确认——会在首次调用时被自动注册见 base.pyfrom workflows.interactions import get_default_registry registry get_default_registry() # 禁用需求分析插件 registry.disable(requirement_analysis) # 启用计划确认插件 registry.enable(plan_review)InteractionRegistry提供的方法在 base.py 中实现方法作用register(handler)把 Handler 挂到其hook_point并按priority升序排序unregister(name)按名字移除 Handlerenable(name)/disable(name)动态启用/禁用某个 Handlerset_interaction_callback(cb)设置请求交互 → 取回响应的回调get_handlers(hook_point)获取某交互点上的 Handler 列表run_hook(hook_point, context, task_id)按优先级执行某交互点上所有已启用 Handlertests/test_workflow_interactions.py的test_interaction_registry_lifecycle_has_no_plugin_semantics用例验证了完整的生命周期注册 → 触发无回调时自动 skip→ 禁用后不再执行 → 重新启用 → 注销见 tests/test_workflow_interactions.py。run_hook 的执行语义优先级、超时与容错run_hook是整套机制的心脏实现于 base.py其执行语义值得逐条拆解按优先级顺序执行同一交互点上的 Handler 按priority升序排列数值越小越先执行默认priority 100禁用即跳过handler.enabled False时直接跳过should_trigger决定是否触发返回False的 Handler 不产生交互有回调 有 task_idasyncio.wait_for(callback(task_id, interaction), timeoutinteraction.timeout_seconds)等待用户响应响应skippedTrue走on_skip否则走process_response超时走on_timeout无回调非必需交互requiredFalse自动走on_skip必需交互requiredTrue则抛出RuntimeError防止静默吞掉关键关卡异常隔离单个 Handler 抛错只会记录error日志并继续执行后续 Handler不影响其余交互点。内置 Handler 深度剖析RequirementAnalysisHandlerAI 引导的需求澄清RequirementAnalysisHandler挂在BEFORE_PLANNING优先级 10最先执行实现在 workflows/interactions/requirement_analysis.py。其流程为用户在计划生成前提交初始需求Handler 通过RequirementAnalysisAgent.generate_guiding_questions生成 1-3 个针对性问题功能、技术、性能、UI、部署等维度用户回答问题或直接跳过若提交了答案调用agent.summarize_detailed_requirements生成增强版需求文档增强后的需求通过上下文键requirements传递到计划阶段。should_trigger的判定条件见 requirement_analysis.py上下文中未设置skip_requirement_analysis尚未处理过requirements_enhanced为假存在初始输入且长度 ≥ 10 字符。process_response处理用户答案后写入context[requirements]与context[user_input]并标记requirements_enhancedTrue见 requirement_analysis.pyon_skip/on_timeout则只标记已处理、不修改需求见 requirement_analysis.py。底层RequirementAnalysisAgent实现在 workflows/agents/requirement_analysis_agent.py通过core.compat.Agentattach_workflow_llm(phaseplanning)接入 LLM用较低温度问题生成 0.5、需求总结 0.3换取更稳定的结构化 JSON 输出。PlanReviewHandler带修订轮次的计划审批PlanReviewHandler挂在AFTER_PLANNING实现在 workflows/interactions/plan_review.py。用户可以对生成的 YAML 实施计划执行四种动作动作行为confirm批准计划设置plan_approvedTrue进入代码生成modify携带feedback反馈调用revise_plan_with_feedback让 AI 修订计划受max_modification_rounds限制默认 3 轮replace/edit用户直接提供新计划文本通过validate_plan_text校验后替换cancel设置workflow_cancelledTrue与cancel_reason工作流据此提前返回should_trigger会跳过skip_plan_reviewTrue或已批准plan_approvedTrue的上下文并且只有在存在有效计划上下文中的implementation_plan/planning_result或initial_plan_path指向的文件时才触发见 plan_review.py。on_skip/on_timeout都执行自动批准语义并打上plan_auto_approved标记保证无人值守场景下流水线不会被卡死见 plan_review.py。计划修订与审批的持久化、版本化逻辑由 workflows/plan_review_runtime.py 承担这是计划审批的完整运行时计划校验validate_plan_text检查file_structure、implementation_components、validation_approach、environment_setup、implementation_strategy等必需节见 plan_review_runtime.py修订闭环revise_plan_with_feedback用PlanRevisionAgent以温度 0.1 生成修订计划失败时携带上一次校验错误重试最多 2 次见 plan_review_runtime.py版本归档每次修订都把计划保存到plan_versions/initial_plan.vNN.label.txt并把事件追加到plan_review_history.jsonl见 plan_review_runtime.py审批门禁run_plan_review_gate循环生成请求 → 等待决策 → 处理动作支持最多max_rounds 4次交互超限后自动批准当前最新有效计划见 plan_review_runtime.py。应用层事件契约与传输层解耦的 JSON 结构interaction_required工作流发出的交互请求workflows/interactions/USAGE.md 规定WorkflowService 应将interaction_required作为结构化事件交给应用层事件槽。这一广播动作在_handle_interaction中真实执行见 integration.py{ type: interaction_required, task_id: xxx, interaction_type: requirement_questions, title: Lets clarify your requirements, description: Answer these questions..., data: { questions: [...] }, options: { submit: Submit Answers, skip: Skip }, timestamp: 2024-01-01T00:00:00Z }字段说明对照 base.py 的InteractionRequest数据结构interaction_type是交互类型标识data承载交互专属数据问题列表、计划文本、校验结果等options是可用动作按钮映射required表示是否可跳过timeout_seconds表示等待响应超时默认 300 秒PlanReviewHandler覆盖为 600 秒。广播的同时任务状态被置为waiting_for_input并挂上pending_interaction见 integration.py。用户响应结构callback 注入不绑定传输层用户响应通过注入的 callback 返回不绑定 HTTP 或 WebSocket transportworkflows/interactions/USAGE.md{ action: submit, data: { answers: { q1: Answer 1, q2: Answer 2 } }, skipped: false }这对应InteractionResponse的三个字段见 base.pyaction为动作标识如confirm/modify/submitdata为响应数据skipped标记用户是否选择跳过。由于工作流侧只依赖asyncio.Future等待响应、通过submit_response注入结果交互机制本身与 HTTP/WebSocket 完全解耦——这正是 workflows/interactions/USAGE.md 强调的App Server 会在后续阶段把这两个结构映射到版本化 JSON-RPC notification 和approval/respond/workflow/respond方法而 workflow 插件自身不得依赖传输层。事实上core/application/workflow_service.py中已存在面向多进程场景的interaction_id持久化交互等待器_interaction_lock、_InteractionWaiter、checkpoint 中的interaction字段等见 core/application/workflow_service.py说明该事件契约正沿可跨进程续接的方向演进。创建自定义交互插件实现一个 InteractionHandlerworkflows/interactions/USAGE.md 给出了完整的自定义插件模板。继承InteractionHandler需要实现三个抽象方法见 base.pyfrom workflows.interactions import InteractionHandler, InteractionPoint, InteractionRequest class MyCustomHandler(InteractionHandler): name my_custom_handler description My custom interaction hook_point InteractionPoint.BEFORE_IMPLEMENTATION priority 50 async def should_trigger(self, context): return context.get(enable_my_handler, True) async def create_interaction(self, context): return InteractionRequest( interaction_typecustom_interaction, titleCustom Check, descriptionPlease confirm..., data{key: value}, options{yes: Confirm, no: Cancel}, ) async def process_response(self, response, context): if response.action yes: context[custom_confirmed] True else: context[workflow_cancelled] True return context # 注册插件 registry.register(MyCustomHandler())三个钩子的职责划分should_trigger(context) - bool基于上下文决定是否发起交互例如读取context.get(skip_xxx)或检查前置产物是否存在create_interaction(context) - InteractionRequest构造发给用户的交互请求若想控制响应时限可覆盖timeout_seconds如计划审批设为 600 秒若不允许跳过设置requiredTrueprocess_response(response, context) - context处理用户响应并返回更新后的上下文on_skip与on_timeout可按需覆盖以提供默认行为基类默认把超时当跳过处理见 base.py。装饰器式零侵入接入除了在 WorkflowService 内部显式调用run_hookintegration.py还提供create_interaction_wrapper工厂函数可把既有工作流函数包进交互点而不改动其代码见 integration.pyexecute_planning_with_interactions create_interaction_wrapper( execute_planning, # 原始函数 before_hooks[InteractionPoint.BEFORE_PLANNING], after_hooks[InteractionPoint.AFTER_PLANNING], integrationinteraction_integration, )包装器会在调用原始函数前后依次执行 before/after 钩子任一环节出现workflow_cancelled即提前返回{status: cancelled, reason: ...}适合在不动核心流水线源码的前提下为旧流程快速叠加交互能力。交互点速查表workflows/interactions/USAGE.md 给出了交互点与默认插件的对照表结合源码整理如下Hook Point位置默认插件优先级BEFORE_PLANNING生成计划前RequirementAnalysisHandler10AFTER_PLANNING计划生成后PlanReviewHandler10BEFORE_IMPLEMENTATION代码生成前(无)—AFTER_IMPLEMENTATION代码生成后(无)—BEFORE_RESEARCH_ANALYSIS论文分析前(无)—AFTER_RESEARCH_ANALYSIS论文分析后(无)—AFTER_CODE_PLANNING代码计划生成后(无)—默认注册表仅自动注册需求分析与计划确认两个 Handler见 base.py其余交互点预留给业务方自行注册自定义插件。另可参考 workflows/interactions/init.py 了解包的对外导出以及 workflows/interactions/base.py 中get_default_registry(auto_registerFalse)参数在规避循环导入时的用法。优势总结workflows/interactions/USAGE.md 列出的五大优势均有对应的源码支撑无侵入—— 工作流只需在 Hook 点调用run_hook核心逻辑一行不改create_interaction_wrapper甚至能让旧函数零改动接入可插拔——InteractionRegistry.enable/disable/unregister支持运行时动态启停与移除base.py可扩展—— 新增一个交互点只需在InteractionPoint枚举中加一个成员再继承InteractionHandler实现三个钩子可配置—— 可通过上下文开关如skip_plan_review、skip_requirement_analysis、构造参数如PlanReviewHandler(config{max_modification_rounds: 5})以及注册表启停来控制行为解耦合—— 交互事件以 JSON 结构广播、响应经 callback 注入InteractionRequest/InteractionResponse数据结构与传输层完全隔离App Server 可自由选择 JSON-RPC notification 或approval/respond、workflow/respond方法承载而插件侧无需任何改动。延伸阅读完整使用指南workflows/interactions/USAGE.md基类与注册表实现workflows/interactions/base.py工作流集成桥workflows/interactions/integration.py需求分析插件workflows/interactions/requirement_analysis.py计划审批插件workflows/interactions/plan_review.py计划修订与审批运行时workflows/plan_review_runtime.py生命周期测试tests/test_workflow_interactions.py【免费下载链接】DeepCodeDeepCode: Open Agentic Coding (Agent Harness Loop Engineering Multi-Agent Orchestration)项目地址: https://gitcode.com/GitHub_Trending/deepc/DeepCode创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考