企业官网建设流程全解析

DeepCode Workflows 的 User-in-Loop 交互处理器：Hook 机制、插件式注册与前后端事件桥接实战指南

【免费下载链接】DeepCode"DeepCode: 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_trigger）；
• Handler 是可选的，可通过配置启用/禁用；
• 工作流只需在交互点调用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_*（阶段完成后）：

从源码结构看，交互点被划分为三类：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_indexing=False): # ===== 添加插件支持 (仅需3行代码) ===== # 1. 创建上下文（自动携带 task_id 与时间戳） context = self._interaction_integration.create_context( task_id=task_id, user_input=requirements, enable_indexing=enable_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.py；
• run_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_response，workflows/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_id=task_id, action=response.action, data=response.data, skipped=response.skipped, ) if not success: raise HTTPException(status_code=404, detail="No 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.py）：

• has_pending_interaction(task_id)：查询某个任务是否存在未决交互；
• cancel_interaction(task_id)：任务被取消时调用，取消对应 Future 并清理记录；
• 等待超时时，_handle_interaction会返回InteractionResponse(action="timeout", skipped=True)并自动清理，见 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

通过默认注册表启停内置 Handler

workflows/interactions暴露了一个进程级默认注册表，可通过get_default_registry()获取。内置的两个 Handler——RequirementAnalysisHandler（需求分析）与PlanReviewHandler（计划确认）——会在首次调用时被自动注册（见 base.py）：

from workflows.interactions import get_default_registry registry = get_default_registry() # 禁用需求分析插件 registry.disable("requirement_analysis") # 启用计划确认插件 registry.enable("plan_review")

InteractionRegistry提供的方法在 base.py 中实现：

tests/test_workflow_interactions.py的test_interaction_registry_lifecycle_has_no_plugin_semantics用例验证了完整的生命周期：注册 → 触发（无回调时自动 skip）→ 禁用后不再执行 → 重新启用 → 注销，见 tests/test_workflow_interactions.py。

run_hook 的执行语义：优先级、超时与容错

run_hook是整套机制的心脏，实现于 base.py，其执行语义值得逐条拆解：

1. 按优先级顺序执行：同一交互点上的 Handler 按priority升序排列，数值越小越先执行（默认priority = 100）；
2. 禁用即跳过：handler.enabled == False时直接跳过；
3. should_trigger决定是否触发：返回False的 Handler 不产生交互；
4. 有回调 + 有 task_id：asyncio.wait_for(callback(task_id, interaction), timeout=interaction.timeout_seconds)等待用户响应；响应skipped=True走on_skip，否则走process_response；超时走on_timeout；
5. 无回调：非必需交互（required=False）自动走on_skip；必需交互（required=True）则抛出RuntimeError，防止静默吞掉关键关卡；
6. 异常隔离：单个 Handler 抛错只会记录error日志并继续执行后续 Handler，不影响其余交互点。

内置 Handler 深度剖析

RequirementAnalysisHandler：AI 引导的需求澄清

RequirementAnalysisHandler挂在BEFORE_PLANNING，优先级 10（最先执行），实现在 workflows/interactions/requirement_analysis.py。其流程为：

1. 用户在计划生成前提交初始需求；
2. Handler 通过RequirementAnalysisAgent.generate_guiding_questions生成 1-3 个针对性问题（功能、技术、性能、UI、部署等维度）；
3. 用户回答问题或直接跳过；
4. 若提交了答案，调用agent.summarize_detailed_requirements生成增强版需求文档；
5. 增强后的需求通过上下文键requirements传递到计划阶段。

should_trigger的判定条件（见 requirement_analysis.py）：

• 上下文中未设置skip_requirement_analysis；
• 尚未处理过（requirements_enhanced为假）；
• 存在初始输入且长度 ≥ 10 字符。

process_response处理用户答案后写入context["requirements"]与context["user_input"]，并标记requirements_enhanced=True（见 requirement_analysis.py）；on_skip/on_timeout则只标记已处理、不修改需求（见 requirement_analysis.py）。底层RequirementAnalysisAgent实现在 workflows/agents/requirement_analysis_agent.py，通过core.compat.Agent+attach_workflow_llm(phase="planning")接入 LLM，用较低温度（问题生成 0.5、需求总结 0.3）换取更稳定的结构化 JSON 输出。

PlanReviewHandler：带修订轮次的计划审批

PlanReviewHandler挂在AFTER_PLANNING，实现在 workflows/interactions/plan_review.py。用户可以对生成的 YAML 实施计划执行四种动作：

should_trigger会跳过skip_plan_review=True或已批准（plan_approved=True）的上下文，并且只有在存在有效计划（上下文中的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": "Let's 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 transport（workflows/interactions/USAGE.md）：

{ "action": "submit", "data": { "answers": { "q1": "Answer 1", "q2": "Answer 2" } }, "skipped": false }

这对应InteractionResponse的三个字段（见 base.py）：action为动作标识（如confirm/modify/submit），data为响应数据，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），说明该事件契约正沿"可跨进程续接"的方向演进。

创建自定义交互插件

实现一个 InteractionHandler

workflows/interactions/USAGE.md 给出了完整的自定义插件模板。继承InteractionHandler需要实现三个抽象方法（见 base.py）：

from 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_type="custom_interaction", title="Custom Check", description="Please 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 秒）；若不允许跳过，设置required=True；
• process_response(response, context) -> context：处理用户响应并返回更新后的上下文；on_skip与on_timeout可按需覆盖以提供默认行为（基类默认把超时当跳过处理，见 base.py）。

装饰器式零侵入接入

除了在 WorkflowService 内部显式调用run_hook，integration.py还提供create_interaction_wrapper工厂函数，可把既有工作流函数"包"进交互点而不改动其代码（见 integration.py）：

execute_planning_with_interactions = create_interaction_wrapper( execute_planning, # 原始函数 before_hooks=[InteractionPoint.BEFORE_PLANNING], after_hooks=[InteractionPoint.AFTER_PLANNING], integration=interaction_integration, )

包装器会在调用原始函数前后依次执行 before/after 钩子，任一环节出现workflow_cancelled即提前返回{"status": "cancelled", "reason": ...}，适合在不动核心流水线源码的前提下为旧流程快速叠加交互能力。

交互点速查表

workflows/interactions/USAGE.md 给出了交互点与默认插件的对照表，结合源码整理如下：

默认注册表仅自动注册需求分析与计划确认两个 Handler（见 base.py），其余交互点预留给业务方自行注册自定义插件。另可参考 workflows/interactions/init.py 了解包的对外导出，以及 workflows/interactions/base.py 中get_default_registry(auto_register=False)参数在规避循环导入时的用法。

优势总结

workflows/interactions/USAGE.md 列出的五大优势，均有对应的源码支撑：

1. 无侵入—— 工作流只需在 Hook 点调用run_hook，核心逻辑一行不改；create_interaction_wrapper甚至能让旧函数零改动接入；
2. 可插拔——InteractionRegistry.enable/disable/unregister支持运行时动态启停与移除（base.py）；
3. 可扩展—— 新增一个交互点只需在InteractionPoint枚举中加一个成员，再继承InteractionHandler实现三个钩子；
4. 可配置—— 可通过上下文开关（如skip_plan_review、skip_requirement_analysis）、构造参数（如PlanReviewHandler(config={"max_modification_rounds": 5})）以及注册表启停来控制行为；
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

【免费下载链接】DeepCode"DeepCode: Open Agentic Coding (Agent Harness & Loop Engineering & Multi-Agent Orchestration)"项目地址: https://gitcode.com/GitHub_Trending/deepc/DeepCode