Kairos：契约约束的 AI Agent 工作流安全编排框架

Kairos 是一个安全强化的 Python SDK，通过契约验证、敏感信息脱敏和失败恢复机制，为 AI Agent 多步骤工作流提供生产级的安全保障。

AI Agent 工作流的安全困境\n\n随着大语言模型（LLM）能力的不断提升，基于 AI Agent 的自动化工作流正在成为企业应用的新范式。然而，在构建复杂的多步骤 Agent 工作流时，开发者往往面临一系列安全挑战：\n\n静默故障传播：一个步骤产生错误输出，却未被检测，导致错误在后续步骤中累积放大。\n\n上下文丢失：Agent 在执行长任务过程中丢失关键上下文，导致行为偏离预期。\n\n提示词注入风险：当步骤失败需要重试时，直接将原始错误信息或异常堆栈喂给模型，可能成为提示词注入的攻击向量。\n\n敏感信息泄露：日志中可能意外暴露 API 密钥、密码等敏感数据。\n\n缺乏恢复机制：当某个步骤失败时，系统缺乏有效的重试、跳过或重新规划策略。\n\n现有的编排工具（如 LangGraph、CrewAI）和验证工具（如 Guardrails AI）各有所长，但鲜有方案将编排与验证深度结合，并将安全作为一等公民来设计。Kairos 正是为填补这一空白而生。\n\n## Kairos 的核心设计理念\n\nKairos 的名字源自古希腊语"καιρός"，意为"恰当时机"——强调在正确的时间采取正确的行动。这一理念体现在其严格的执行循环设计中：\n\n\n目标 → 规划 → 执行步骤 → 验证输出 → 通过/重试/重新规划 → 下一步 → 完成\n\n\n与传统 Agent 框架不同，Kairos 将契约验证内建于每一步骤之间，确保数据质量在流程中始终保持。\n\n## 六大安全特性详解\n\n### 1. 步骤间契约验证（Inter-step Contract Validation）\n\n每个步骤都必须声明其输入和输出的数据结构契约。Kairos 支持多种契约定义方式：\n\npython\nfrom kairos import Schema\n\nschema = Schema({\n \"name\": str,\n \"products\": list[str],\n \"score\": float | None,\n})\n\n\n步骤执行完成后，系统会自动验证输出是否符合契约。只有验证通过的数据才会流入下一步，从根本上杜绝了错误数据的静默传播。\n\n### 2. 净化重试上下文（Sanitized Retry Context）\n\n这是 Kairos 最具创新性的安全特性之一。当步骤执行失败需要重试时，传统做法是将原始错误信息直接反馈给模型。然而，错误信息中可能包含恶意内容或敏感数据，直接注入提示词存在安全风险。\n\nKairos 的做法是：仅注入结构化的元数据（字段名、类型、尝试次数），原始 LLM 输出和异常消息永远不会进入提示词。这种设计有效防御了通过错误消息进行的提示词注入攻击。\n\n### 3. 最小权限状态访问（Scoped State Access）\n\n在多步骤工作流中，每个步骤默认可以访问整个状态空间，这违反了最小权限原则。Kairos 通过 read_keys 和 write_keys 参数，精确控制每个步骤可以读取和写入的状态键：\n\npython\nstep = Step(\n name=\"analyze\",\n action=analyze_fn,\n read_keys=[\"input_data\"],\n write_keys=[\"analysis_result\"]\n)\n\n\n这种设计不仅提升了安全性，还增强了代码的可读性和可维护性。\n\n### 4. 敏感键自动脱敏（Sensitive Key Redaction）\n\nKairos 内置了敏感键模式识别，自动识别包含 password、token、api_key 等关键词的状态项。这些敏感值在日志、导出和最终状态中都会被自动脱敏，防止意外泄露。\n\n### 5. 异常信息净化（Exception Sanitization）\n\n当步骤抛出异常时，Kairos 会在存储或记录之前自动剥离凭证、文件路径和原始堆栈跟踪信息。这确保了即使在调试过程中，敏感信息也不会暴露。\n\n### 6. 可配置失败策略（Configurable Failure Policies）\n\n不同的步骤可能需要不同的失败处理策略。Kairos 提供了灵活的策略配置：\n\npython\nfrom kairos import FailurePolicy, FailureAction\n\nstep = Step(\n name=\"critical_step\",\n action=critical_fn,\n failure_policy=FailurePolicy(\n on_validation_fail=FailureAction.RETRY,\n on_execution_fail=FailureAction.ABORT,\n max_retries=3,\n ),\n)\n\n\n支持的动作包括：\n- RETRY：重试执行\n- SKIP：跳过该步骤\n- ABORT：中止整个工作流\n- REPLAN：重新规划后续步骤\n\n策略支持三级继承：步骤级 → 工作流级 → Kairos 默认值，最具体的配置优先。\n\n## 与现有方案的对比\n\n| 需求 | LangGraph | CrewAI | Guardrails AI | Kairos |\n|------|-----------|--------|---------------|--------|\n| 多步骤工作流编排 | ✓ | ✓ | ✗ | ✓ |\n| 步骤间契约验证 | ✗ | 部分 | ✗（仅单输出） | ✓ |\n| 净化重试上下文 | ✗ | ✗ | N/A | ✓ |\n| 最小权限状态访问 | ✗ | ✗ | N/A | ✓ |\n| 敏感键脱敏 | ✗ | ✗ | N/A | ✓ |\n| 可配置失败策略 | 部分 | 部分 | N/A | ✓ |\n\nKairos 的独特价值在于将安全作为架构的核心考量，而非事后补丁。\n\n## 依赖关系与循环处理\n\nKairos 支持复杂的工作流拓扑，包括步骤间的依赖关系和循环处理：\n\npython\nworkflow = Workflow(\n name=\"competitive_analysis\",\n steps=[\n Step(name=\"fetch_competitors\", action=fetch_fn),\n Step(name=\"analyze_each\", action=analyze_fn,\n depends_on=[\"fetch_competitors\"],\n foreach=\"fetch_competitors\"),\n Step(name=\"summarize\", action=summarize_fn,\n depends_on=[\"analyze_each\"]),\n ],\n)\n\n\nforeach 支持基于前序步骤输出的动态循环，每个迭代独立执行并验证。\n\n## 模型无关设计\n\nKairos 不绑定任何特定的 LLM 提供商。任何接受 StepContext 参数的可调用对象都可以作为步骤动作——无论是调用 OpenAI API、本地 Ollama 模型，还是纯函数逻辑。\n\npython\nfrom kairos import Workflow, Step, StepContext\n\ndef greet(ctx: StepContext) -> str:\n name = ctx.inputs.get(\"name\", \"World\")\n return f\"Hello, {name}!\"\n\ndef shout(ctx: StepContext) -> str:\n greeting = ctx.inputs[\"greet\"]\n return greeting.upper()\n\nworkflow = Workflow(\n name=\"hello\",\n steps=[\n Step(name=\"greet\", action=greet),\n Step(name=\"shout\", action=shout, depends_on=[\"greet\"]),\n ],\n)\n\nresult = workflow.run({\"name\": \"Kairos\"})\nprint(result.output) # \"HELLO, KAIROS!\"\n\n\n## 模块化架构\n\nKairos 的架构由六个核心模块组成：\n\n计划分解器（Plan Decomposer）：构建结构化的任务图，解析步骤依赖关系。\n\n步骤执行器（Step Executor）：管理步骤生命周期，支持超时、带抖动的重试和 foreach 扇出。\n\n状态存储（State Store）：带作用域的键值存储，支持大小限制和敏感键脱敏。\n\n契约注册表（Schema Registry）：支持 Kairos DSL、Pydantic 和 JSON Schema 的契约定义。\n\n验证引擎（Validation Engine）：在步骤间执行结构和语义验证。\n\n失败路由器（Failure Router）：基于策略的恢复机制，支持重试、重新规划、跳过和中止。\n\n## 安装与使用\n\nbash\npip install kairos-sdk\n\n\n可选依赖：\n\nbash\npip install kairos-sdk[pydantic] # Pydantic 契约支持\n\n\n## 项目状态与许可\n\nKairos 目前处于预发布阶段，架构设计已完成，正在遵循测试驱动开发（TDD）进行实现。项目采用 Apache 2.0 许可证开源。\n\n## 总结\n\nKairos 代表了 AI Agent 工作流框架的新方向——将安全性作为一等公民，而非事后考虑。通过契约验证、净化重试、最小权限访问、敏感信息脱敏等机制，Kairos 为生产环境中的 Agent 应用提供了坚实的安全基础。对于正在构建企业级 AI 应用的开发者来说，Kairos 是一个值得关注和尝试的框架。

