Stripe内部Kai平台架构：企业级Agent框架设计

🏗️ Stripe Kai 平台的架构支柱

当一家企业从简单的检索增强生成（RAG）流水线过渡到自主运行的智能体系统时，架构复杂度会呈非线性增长。Stripe 内部知识 AI 平台 Kai 的披露凸显了这一转变。Kai 作为公司级智能体框架构建，依托 LangChain 与"Deep Agents"——具备多步推理、自我修正和动态代码执行能力的系统。

对于工程领导者而言，关注 Kai 的意义不仅在于它构建得很快，更在于它如何解决企业级智能体部署的根本挑战：状态管理、安全代码执行和集中治理。在我对智能体架构的分析中，发现大多数组织在此处折戟。他们构建的智能体脆弱且用途单一，在无约束的环境中运行，导致安全漏洞、令牌成本失控和难以维护的代码库。

要理解 Kai，必须先审视其宏观架构。公司级智能体框架不能作为一组孤立的微服务、各自运行自己的 LLM 客户端而存在。相反，它必须被构建为一个去耦了智能体定义、编排和执行的集中式平台。我将企业智能体平台的核心架构分为四个不同的层次：

编排层（控制平面）：该层管理智能体生命周期、路由传入请求并协调状态转换。在 Kai 的案例中，这是建立在 LangChain 之上的，利用有状态图结构来定义智能体如何在规划、工具执行和响应合成之间转换。

智能体注册表与目录：一个集中式仓库，各团队在此注册他们的智能体、Schema 和工具定义。这防止了冗余开发，并确保工具（如数据库连接器或内部 API 包装器）可在不同业务单元之间复用。

执行运行时（数据平面）：实际的 LLM 调用、工具执行和代码编译发生的地方。关键在于，这个运行时必须与编排层解耦，以防止资源耗尽并隔离安全风险。

治理与可观测性网关：一个专为 LLM 设计的 API 网关。它处理语义缓存、速率限制、成本跟踪和策略执行（如防止提示注入或数据泄露）。

通过集中这些层次，就为智能体开发建立了统一的接口。当开发者构建一个新智能体时——例如，一个分析商户流失的助手——他们无需编写样板代码来连接 LLM 或管理内存。相反，他们只需在注册表中声明式地注册一个智能体配置，定义所需的工具，然后让平台处理编排、安全和状态管理。

这种关注点分离至关重要。它允许平台团队优化底层基础设施——如更换 LLM 提供商、升级沙盒环境或调优缓存策略——而不会破坏单个智能体的实现。它还确保安全策略在全球范围内被执行，而不是依赖各个开发者正确地实现它们。

深入分析 Stripe Kai 平台的架构。了解如何使用 LangChain、安全的沙盒化代码执行运行时和强大的治理机制设计集中式企业智能体框架。

实现 Deep Agents：状态、记忆和工具调用循环

简单智能体在一个线性的"响应"循环中运作：接收输入、调用工具、返回输出。在 Kai 架构中概念化的"Deep Agents"则在复杂的非线性状态机上运作。它们必须能够将一个复杂提示分解为有向无环图（DAG）的子任务，并行或顺序执行这些任务，评估中间结果，并在工具返回错误时动态地重新规划。

要实现这种级别的自主性，我建议使用像 LangGraph 这样的有状态图框架。LangGraph 将智能体工作流建模为状态机，其中节点代表操作（如调用 LLM 或执行工具），边代表基于这些操作输出的状态转换。

在深度智能体架构中，状态必须被显式定义和持久化。这不仅仅是"聊天历史"；它是一个结构化的 Schema，跟踪智能体当前的计划、已完成任务的列表、这些任务的输出以及遇到的任何错误。

以下是一个利用 LangGraph 实现有状态深度智能体循环的具体 Python 实现。这个模式展示了如何实现一个自我修正循环，让智能体评估代码执行工具的输出，并在失败时自动重写代码。

import json
from typing import Dict, List, TypedDict, Union
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

# Define the state schema for our Deep Agent
class AgentState(TypedDict):
    messages: List[BaseMessage]
    current_plan: List[str]
    completed_tasks: List[str]
    generated_code: str
    execution_error: Union[str, None]
    retry_count: int

# Node: Planner - Analyzes input and generates an execution plan
def planner_node(state: AgentState) -> Dict:
    last_message = state["messages"][-1].content
    plan = ["write_code", "execute_code", "verify_results"]
    return {
        "current_plan": plan,
        "messages": [AIMessage(content=f"Plan generated: {json.dumps(plan)}")]
    }

