# Kairos：契约约束的 AI Agent 工作流安全编排框架 > Kairos 是一个安全强化的 Python SDK，通过契约验证、敏感信息脱敏和失败恢复机制，为 AI Agent 多步骤工作流提供生产级的安全保障。 - 板块: [Openclaw Llm](https://www.zingnex.cn/forum/board/openclaw-llm) - 发布时间: 2026-04-11T22:13:35.000Z - 最近活动: 2026-04-11T22:23:03.921Z - 热度: 112.8 - 关键词: AI Agent, 工作流编排, LLM安全, 契约验证, 提示词注入防护, Python SDK, 安全框架 - 页面链接: https://www.zingnex.cn/forum/thread/kairos-ai-agent - Canonical: https://www.zingnex.cn/forum/thread/kairos-ai-agent - Markdown 来源: ingested_event --- ## 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\n```python\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\n```python\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\n```python\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\n```python\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\n`foreach` 支持基于前序步骤输出的动态循环，每个迭代独立执行并验证。\n\n## 模型无关设计\n\nKairos 不绑定任何特定的 LLM 提供商。任何接受 `StepContext` 参数的可调用对象都可以作为步骤动作——无论是调用 OpenAI API、本地 Ollama 模型，还是纯函数逻辑。\n\n```python\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\n```bash\npip install kairos-sdk\n```\n\n可选依赖：\n\n```bash\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 是一个值得关注和尝试的框架。