暂存token使用增量（P0-2 Phase 2） 用于批量提交，减少锁获取次数，提升高频路径性能。 参数: prompt_tokens: 提示token增量 completion_tokens: 完成token增量

对节点进行拓扑排序 返回： 按拓扑顺序排列的节点ID列表

状态转换 参数： new_state: 目标状态 返回： 是否转换成功

添加并验证子节点 参数： parent_id: 父节点ID child_type: 子节点类型 child_config: 子节点配置 返回： {success: bo

添加解析模式

向工作流广播消息 参数： workflow_id: 工作流ID message: 消息内容 exclude_client: 排除的客户端ID

检查代码安全性 参数： code: 要检查的代码 返回： 安全违规列表

预编译表达式为 AST 节点，供重复使用 此方法将表达式字符串解析为 AST，并缓存结果以提升性能。 适用于需要多次执行相同表达式的场景。 参数： expression: 表达式字符串 返回：

转换目标为节点和边 参数： goals: 目标列表 返回： (nodes, edges) 元组

创建JWT访问令牌 生成包含用户信息的JWT token，用于API认证。 Args: data: 要编码到token中的数据（通常包含sub, email, role） expires_delta: 可选的过期时间

创建数据采集节点定义（DATABASE 类型） 用于从数据库获取数据的场景。 参数： name: 节点名称 table: 数据表名 query: 自定义 SQL 查询（优先于 table）

创建引擎快照 返回： 快照数据字典

决定干预措施 参数： rule_violations: 规则违规列表 alignment_score: 对齐分数 resource_usage_ratio: 资源使用比例 suggest

发送增量内容 参数: content: 增量文本 **metadata: 额外元数据

确保规划前有足够预算 如果预算不足，执行压缩。 参数： session: 会话上下文 返回： 是否执行了压缩

估算文本 Token 数 Args: text: 文本 Returns: 估算的 Token 数

执行节点

执行工作流并返回标准结果 参数： workflow: 工作流定义，包含 id, nodes, edges 返回： WorkflowExecutionResult 标准执行结果

执行用例: 追加事件并 (可选) 更新 Run 状态 并发安全: - 使用 CAS 条件更新，不依赖 count_by_run_id - 多个并发事件只有一个能成功触发状态流转 - 终态不会被覆盖回中间态

执行工作流（非流式） 实现 WorkflowExecutorPort.execute 接口

查找相似解法 参数： task_type: 任务类型 task_description: 任务描述 context: 上下文 top_k: 返回前 k 个结果 m

格式化进度事件为用户可读消息 参数: event: ExecutionProgressEvent 实例 返回: 格式化的消息字符串

生成审计报告 参数： start_time: 开始时间 end_time: 结束时间 title: 报告标题 返回： 审计报告字典

创建 LLM 客户端（OpenAI 兼容） 为什么使用工厂函数？ - 延迟初始化：只在需要时创建 LLM - 配置灵活：可以传入不同的配置 - 便于测试：可以传入 Mock 配置 - 便于管理：可以在应用启动时创建，关闭时销毁 参数说明：

非阻塞获取消息 返回: ConversationStep 或 None（如果队列为空）

获取缓存或计算 参数： key: 缓存键 compute_fn: 计算函数（未命中时调用） ttl: 可选的TTL覆盖 返回： 缓存或计算的值

返回 APScheduler job 对象

获取Schema 参数： node_type: 节点类型 返回： 节点Schema，不存在返回None

获取级别严重程度 参数： level: 干预级别 返回： 严重程度数值（越大越严重）

获取工作流状态

将上下文注入到子 Agent 配置 Args: context_package: 上下文包 target_agent_id: 目标子 Agent ID Returns: 子

记录日志并检查告警

创建测试工作流（固定ID以便验证）

注册事件处理器 参数： callback: 事件回调函数

处理工作流事件 参数： event: 工作流事件 - event_type: "workflow_success" 或 "workflow_failure" - workflow_id: 工作流

处理用户消息，修改工作流 参数： workflow: 当前工作流实体 user_message: 用户消息 返回： ModificationResult 包含修改结果和详细信息

按会话 ID 查询

记录失败请求

记录 Agent 启动

记录工作流执行

反思执行结果并生成评估 参数： execution_result: 执行结果 返回： ReflectionResult 反思结果

注册会话 参数： session: 要管理的会话上下文

解析节点间依赖关系 参数： nodes: 节点定义列表 返回： 节点名 -> 依赖节点名列表的映射

回滚到指定版本 回滚会创建一个新版本，内容与目标版本相同。 参数： node_id: 节点ID target_version: 目标版本号 返回： 回滚后的版本详情

向指定会话发送消息 参数： session_id: 会话ID message: 消息 返回： 是否发送成功

设置活跃版本 Args: module_name: 模块名称 version: 版本号 Raises: VersionNotFoundError: 版本不存在

为特定节点设置失败处理策略 参数： node_id: 节点ID strategy: 失败处理策略

监督用户输入（委托到 SupervisionFacade） 对输入文本进行全面检查，包括偏见、有害内容、稳定性检测。 参数： text: 用户输入文本 返回： 包含检查结果的字典：

终止 Agent 参数： agent_id: Agent ID reason: 终止原因 返回： LifecycleResult: 操作结果

转换为字典

处理桥接请求 参数： request: 桥接请求 返回： 桥接结果

更新 Run (必须已存在) 语义: 更新已存在的 Run 记录 实现: 使用 merge() 合并状态 Phase 1 规则: 只 merge，不 commit

验证API请求安全性（代理到 RuleEngineFacade.SafetyGuard） P1-1清理: 改为调用 Facade 以减少重复依赖 参数: node_id: 节点ID url: 请求URL

验证文件操作安全性（代理到 RuleEngineFacade.SafetyGuard） P1-1清理: 改为调用 Facade 以减少重复依赖 参数: node_id: 节点ID operation: 文件操作类型（

获取下一个步骤

聚合子节点输出

广播消息给指定工作流的所有连接 参数： workflow_id: 工作流ID message: 要发送的消息

记录日志（如果有 log_collector）

解析版本号为元组

聚合计算 参数： metric_name: 指标名称 aggregation: 聚合方式 (avg, sum, min, max, count) time_window_minutes: 时间窗口（分钟）

分析上下文 参数： context: 上下文数据 返回： 触发的监督信息列表

广播消息到所有会话 参数： message: 消息 exclude_session: 排除的会话ID 返回： 成功发送的会话数

构建配置对象 返回： CoordinatorAgentConfig 实例 异常： ValueError: 配置验证失败

计算笔记得分 参数： note: 笔记实例 query: 查询字符串 返回： 得分（0-1之间）

检查决策是否与目标对齐 参数： goal: 当前目标 decision: 决策字典 history: 可选的历史决策记录 返回： 对齐检查结果

创建运行上下文

根据依赖关系创建边 参数： nodes: 节点定义列表 返回： 边列表，每个边包含 source 和 target

创建 LangGraph TaskExecutor 适配器 返回一个适配器函数，该函数： 1. 接受 task_name 和 task_description 2. 创建 LangGraph TaskExecutor 3. 构造初始状态 4. 执行任务

创建指标计算节点定义（PYTHON 类型） 用于计算统计指标的场景（使用 pandas）。 参数： name: 节点名称 metrics: 指标列表，如 ["sum", "avg", "max", "min", "c

获取当前目标（栈顶） 返回： 栈顶目标，如果栈为空返回None

发送增量内容

发送工具调用

启用保存请求通道 启用后，ConversationAgent 将通过 SaveRequest 事件请求持久化操作， 而不是直接执行文件写入。

将请求入队 参数： request: 保存请求 返回： 队列位置 异常： SaveRequestQueueFullError: 队列已满

根据目标和错误丰富上下文（委托） 自动检索与目标和错误相关的知识，并将结果附加到上下文中。 参数： workflow_id: 工作流ID goal: 任务目标（可选） errors: 错误列表

执行条件判断 参数： config: 条件配置 inputs: 输入数据 返回： 包含分支信息和下一节点的结果

执行子Agent任务（代理到 SubAgentOrchestrator）

根据工作流 ID 查询历史记录 业务场景： - 用户打开工作流的对话历史页面 - 显示最近的 N 条对话记录 实现： - 按 timestamp 升序排列（旧 → 新） - 限制返回数量（性能优化）

生成用户友好消息 参数: category: 错误分类 details: 详细信息 返回: 用户友好的消息

从用户输入生成规则 参数： start: 起点描述 goal: 目标描述 description: 额外描述 constraints: 额外约束 返回：

获取活跃告警列表

获取执行上下文 参数： workflow_id: 工作流ID 返回： 执行上下文

Return the LangChain tool instance.

处理告警 参数： alert: 告警对象 返回： 处理结果

检索并注入笔记 参数: query: 查询字符串 available_notes: 可用笔记列表 session_id: 会话ID max_total: 最大注入数量 (可选)

判断会话是否被冻结（Step 3） 返回： 是否被冻结

检查节点类型是否已注册 参数： node_type: 节点类型 返回： 是否已注册

判断节点是否需要容器执行 参数： name: 节点名称 返回： 是否需要容器执行

解析输入引用 参数： node_def: 节点定义 返回： 输入名 -> {node: 依赖节点名, path: 输出路径} 的映射