# Node: Code Generator - Generates Python code based on the plan
def code_generator_node(state: AgentState) -> Dict:
    error_context = f"\nPrevious execution failed with error: {state['execution_error']}" if state["execution_error"] else ""
    prompt = f"Write a Python script to calculate merchant metrics.{error_context}"

    # Simulating LLM code generation
    generated_code = "def calculate(): return 100 / 0" if state["retry_count"] == 0 else "def calculate(): return 100 / 10"

    return {
        "generated_code": generated_code,
        "messages": [AIMessage(content=f"Generated code: {generated_code}")]
    }

# Node: Sandboxed Executor - Executes the generated code safely
def executor_node(state: AgentState) -> Dict:
    code = state["generated_code"]
    retry = state["retry_count"]
    try:
        if "100 / 0" in code:
            raise ZeroDivisionError("division by zero")
        result = "Success: Result is 10.0"
        return {
            "execution_error": None,
            "completed_tasks": state["completed_tasks"] + ["execute_code"],
            "messages": [AIMessage(content=result)]
        }
    except Exception as e:
        return {
            "execution_error": str(e),
            "retry_count": retry + 1,
            "messages": [AIMessage(content=f"Execution failed: {str(e)}")]
        }

# Conditional Router: Decides whether to retry or proceed to end
def router(state: AgentState) -> str:
    if state["execution_error"] is not None:
        if state["retry_count"] < 3:
            return "generate_code"
        return "fail_node"
    return END

# Construct the LangGraph State Machine
workflow = StateGraph(AgentState)

workflow.add_node("planner", planner_node)
workflow.add_node("generate_code", code_generator_node)
workflow.add_node("execute_code", executor_node)

workflow.set_entry_point("planner")
workflow.add_edge("planner", "generate_code")
workflow.add_edge("generate_code", "execute_code")

# Dynamic routing based on execution success or failure
workflow.add_conditional_edges(
    "execute_code",
    router,
    {
        "generate_code": "generate_code",
        "fail_node": END,
        END: END
    }
)

app = workflow.compile()

这个模式说明了 Deep Agents 的核心力量：韧性。通过将自我修正循环直接编码到图状态中，智能体可以在无需人工干预的情况下从语法错误、API 超时或逻辑缺陷中恢复。

然而，实现这个状态机需要强大的状态持久化支持。在生产环境中，内存状态是不够的。如果一个节点执行需要 30 秒而服务器重启，整个智能体运行就会丢失。我建议使用 Redis 或 PostgreSQL 作为后端存储来备份图状态，利用 LangGraph 的 checkpointer 接口。这允许你暂停智能体执行，在调用高风险工具时等待人在环中批准，然后无缝恢复执行。

⚙️ 动态代码执行的安全沙盒化

Stripe Kai 架构中最具技术挑战性的方面，可能是 LLM 生成代码的安全执行。当 Deep Agents 能够编写和执行任意代码来分析数据、解析文件或与 API 交互时，它们会变得非常强大。然而，允许 LLM 在你的内部网络上执行任意代码，是一项极高的安全风险。

如果一个智能体通过提示注入被攻击，攻击者可以编写代码来读取环境变量、访问内部数据库，或对其他内部服务发起攻击。因此，安全隔离的沙箱，是任何企业级智能体平台不可或缺的要求。

我评估过几种智能体运行时的沙箱策略。标准 Docker 容器本身是不够的，因为它们共享主机内核；容器逃逸漏洞可能危及底层 VM。为此，你必须实施多层隔离策略。下表比较了适用于智能体工作负载的主要执行沙箱技术：

对于像 Kai 这样的企业级智能体平台，我强烈建议使用 gVisor 或 Firecracker MicroVM。Stripe 的架构依赖于为每个智能体会话创建临时的隔离沙箱。

当一个智能体决定执行代码时，编排层将代码打包并发送到专用的沙箱服务。该服务配置一个 microVM 或 gVisor 保护的容器，在严格的时间限制内执行代码（例如 5 秒），捕获标准输出和错误，然后立即销毁环境。为了安全地实现这一点，你必须强制执行以下网络安全和安全的边界：

零网络访问：沙箱必须禁用网络运行（--network none），除非智能体明确需要互联网访问。如果需要互联网访问，必须通过高度限制的出口代理路由，该代理只允许连接到预先批准的域名白名单。