AI Agent 工作流的安全困境\n\n随着大语言模型（LLM）能力的不断提升，基于 AI Agent 的自动化工作流正在成为企业应用的新范式。然而，在构建复杂的多步骤 Agent 工作流时，开发者往往面临一系列安全挑战：\n\n静默故障传播：一个步骤产生错误输出，却未被检测，导致错误在后续步骤中累积放大。\n\n上下文丢失：Agent 在执行长任务过程中丢失关键上下文，导致行为偏离预期。\n\n提示词注入风险：当步骤失败需要重试时，直接将原始错误信息或异常堆栈喂给模型，可能成为提示词注入的攻击向量。\n\n敏感信息泄露：日志中可能意外暴露 API 密钥、密码等敏感数据。\n\n缺乏恢复机制：当某个步骤失败时，系统缺乏有效的重试、跳过或重新规划策略。\n\n现有的编排工具（如 LangGraph、CrewAI）和验证工具（如 Guardrails AI）各有所长，但鲜有方案将编排与验证深度结合，并将安全作为一等公民来设计。Kairos 正是为填补这一空白而生。\n\nKairos 的核心设计理念\n\nKairos 的名字源自古希腊语"καιρός"，意为"恰当时机"——强调在正确的时间采取正确的行动。这一理念体现在其严格的执行循环设计中：\n\n\n目标 → 规划 → 执行步骤 → 验证输出 → 通过/重试/重新规划 → 下一步 → 完成\n\n\n与传统 Agent 框架不同，Kairos 将契约验证内建于每一步骤之间，确保数据质量在流程中始终保持。\n\n六大安全特性详解\n\n1. 步骤间契约验证（Inter-step Contract Validation）\n\n每个步骤都必须声明其输入和输出的数据结构契约。Kairos 支持多种契约定义方式：\n\npython\nfrom kairos import Schema\n\nschema = Schema({\n \"name\": str,\n \"products\": list[str],\n \"score\": float | None,\n})\n\n\n步骤执行完成后，系统会自动验证输出是否符合契约。只有验证通过的数据才会流入下一步，从根本上杜绝了错误数据的静默传播。\n\n2. 净化重试上下文（Sanitized Retry Context）\n\n这是 Kairos 最具创新性的安全特性之一。当步骤执行失败需要重试时，传统做法是将原始错误信息直接反馈给模型。然而，错误信息中可能包含恶意内容或敏感数据，直接注入提示词存在安全风险。\n\nKairos 的做法是：仅注入结构化的元数据（字段名、类型、尝试次数），原始 LLM 输出和异常消息永远不会进入提示词。这种设计有效防御了通过错误消息进行的提示词注入攻击。\n\n3. 最小权限状态访问（Scoped State Access）\n\n在多步骤工作流中，每个步骤默认可以访问整个状态空间，这违反了最小权限原则。Kairos 通过 read_keys 和 write_keys 参数，精确控制每个步骤可以读取和写入的状态键：\n\npython\nstep = Step(\n name=\"analyze\",\n action=analyze_fn,\n read_keys=[\"input_data\"],\n write_keys=[\"analysis_result\"]\n)\n\n\n这种设计不仅提升了安全性，还增强了代码的可读性和可维护性。\n\n4. 敏感键自动脱敏（Sensitive Key Redaction）\n\nKairos 内置了敏感键模式识别，自动识别包含 password、token、api_key 等关键词的状态项。这些敏感值在日志、导出和最终状态中都会被自动脱敏，防止意外泄露。\n\n5. 异常信息净化（Exception Sanitization）\n\n当步骤抛出异常时，Kairos 会在存储或记录之前自动剥离凭证、文件路径和原始堆栈跟踪信息。这确保了即使在调试过程中，敏感信息也不会暴露。\n\n6. 可配置失败策略（Configurable Failure Policies）\n\n不同的步骤可能需要不同的失败处理策略。Kairos 提供了灵活的策略配置：\n\npython\nfrom kairos import FailurePolicy, FailureAction\n\nstep = Step(\n name=\"critical_step\",\n action=critical_fn,\n failure_policy=FailurePolicy(\n on_validation_fail=FailureAction.RETRY,\n on_execution_fail=FailureAction.ABORT,\n max_retries=3,\n ),\n)\n\n\n支持的动作包括：\n- RETRY：重试执行\n- SKIP：跳过该步骤\n- ABORT：中止整个工作流\n- REPLAN：重新规划后续步骤\n\n策略支持三级继承：步骤级 → 工作流级 → Kairos 默认值，最具体的配置优先。\n\n与现有方案的对比\n\n| 需求 | LangGraph | CrewAI | Guardrails AI | Kairos |\n|------|-----------|--------|---------------|--------|\n| 多步骤工作流编排 | ✓ | ✓ | ✗ | ✓ |\n| 步骤间契约验证 | ✗ | 部分 | ✗（仅单输出） | ✓ |\n| 净化重试上下文 | ✗ | ✗ | N/A | ✓ |\n| 最小权限状态访问 | ✗ | ✗ | N/A | ✓ |\n| 敏感键脱敏 | ✗ | ✗ | N/A | ✓ |\n| 可配置失败策略 | 部分 | 部分 | N/A | ✓ |\n\nKairos 的独特价值在于将安全作为架构的核心考量，而非事后补丁。\n\n依赖关系与循环处理\n\nKairos 支持复杂的工作流拓扑，包括步骤间的依赖关系和循环处理：\n\npython\nworkflow = Workflow(\n name=\"competitive_analysis\",\n steps=[\n Step(name=\"fetch_competitors\", action=fetch_fn),\n Step(name=\"analyze_each\", action=analyze_fn,\n depends_on=[\"fetch_competitors\"],\n foreach=\"fetch_competitors\"),\n Step(name=\"summarize\", action=summarize_fn,\n depends_on=[\"analyze_each\"]),\n ],\n)\n\n\nforeach 支持基于前序步骤输出的动态循环，每个迭代独立执行并验证。\n\n模型无关设计\n\nKairos 不绑定任何特定的 LLM 提供商。任何接受 StepContext 参数的可调用对象都可以作为步骤动作——无论是调用 OpenAI API、本地 Ollama 模型，还是纯函数逻辑。\n\npython\nfrom kairos import Workflow, Step, StepContext\n\ndef greet(ctx: StepContext) -> str:\n name = ctx.inputs.get(\"name\", \"World\")\n return f\"Hello, {name}!\"\n\ndef shout(ctx: StepContext) -> str:\n greeting = ctx.inputs[\"greet\"]\n return greeting.upper()\n\nworkflow = Workflow(\n name=\"hello\",\n steps=[\n Step(name=\"greet\", action=greet),\n Step(name=\"shout\", action=shout, depends_on=[\"greet\"]),\n ],\n)\n\nresult = workflow.run({\"name\": \"Kairos\"})\nprint(result.output) \"HELLO, KAIROS!\"\n\n\n模块化架构\n\nKairos 的架构由六个核心模块组成：\n\n计划分解器（Plan Decomposer）：构建结构化的任务图，解析步骤依赖关系。\n\n步骤执行器（Step Executor）：管理步骤生命周期，支持超时、带抖动的重试和 foreach 扇出。\n\n状态存储（State Store）：带作用域的键值存储，支持大小限制和敏感键脱敏。\n\n契约注册表（Schema Registry）：支持 Kairos DSL、Pydantic 和 JSON Schema 的契约定义。\n\n验证引擎（Validation Engine）：在步骤间执行结构和语义验证。\n\n失败路由器（Failure Router）：基于策略的恢复机制，支持重试、重新规划、跳过和中止。\n\n安装与使用\n\nbash\npip install kairos-sdk\n\n\n可选依赖：\n\nbash\npip install kairos-sdk[pydantic] Pydantic 契约支持\n\n\n项目状态与许可\n\nKairos 目前处于预发布阶段，架构设计已完成，正在遵循测试驱动开发（TDD）进行实现。项目采用 Apache 2.0 许可证开源。\n\n总结\n\nKairos 代表了 AI Agent 工作流框架的新方向——将安全性作为一等公民，而非事后考虑。通过契约验证、净化重试、最小权限访问、敏感信息脱敏等机制，Kairos 为生产环境中的 Agent 应用提供了坚实的安全基础。对于正在构建企业级 AI 应用的开发者来说，Kairos 是一个值得关注和尝试的框架。