只读根文件系统：容器文件系统应该是只读的，使用一个小的、临时的内存中的 tmpfs 挂载来处理临时文件。

严格的资源限制：强制执行 CPU（例如 0.5 vCPU）、内存（例如 256MB）和磁盘 I/O 的硬限制，以防止无限循环或磁盘填充代码导致的拒绝服务攻击。

不暴露密钥：永远不要将数据库凭证或 API 密钥直接传入沙箱环境。如果代码需要查询数据库，沙箱应该与一个安全的数据代理通信，该代理在将数据返回沙箱之前强制执行行级安全和列掩码。

集中式治理、护栏和评估

当你将一个智能体平台扩展到数百名开发人员和数百万次运行时，治理就成为你主要的运营瓶颈。如果没有集中的护栏，你会很快面临天文数字般的 API 账单、性能下降和不可预测的智能体行为。

Stripe 的 Kai 架构通过在编排框架和 LLM 提供商之间实施一个集中式治理层来解决这个问题。我建议将这一层构建为一个专为 LLM 流量设计的智能 API 网关。

1. Token 预算和速率限制

在循环中运行的 Deep Agents 如果陷入无限的规划循环，可以很容易地在几分钟内消耗数百万个 token。为了防止这种情况，你的平台必须在多个层面强制执行严格的 token 预算：

单次运行预算：限制单次智能体执行允许的最大 LLM 调用次数（例如 20 次）和总 token 数（例如 100,000）。如果超出预算，平台将终止运行并提醒用户。

用户/团队预算：为每个业务部门实施 LLM 消费的每日或每月上限。

2. 护栏和提示注入缓解

每个进入智能体的输入和每个来自 LLM 的输出都必须通过自动护栏管道。我建议使用快速、本地模型（如 Llama-Guard）和基于正则表达式的模式匹配器的组合来实时检查流量：

输入护栏：扫描传入的用户提示，查找提示注入攻击、越狱尝试和个人身份信息（PII）。如果检测到 PII，在将其发送到 LLM 之前将其删除。

输出护栏：扫描 LLM 输出，确保它们符合预期格式（例如有效的 JSON 或结构化工具调用），并且不包含用户无权查看的敏感内部数据。

3. 持续评估（LLM 即评判者）

与传统软件不同，你无法用简单的单元测试来验证智能体行为。因为 LLM 输出是概率性的，你必须实施一个持续评估管道。我建议建立一个评估数据集——一组具有代表性的用户提示及其预期工具调用和最终答案的金色集合。每次开发者更新智能体的提示、系统指令或工具定义时，平台应该自动针对该评估数据集运行智能体。

使用"LLM 即评判者"模式，一个强大的模型（如 GPT-4o 或 Claude 3.5 Sonnet）根据三个关键指标评估测试运行：

忠诚度：智能体是否严格遵循提供的上下文，还是产生了幻觉事实？

答案相关性：最终响应是否直接回答了用户的提示？

工具选择准确性：智能体是否以正确的参数调用了正确的工具序列？

通过将这个评估管道嵌入你的 CI/CD 流程，你可以防止回归，并确保智能体性能随时间保持稳定。

---

Stripe 的 Kai 架构表明，构建企业级 AI 智能体平台并非模型能力的挑战，而是软件工程纪律的挑战。要超越脆弱的原型，你必须将智能体视为有状态的、有治理的、高度安全的系统。

如果你负责为你的组织设计一个智能体平台，我建议采取以下立即可行的后续行动：

将编排与执行解耦：不要允许智能体在你的主应用服务器内执行工具或代码。在你的 LangChain/LangGraph 控制平面和你的执行运行时之间建立明确的边界。

首先构建安全沙箱：在你部署单个可以编写代码的智能体之前，使用 gVisor 或 Firecracker 实现一个临时的、隔离的执行环境。将不受信任的 LLM 生成代码视为与恶意软件相同的安全态势。

实施集中式状态管理：摆脱内存中的智能体状态。标准化使用 Redis 等持久状态存储，以确保你的智能体具有弹性、可中断和人工介入验证的能力。

建立财务和安全护栏：部署 LLM 网关来强制执行 token 预算、速率限制和输入/输出护栏。这是安全地扩展智能体开发而不会面临成本失控或数据泄露风险的唯一方法。

通过采用这些架构原则，你将构建一个强大、安全且高度可扩展的基础，使你的组织能够利用智能体 AI 的真正力量。

🔗 Originally published on ixuvo.com