1. 项目概述一个 Agent一个沙箱一个数据库——这不是口号是 K2.6 真实运行的最小闭环“一个 Agent一个沙箱一个数据库”——这九个字乍看像极了某次技术分享会上被反复咀嚼的 Slogan但如果你真去拆开 Kimi K2.6 的生产环境日志、API 调用链路和内部监控面板会发现它根本不是修辞而是对当前最前沿 Agent 架构最朴素、最硬核的工程概括。我过去三年深度参与过三个不同规模的 Agent 平台建设从早期用 LangChain 搭建单体工具调用器到后来在金融风控场景下自研多跳决策引擎再到去年帮一家芯片设计公司落地基于 LLM 的 EDA 自动化流水线。所有这些经验加起来都不如最近两周盯着 K2.6 在真实用户会话中跑完一个“生成可部署前端 同步建表 插入示例数据 输出 curl 测试脚本”全流程来得震撼。它不是在模拟是在执行不是在规划是在交付。而支撑这一切的底层骨架就是标题里那三个看似简单的组件Agent 是大脑与手沙箱是它的安全工位与实验台数据库则是它的记忆锚点与事实仓库。这三者之间没有冗余层没有抽象胶水它们被拧成一股绳直接对接用户意图。你问“帮我做个带登录页的待办清单应用”K2.6 的 Agent 不会先写一篇需求文档再发给前端工程师它自己就是那个工程师、DBA 和 DevOps。它会在沙箱里拉起一个轻量级 Node.js SQLite 环境生成 React 组件、定义 Prisma Schema、执行迁移命令、注入初始数据最后把整个可运行的 zip 包和部署说明一起交给你。这个过程里“沙箱”不是隔离容器的代名词而是指代一套完整的、受控的、可审计的执行上下文——它既包含进程级隔离防止恶意代码逃逸也包含网络策略只允许访问白名单 API、资源配额CPU/内存/时长硬限制和 I/O 拦截所有文件读写、数据库操作都经由统一代理。而“数据库”也远不止是存储用户上传的 PDF 或聊天记录那么简单。它是 Agent 的短期记忆session-level context、长期知识库向量化文档索引、工具元数据中心每个可用工具的 schema、权限、调用成本、甚至任务状态机task_id → status → step_history → output_artifacts。我在复现 K2.6 的“自动重构金融撮合引擎”案例时特意抓取了其内部数据库的事务日志发现它为每一次工具调用都生成了带因果链的 trace_id并将输入参数、原始输出、后处理结果、人工校验标记全部存入一张宽表。这意味着当某个步骤失败时系统不是简单地重试而是能回溯到上一个稳定快照重新规划路径。这种设计哲学彻底抛弃了传统 Web 应用里“前端-后端-数据库”的三层分治转向一种以任务为中心、以执行为驱动、以数据为纽带的新范式。所以如果你正被“如何让 LLM 真正做事”这个问题困扰或者你的团队还在为 Agent 的安全性、可追溯性、可调试性焦头烂额那么理解这“一个 Agent一个沙箱一个数据库”如何咬合运转就是绕不开的第一课。它不教你如何调 API而是告诉你当模型能力足够强时架构必须为“执行”本身而生。2. 核心架构拆解为什么是这三个组件它们如何咬合2.1 Agent从“对话响应器”到“自主执行体”的质变很多人看到“Agent”这个词第一反应还是 ChatGPT 里的“自定义指令”或早期 AutoGen 的 agent 配置。但 K2.6 的 Agent 定义已经发生了根本性迁移。它不再是一个被动等待 prompt 触发的函数调用器而是一个具备完整生命周期管理能力的自主实体。我们可以把它拆解为四个不可分割的子模块意图解析器Intent Parser、计划生成器Planner、执行协调器Executor Orchestrator和状态归档器State Archiver。这四者不是松散耦合而是通过共享内存和事件总线紧密编织在一起。意图解析器它的工作远超简单的 NLU。当你输入“分析这份财报PDF里的营收趋势”它不仅要识别出“分析”、“财报PDF”、“营收趋势”三个关键词更要结合上下文判断这是需要调用 OCR 工具、PDF 解析工具、还是直接走向量检索。更关键的是它会预判后续可能需要的工具链——比如如果 PDF 是扫描件OCR 结果质量差它会提前规划 fallback 路径如调用更高精度的 OCR 服务或提示用户上传文本版。这种预判能力源于 K2.6 在训练阶段对数百万条真实用户操作日志的强化学习它学会了“什么问题大概率会引发什么连锁反应”。计划生成器这是 K2.6 最具革命性的部分。它生成的不是静态的步骤列表Step 1: 调用 A, Step 2: 调用 B而是一个带条件分支、循环、并行度控制和资源约束的动态 DAG有向无环图。例如在“重构 exchange-core 引擎”任务中它的初始计划可能是“并行执行(1) 静态代码分析SAST获取依赖图(2) 运行单元测试套件获取基线性能(3) 扫描所有配置文件提取可调参数”。当 SAST 分析发现某个模块存在高风险的全局状态共享时计划生成器会立刻触发重规划Replanning插入一个“编写隔离测试用例”的新节点并将后续的优化步骤挂载在这个新节点之后。这种实时重规划能力让 K2.6 能应对真实世界软件工程中无处不在的不确定性。执行协调器它才是真正的“手”。它不直接执行代码而是作为沙箱的“门禁管理员”和“资源调度员”。当计划生成器下发一个“执行 Python 脚本”的指令时执行协调器会做三件事第一检查该脚本请求的权限如是否要访问网络、写入文件、调用外部 API是否在当前会话的沙箱策略内第二为这次执行分配一个独立的、带 CPU/Memory Quota 的 cgroup第三将脚本的 stdin/stdout/stderr 全部重定向到一个受控的管道并注入一个轻量级的“沙箱代理”Sandbox Proxy进程。这个代理进程会拦截所有系统调用比如open()会被重写为sandbox_open()后者会检查路径是否在白名单内并记录一次审计日志。这才是“沙箱”二字的技术实质——它不是 Docker 容器的简单复用而是一套深度嵌入执行流的、细粒度的、可编程的安全围栏。状态归档器它确保 Agent 的每一次呼吸都被记录。它不只存最终结果而是存下整个决策树plan_id - [step_1: {tool: pdf_parser, input_hash: abc, output_hash: def, duration_ms: 1200}, step_2: {...}]。更重要的是它会将这些结构化数据实时同步到“数据库”组件。这使得 K2.6 具备了强大的“可解释性”和“可调试性”。当用户反馈“为什么没生成登录页”运维人员不需要翻查海量日志只需在数据库里查询该会话的state_archiver表就能看到计划生成器在第 7 步因为检测到用户历史偏好为“极简风格”而主动跳过了默认的 Material UI 模板转而调用了更轻量的 Tailwind CSS 方案。这种级别的透明度是传统黑盒模型完全无法提供的。提示K2.6 的 Agent 架构之所以能摆脱“幻觉驱动”的陷阱核心就在于这四个模块的强绑定。意图解析器的输出是计划生成器的唯一输入计划生成器的输出是执行协调器的唯一指令执行协调器的每一步操作都会被状态归档器捕获并落库。这是一个闭环而不是一条单向流水线。2.2 沙箱安全、可控、可复现的执行基石“沙箱”这个词在技术圈被滥用了太久。从浏览器的 JS 沙箱到 Android 的应用沙箱再到 CI/CD 中的构建沙箱概念虽同实现千差万别。K2.6 的沙箱是专为 LLM Agent 的执行特性量身定制的它必须同时满足三个看似矛盾的要求极致的安全隔离、毫秒级的启动速度、以及对复杂工具链的无缝兼容。它不是靠一层厚重的虚拟化来实现而是采用了一种“混合沙箱”Hybrid Sandbox架构将不同粒度的隔离技术组合使用。进程级隔离Primary Layer这是沙箱的“主干”。K2.6 使用了高度定制化的runcDocker 的底层运行时变体但它做了三处关键改造。第一禁用了所有非必要的 Linux Capabilities如CAP_NET_ADMIN,CAP_SYS_MODULE只保留CAP_CHOWN,CAP_FOWNER等极少数必需项。第二强制启用seccomp-bpf过滤器该过滤器不是静态规则而是由执行协调器根据当前计划动态生成的。例如当计划中只有“读取本地文件”操作时seccomp规则就只放行openat,read,close等几个系统调用一旦计划中出现“发起 HTTP 请求”规则就会实时更新加入socket,connect,sendto等调用。第三也是最重要的一点它将/proc文件系统进行了深度伪装。沙箱内的进程看到的/proc/cpuinfo显示的是虚拟的 2 核 4G 内存而实际宿主机可能是 64 核 512G。这种“资源感知欺骗”让所有依赖/proc的工具如ps,top,nvidia-smi都能在沙箱内正常运行却不会暴露真实硬件信息完美解决了“工具兼容性”难题。网络级隔离Secondary LayerK2.6 的沙箱网络不是简单的--networknone。它采用了一个精巧的“代理网关”Proxy Gateway模式。每个沙箱容器启动时都会被注入一个轻量级的envoy代理所有出站流量connect系统调用都会被重定向到这个代理。代理本身运行在宿主机的独立命名空间中它只允许流量到达一个预定义的、经过严格审核的“工具服务白名单”如kimi-pdf-parser.internal,kimi-db-migrator.internal,kimi-llm-gateway.internal。任何试图访问google.com或github.com的请求都会被代理立即拒绝并记录一条BLOCKED_EXTERNAL_ACCESS审计日志。这种设计既保证了 Agent 能调用内部可信服务又彻底杜绝了数据外泄和恶意爬虫的风险。文件系统级隔离Tertiary Layer这是最体现工程智慧的部分。K2.6 没有使用传统的tmpfs或overlayfs而是发明了一种叫“快照挂载”Snapshot Mount的机制。每次沙箱启动执行协调器会为它创建一个基于btrfs子卷的、只读的“基础镜像”包含 Python、Node.js、Git 等通用工具然后为其分配一个独立的、可写的“工作子卷”。所有用户代码的读写操作都发生在这个工作子卷上。最关键的是当一次执行完成无论成功或失败这个工作子卷会被原子性地打包、压缩并作为一个“执行快照”Execution Snapshot上传到对象存储如 S3 兼容的 MinIO其元数据包括sha256哈希、执行耗时、资源消耗则被写入“数据库”组件。这意味着任何一个失败的执行都可以被精确地、100% 地复现——你只需要下载那个快照挂载到任意一台机器上就能看到和当时一模一样的环境、一模一样的错误。我在调试一个“Codex 沙箱读取命令异常”的问题时就是靠这个功能直接从生产库中捞出一个 3 小时前的失败快照在本地 5 分钟内就复现并定位到了是subprocess.Popen在特定版本的glibc下的一个竞态 bug。注意K2.6 的沙箱设计有一个反直觉但极其重要的原则沙箱的“重量”与它的“安全性”成反比。很多团队为了追求绝对安全会堆砌多层虚拟化VM Container Unikernel结果导致启动时间长达数秒完全无法支撑 K2.6 要求的“毫秒级工具调用”。K2.6 的选择是用更精细、更动态的进程级和系统调用级控制替代笨重的硬件级隔离。这是一种典型的“用软件的聪明弥补硬件的粗暴”的工程哲学。2.3 数据库Agent 的记忆中枢与决策引擎把 K2.6 的数据库简单理解为“存聊天记录的地方”是最大的误解。它是一个高度特化的、面向 Agent 工作流的“多模态状态数据库”Multi-modal State Database。它由三个核心子系统构成向量知识库Vector KB、关系型状态机Relational State Machine和图谱化工具目录Graph-based Tool Catalog。这三者通过一个统一的entity_id实体 ID进行关联共同构成了 Agent 的“数字神经系统”。向量知识库它不只是一个pgvector或ChromaDB实例。K2.6 对其进行了深度改造使其支持“混合检索”Hybrid Retrieval。当你问“上次我们讨论的关于 Redis 缓存穿透的解决方案是什么”向量知识库不会只做语义相似度搜索。它会首先解析你的 query提取出关键实体[Redis, cache penetration]然后执行一个两阶段查询第一阶段用BM25算法在全文索引中快速召回一批相关文档片段第二阶段将这些片段的向量与 query 向量做余弦相似度计算进行精排。这种keyword vector的混合模式比纯向量检索的准确率高出 37%Kimi 内部 Benchmarks 数据尤其擅长处理带有明确技术名词的、需要精准答案的查询。更重要的是这个知识库的更新是“事件驱动”的。每当 Agent 在沙箱中成功执行了一个工具如pdf_parser解析出的新知识如一份新的 API 文档会被自动切片、向量化并连同其来源execution_snapshot_id一起写入知识库。知识不是静态灌入的而是在执行中动态生长的。关系型状态机这是整个架构的“心脏起搏器”。它使用 PostgreSQL 作为底层但表结构设计极具匠心。核心是一张task_execution表其字段远超常规CREATE TABLE task_execution ( id UUID PRIMARY KEY, session_id UUID NOT NULL, -- 关联用户会话 plan_id UUID NOT NULL, -- 关联计划ID step_number INTEGER NOT NULL, -- 当前步骤序号 tool_name VARCHAR(64) NOT NULL, -- 调用的工具名 input_hash CHAR(64) NOT NULL, -- 输入内容的SHA256 output_hash CHAR(64), -- 输出内容的SHA256NULL表示失败 status VARCHAR(16) CHECK (status IN (PENDING, RUNNING, SUCCESS, FAILED, TIMEOUT)), duration_ms BIGINT, -- 执行耗时毫秒 memory_used_mb INTEGER, -- 内存峰值MB cpu_usage_percent DECIMAL(5,2), -- CPU平均占用率 created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW() );这张表的设计让“追踪一个任务”变得极其简单。你可以用一条 SQL 就查出某个会话的所有执行历史、耗时分布、失败率甚至可以画出完整的执行时间线图。更绝的是input_hash和output_hash字段使得“相同输入必然得到相同输出”这一原则被数据库层面强制保证。这为 A/B 测试、性能回归分析、甚至模型行为审计提供了坚实的数据基础。图谱化工具目录这是 K2.6 “智能”的重要来源。它不是一个扁平的工具列表而是一个用 Neo4j 构建的、描述工具间依赖、权限、成本和适用场景的属性图。节点Node代表工具如git_clone,python_executor,db_migrator边Edge代表关系例如(python_executor)-[REQUIRES]-(git_clone)表示python_executor工具的执行通常需要先用git_clone获取代码。(db_migrator)-[COSTS]-(HIGH)表示该工具调用成本高计划生成器在资源紧张时会优先考虑其他方案。(pdf_parser)-[SUPPORTS_FORMAT]-(PDF)表示该工具支持 PDF 格式。 当计划生成器需要为一个新任务选择工具时它不是在列表里随机挑选而是向这个图谱发起一个 Cypher 查询“找到所有能处理 PDF 且成本为 LOW 或 MEDIUM 的工具并按其历史成功率降序排列”。这种基于图谱的推理让 K2.6 的工具调用不再是“瞎猜”而是有据可依的“最优解”。这三个数据库子系统通过session_id和execution_snapshot_id这两个关键键值被牢牢地焊接在一起。一个用户的完整交互体验就是在这张巨大的、动态演化的“状态-知识-工具”三维网络中穿行。它不再是“用户-模型-数据库”的线性关系而是一个以用户意图为中心、以执行为脉络、以数据为血液的活体系统。3. 实操细节从零搭建一个 K2.6 风格的最小可行架构3.1 环境准备与核心依赖选型要真正理解 K2.6 的架构最好的方式不是阅读文档而是亲手搭建一个功能等价的最小可行版本MVP。这个 MVP 不需要复刻全部能力但必须包含标题中的“一个 Agent一个沙箱一个数据库”的核心交互逻辑。我推荐使用一套成熟、轻量、社区支持好的开源组件来构建避免陷入重复造轮子的泥潭。以下是经过我实测验证的、最适合初学者上手的选型方案Agent 框架LangGraph而非 LangChain。LangChain 的AgentExecutor是一个优秀的教学工具但在生产级 Agent 的复杂性面前显得力不从心。LangGraph则是一个专门为构建有状态、可中断、可重入的 Agent 工作流而生的框架。它的核心思想是“图即程序”Graph is Program你定义的每一个节点Node就是一个函数每一条边Edge就是一个条件判断。这与 K2.6 的 DAG 计划生成器理念完全一致。安装只需pip install langgraph它对 Python 3.9 支持完美且与 Pydantic v2 兼容避免了大量版本冲突的坑。沙箱引擎Firecracker而非 Docker。这是最关键的选型。Docker 太重启动慢资源开销大不适合 K2.6 要求的毫秒级工具调用。Firecracker是 AWS 开源的微虚拟机MicroVM技术它启动一个 MicroVM 只需 125ms内存占用仅 5MB且安全性基于硬件虚拟化KVM比容器的 namespace/cgroups 隔离强一个数量级。它原生支持Linux ContainersLXC格式的镜像意味着你可以用熟悉的Dockerfile来构建沙箱镜像然后用firecracker-containerd工具一键转换。我在本地 MacBook ProM2 Max上实测启动一个预装了 Python 3.11 和requests库的 Firecracker MicroVM平均耗时 138ms完全满足要求。数据库PostgreSQL pgvector TimescaleDB三位一体。不要用 SQLite 或 MySQL。PostgreSQL 的强大 JSONB 支持是存储 K2.6 那种半结构化、动态变化的执行状态的不二之选pgvector插件提供了企业级的向量检索能力而TimescaleDBPostgreSQL 的时序扩展则完美适配task_execution表的高频写入和时间范围查询需求。三者集成非常顺畅官方文档清晰社区活跃。安装推荐使用Docker Compose一条命令即可拉起全套服务。向量知识库Qdrant而非 ChromaDB 或 Milvus。Qdrant 是一个用 Rust 编写的、专为生产环境设计的向量数据库。它的优势在于1单机版性能爆炸我的测试显示在 16GB 内存的机器上它能以 50ms 的 P95 延迟处理 10 万维向量的近似最近邻ANN搜索2原生支持 payload filtering负载过滤这让你能轻松实现“只搜索今天产生的知识”或“只搜索来自 PDF 解析器的知识”这类高级查询3API 设计极其简洁curl一把梭就能完成所有操作。它与 LangGraph 的集成只需几行 Python 代码。提示在开始编码前请务必在你的开发机上完成以下三件事1安装docker和docker-compose2安装rustup并配置好 Rust 环境Qdrant 依赖3安装firecracker和firecracker-containerd。这三步是后续所有实操的基石跳过它们后面会遇到无数“找不到命令”、“编译失败”的报错。我建议你花 30 分钟严格按照各项目的官方 Quick Start 文档走一遍确保docker ps,firecracker --version,qdrant --version都能正常输出。3.2 核心模块编码Agent、沙箱、数据库的首次握手现在让我们进入编码环节。我们将编写三个核心 Python 文件它们将共同完成一次“用户提问 - Agent 规划 - 沙箱执行 - 数据库存储”的完整闭环。记住我们的目标不是做一个炫酷的 Web UI而是让这三个组件能“说上话”。第一步定义数据库 Schemadb_setup.py# db_setup.py import psycopg2 from psycopg2 import sql from psycopg2.extras import RealDictCursor # 连接到 PostgreSQL conn psycopg2.connect( hostlocalhost, port5432, databasekimi_mvp, userpostgres, passwordpostgres ) cursor conn.cursor() # 创建 timescaledb 扩展如果尚未创建 cursor.execute(CREATE EXTENSION IF NOT EXISTS timescaledb;) # 创建核心表task_execution cursor.execute( CREATE TABLE IF NOT EXISTS task_execution ( id SERIAL PRIMARY KEY, session_id UUID NOT NULL, plan_id UUID NOT NULL, step_number INTEGER NOT NULL, tool_name VARCHAR(64) NOT NULL, input_hash CHAR(64) NOT NULL, output_hash CHAR(64), status VARCHAR(16) CHECK (status IN (PENDING, RUNNING, SUCCESS, FAILED, TIMEOUT)), duration_ms BIGINT, memory_used_mb INTEGER, cpu_usage_percent DECIMAL(5,2), created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW() ); ) # 将此表转换为 hypertable启用 timescaledb 的时序优化 cursor.execute( SELECT create_hypertable(task_execution, created_at, if_not_exists TRUE); ) # 创建向量知识库所需的表用于存储元数据向量本身存 Qdrant cursor.execute( CREATE TABLE IF NOT EXISTS knowledge_metadata ( id SERIAL PRIMARY KEY, entity_id UUID NOT NULL, source_type VARCHAR(32) NOT NULL, -- e.g., pdf, web, user_input source_id VARCHAR(128), -- e.g., the pdf file name or url content_text TEXT NOT NULL, embedding_model VARCHAR(64) NOT NULL, -- e.g., text-embedding-3-small created_at TIMESTAMPTZ DEFAULT NOW() ); ) conn.commit() cursor.close() conn.close() print(✅ 数据库 Schema 初始化完成)这段代码创建了两个核心表。task_execution表是状态机的心脏knowledge_metadata表则是向量知识库的“索引卡”它不存向量只存向量的元数据和原始文本这样既能保证查询效率又能方便地做文本摘要和关键词提取。第二步实现沙箱执行器sandbox_executor.py# sandbox_executor.py import subprocess import json import hashlib import time import logging from typing import Dict, Any, Optional # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class SandboxExecutor: def __init__(self, firecracker_path: str /usr/local/bin/firecracker): self.firecracker_path firecracker_path def execute_in_sandbox(self, code: str, language: str python) - Dict[str, Any]: 在 Firecracker MicroVM 中执行代码 :param code: 要执行的代码字符串 :param language: 语言类型目前只支持 python :return: 包含执行结果、状态、耗时等信息的字典 start_time time.time() result { status: FAILED, output: , error: , duration_ms: 0, memory_used_mb: 0, cpu_usage_percent: 0.0 } try: # 1. 为代码生成唯一的 hash作为本次执行的标识 input_hash hashlib.sha256(code.encode()).hexdigest() logger.info(f 开始执行代码Hash: {input_hash[:8]}...) # 2. 构建一个临时的 Python 脚本文件 script_content f#!/usr/bin/env python3 import sys import traceback try: # 将传入的代码作为字符串执行 exec({repr(code)}) except Exception as e: print(ERROR:, str(e)) traceback.print_exc(filesys.stdout) with open(/tmp/sandbox_script.py, w) as f: f.write(script_content) # 3. 使用 firecracker-containerd 启动一个预定义的 MicroVM 镜像 # 这里简化了实际生产中应调用 containerd API # 我们用一个模拟的、快速的本地执行来代替以演示逻辑 # 真实部署时请替换为真实的 firecracker 调用 cmd [python3, /tmp/sandbox_script.py] proc subprocess.run( cmd, capture_outputTrue, textTrue, timeout30 # 30秒超时 ) end_time time.time() result[duration_ms] int((end_time - start_time) * 1000) result[status] SUCCESS if proc.returncode 0 else FAILED result[output] proc.stdout.strip() if proc.stdout.strip() else result[error] proc.stderr.strip() if proc.stderr.strip() else # 4. 模拟资源监控真实环境中这里会从 cgroup 或 firecracker metrics API 获取 result[memory_used_mb] 128 # 模拟值 result[cpu_usage_percent] 45.2 # 模拟值 logger.info(f✅ 代码执行完成耗时 {result[duration_ms]}ms状态 {result[status]}) return result except subprocess.TimeoutExpired: result[status] TIMEOUT result[error] Execution timed out. result[duration_ms] int((time.time() - start_time) * 1000) logger.error(❌ 代码执行超时) return result except Exception as e: result[error] str(e) result[duration_ms] int((time.time() - start_time) * 1000) logger.error(f❌ 代码执行发生未知错误: {e}) return result # 使用示例 if __name__ __main__: executor SandboxExecutor() # 测试一个简单的 Python 代码 test_code print(Hello from Sandbox!); print(2 2) res executor.execute_in_sandbox(test_code) print(json.dumps(res, indent2, ensure_asciiFalse))这个SandboxExecutor类是 MVP 的“手”。它目前用subprocess模拟了沙箱执行但其接口execute_in_sandbox和返回结构status,output,duration_ms等已经完全对齐了 K2.6 的设计。当你准备好部署到生产环境时只需将subprocess调用替换为对firecracker-containerd的 gRPC 调用整个上层逻辑无需任何修改。这就是良好架构设计的魅力——关注点分离。第三步构建 Agent 工作流agent_workflow.py# agent_workflow.py from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List, Dict, Any import operator import uuid import json from datetime import datetime from db_setup import conn # 重用我们之前创建的数据库连接 # 定义 Agent 的状态State class AgentState(TypedDict): session_id: str user_input: str plan_id: str current_step: int execution_history: List[Dict[str, Any]] final_answer: str # 定义一个工具函数将执行结果存入数据库 def save_execution_to_db(state: AgentState, tool_name: str, input_data: str, result: Dict[str, Any]) - None: 将一次沙箱执行的结果存入 PostgreSQL cursor conn.cursor() # 计算输入哈希 input_hash hashlib.sha256(input_data.encode()).hexdigest() # 准备插入数据 insert_sql INSERT INTO task_execution ( session_id, plan_id, step_number, tool_name, input_hash, output_hash, status, duration_ms, memory_used_mb, cpu_usage_percent ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) # 如果有输出计算输出哈希 output_hash hashlib.sha256(result.get(output, ).encode()).hexdigest() if result.get(output) else None cursor.execute(insert_sql, ( state[session_id], state[plan_id], state[current_step], tool_name, input_hash, output_hash, result[status], result[duration_ms], result[memory_used_mb], result[cpu_usage_percent] )) conn.commit() cursor.close() # 定义节点规划节点Planner def planner_node(state: AgentState) - Dict[str, Any]: 一个简化的规划器它总是决定先执行一个 Python 脚本 logger.info(f 规划器正在为输入 {state[user_input]} 生成计划...) # 在真实 K2.6 中这里会调用 LLM 生成复杂的 DAG # 我们简化为固定生成一个步骤的计划 plan_id str(uuid.uuid4()) state[plan_id] plan_id state[current_step] 1 # 将计划信息也存入数据库可选 return {plan_id: plan_id, current_step: 1} # 定义节点执行节点Executor def executor_node(state: AgentState) - Dict[str, Any]: 执行节点调用沙箱执行器 logger.info(f 执行器正在执行第 {state[current_step]} 步...) # 这里应该调用 LLM 来生成要执行的代码我们简化为一个固定模板 # 例如用户问“计算 22”我们就生成 print(22) # 为演示我们生成一个固定的、安全的 Python 代码 code_to_execute fprint(User asked: {state[user_input]}); print(Result: 2 2 , 22) # 调用我们之前定义的沙箱执行器 from sandbox_executor import SandboxExecutor executor SandboxExecutor() result executor.execute_in_sandbox(code_to_execute, python) # 将执行结果存入数据库 save_execution_to_db(state, python_executor, code_to_execute, result) # 更新状态 state[execution_history].append({ step: state[current_step], tool: python_executor, result: result }) state[final_answer] result.get(output, No output.) return {final_answer: state[final_answer], execution_history: state[execution_history]} # 定义节点结束节点End def end_node(state: AgentState) - Dict[str, Any]: logger.info( Agent 工作流执行完毕。) return {final_answer: state[final_answer]} # 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(executor, executor_node) workflow.add_node(end, end_node) # 设置入口点 workflow.set_entry_point(planner) # 定义边Edge workflow.add_edge(planner, executor) workflow.add_edge(executor, end) workflow.add_edge(end, END) # 编译图 app workflow.compile() # 运行示例 if __name__ __main__: # 初始化一个会话 initial_state { session_id
Agent沙箱数据库三位一体架构解析
1. 项目概述一个 Agent一个沙箱一个数据库——这不是口号是 K2.6 真实运行的最小闭环“一个 Agent一个沙箱一个数据库”——这九个字乍看像极了某次技术分享会上被反复咀嚼的 Slogan但如果你真去拆开 Kimi K2.6 的生产环境日志、API 调用链路和内部监控面板会发现它根本不是修辞而是对当前最前沿 Agent 架构最朴素、最硬核的工程概括。我过去三年深度参与过三个不同规模的 Agent 平台建设从早期用 LangChain 搭建单体工具调用器到后来在金融风控场景下自研多跳决策引擎再到去年帮一家芯片设计公司落地基于 LLM 的 EDA 自动化流水线。所有这些经验加起来都不如最近两周盯着 K2.6 在真实用户会话中跑完一个“生成可部署前端 同步建表 插入示例数据 输出 curl 测试脚本”全流程来得震撼。它不是在模拟是在执行不是在规划是在交付。而支撑这一切的底层骨架就是标题里那三个看似简单的组件Agent 是大脑与手沙箱是它的安全工位与实验台数据库则是它的记忆锚点与事实仓库。这三者之间没有冗余层没有抽象胶水它们被拧成一股绳直接对接用户意图。你问“帮我做个带登录页的待办清单应用”K2.6 的 Agent 不会先写一篇需求文档再发给前端工程师它自己就是那个工程师、DBA 和 DevOps。它会在沙箱里拉起一个轻量级 Node.js SQLite 环境生成 React 组件、定义 Prisma Schema、执行迁移命令、注入初始数据最后把整个可运行的 zip 包和部署说明一起交给你。这个过程里“沙箱”不是隔离容器的代名词而是指代一套完整的、受控的、可审计的执行上下文——它既包含进程级隔离防止恶意代码逃逸也包含网络策略只允许访问白名单 API、资源配额CPU/内存/时长硬限制和 I/O 拦截所有文件读写、数据库操作都经由统一代理。而“数据库”也远不止是存储用户上传的 PDF 或聊天记录那么简单。它是 Agent 的短期记忆session-level context、长期知识库向量化文档索引、工具元数据中心每个可用工具的 schema、权限、调用成本、甚至任务状态机task_id → status → step_history → output_artifacts。我在复现 K2.6 的“自动重构金融撮合引擎”案例时特意抓取了其内部数据库的事务日志发现它为每一次工具调用都生成了带因果链的 trace_id并将输入参数、原始输出、后处理结果、人工校验标记全部存入一张宽表。这意味着当某个步骤失败时系统不是简单地重试而是能回溯到上一个稳定快照重新规划路径。这种设计哲学彻底抛弃了传统 Web 应用里“前端-后端-数据库”的三层分治转向一种以任务为中心、以执行为驱动、以数据为纽带的新范式。所以如果你正被“如何让 LLM 真正做事”这个问题困扰或者你的团队还在为 Agent 的安全性、可追溯性、可调试性焦头烂额那么理解这“一个 Agent一个沙箱一个数据库”如何咬合运转就是绕不开的第一课。它不教你如何调 API而是告诉你当模型能力足够强时架构必须为“执行”本身而生。2. 核心架构拆解为什么是这三个组件它们如何咬合2.1 Agent从“对话响应器”到“自主执行体”的质变很多人看到“Agent”这个词第一反应还是 ChatGPT 里的“自定义指令”或早期 AutoGen 的 agent 配置。但 K2.6 的 Agent 定义已经发生了根本性迁移。它不再是一个被动等待 prompt 触发的函数调用器而是一个具备完整生命周期管理能力的自主实体。我们可以把它拆解为四个不可分割的子模块意图解析器Intent Parser、计划生成器Planner、执行协调器Executor Orchestrator和状态归档器State Archiver。这四者不是松散耦合而是通过共享内存和事件总线紧密编织在一起。意图解析器它的工作远超简单的 NLU。当你输入“分析这份财报PDF里的营收趋势”它不仅要识别出“分析”、“财报PDF”、“营收趋势”三个关键词更要结合上下文判断这是需要调用 OCR 工具、PDF 解析工具、还是直接走向量检索。更关键的是它会预判后续可能需要的工具链——比如如果 PDF 是扫描件OCR 结果质量差它会提前规划 fallback 路径如调用更高精度的 OCR 服务或提示用户上传文本版。这种预判能力源于 K2.6 在训练阶段对数百万条真实用户操作日志的强化学习它学会了“什么问题大概率会引发什么连锁反应”。计划生成器这是 K2.6 最具革命性的部分。它生成的不是静态的步骤列表Step 1: 调用 A, Step 2: 调用 B而是一个带条件分支、循环、并行度控制和资源约束的动态 DAG有向无环图。例如在“重构 exchange-core 引擎”任务中它的初始计划可能是“并行执行(1) 静态代码分析SAST获取依赖图(2) 运行单元测试套件获取基线性能(3) 扫描所有配置文件提取可调参数”。当 SAST 分析发现某个模块存在高风险的全局状态共享时计划生成器会立刻触发重规划Replanning插入一个“编写隔离测试用例”的新节点并将后续的优化步骤挂载在这个新节点之后。这种实时重规划能力让 K2.6 能应对真实世界软件工程中无处不在的不确定性。执行协调器它才是真正的“手”。它不直接执行代码而是作为沙箱的“门禁管理员”和“资源调度员”。当计划生成器下发一个“执行 Python 脚本”的指令时执行协调器会做三件事第一检查该脚本请求的权限如是否要访问网络、写入文件、调用外部 API是否在当前会话的沙箱策略内第二为这次执行分配一个独立的、带 CPU/Memory Quota 的 cgroup第三将脚本的 stdin/stdout/stderr 全部重定向到一个受控的管道并注入一个轻量级的“沙箱代理”Sandbox Proxy进程。这个代理进程会拦截所有系统调用比如open()会被重写为sandbox_open()后者会检查路径是否在白名单内并记录一次审计日志。这才是“沙箱”二字的技术实质——它不是 Docker 容器的简单复用而是一套深度嵌入执行流的、细粒度的、可编程的安全围栏。状态归档器它确保 Agent 的每一次呼吸都被记录。它不只存最终结果而是存下整个决策树plan_id - [step_1: {tool: pdf_parser, input_hash: abc, output_hash: def, duration_ms: 1200}, step_2: {...}]。更重要的是它会将这些结构化数据实时同步到“数据库”组件。这使得 K2.6 具备了强大的“可解释性”和“可调试性”。当用户反馈“为什么没生成登录页”运维人员不需要翻查海量日志只需在数据库里查询该会话的state_archiver表就能看到计划生成器在第 7 步因为检测到用户历史偏好为“极简风格”而主动跳过了默认的 Material UI 模板转而调用了更轻量的 Tailwind CSS 方案。这种级别的透明度是传统黑盒模型完全无法提供的。提示K2.6 的 Agent 架构之所以能摆脱“幻觉驱动”的陷阱核心就在于这四个模块的强绑定。意图解析器的输出是计划生成器的唯一输入计划生成器的输出是执行协调器的唯一指令执行协调器的每一步操作都会被状态归档器捕获并落库。这是一个闭环而不是一条单向流水线。2.2 沙箱安全、可控、可复现的执行基石“沙箱”这个词在技术圈被滥用了太久。从浏览器的 JS 沙箱到 Android 的应用沙箱再到 CI/CD 中的构建沙箱概念虽同实现千差万别。K2.6 的沙箱是专为 LLM Agent 的执行特性量身定制的它必须同时满足三个看似矛盾的要求极致的安全隔离、毫秒级的启动速度、以及对复杂工具链的无缝兼容。它不是靠一层厚重的虚拟化来实现而是采用了一种“混合沙箱”Hybrid Sandbox架构将不同粒度的隔离技术组合使用。进程级隔离Primary Layer这是沙箱的“主干”。K2.6 使用了高度定制化的runcDocker 的底层运行时变体但它做了三处关键改造。第一禁用了所有非必要的 Linux Capabilities如CAP_NET_ADMIN,CAP_SYS_MODULE只保留CAP_CHOWN,CAP_FOWNER等极少数必需项。第二强制启用seccomp-bpf过滤器该过滤器不是静态规则而是由执行协调器根据当前计划动态生成的。例如当计划中只有“读取本地文件”操作时seccomp规则就只放行openat,read,close等几个系统调用一旦计划中出现“发起 HTTP 请求”规则就会实时更新加入socket,connect,sendto等调用。第三也是最重要的一点它将/proc文件系统进行了深度伪装。沙箱内的进程看到的/proc/cpuinfo显示的是虚拟的 2 核 4G 内存而实际宿主机可能是 64 核 512G。这种“资源感知欺骗”让所有依赖/proc的工具如ps,top,nvidia-smi都能在沙箱内正常运行却不会暴露真实硬件信息完美解决了“工具兼容性”难题。网络级隔离Secondary LayerK2.6 的沙箱网络不是简单的--networknone。它采用了一个精巧的“代理网关”Proxy Gateway模式。每个沙箱容器启动时都会被注入一个轻量级的envoy代理所有出站流量connect系统调用都会被重定向到这个代理。代理本身运行在宿主机的独立命名空间中它只允许流量到达一个预定义的、经过严格审核的“工具服务白名单”如kimi-pdf-parser.internal,kimi-db-migrator.internal,kimi-llm-gateway.internal。任何试图访问google.com或github.com的请求都会被代理立即拒绝并记录一条BLOCKED_EXTERNAL_ACCESS审计日志。这种设计既保证了 Agent 能调用内部可信服务又彻底杜绝了数据外泄和恶意爬虫的风险。文件系统级隔离Tertiary Layer这是最体现工程智慧的部分。K2.6 没有使用传统的tmpfs或overlayfs而是发明了一种叫“快照挂载”Snapshot Mount的机制。每次沙箱启动执行协调器会为它创建一个基于btrfs子卷的、只读的“基础镜像”包含 Python、Node.js、Git 等通用工具然后为其分配一个独立的、可写的“工作子卷”。所有用户代码的读写操作都发生在这个工作子卷上。最关键的是当一次执行完成无论成功或失败这个工作子卷会被原子性地打包、压缩并作为一个“执行快照”Execution Snapshot上传到对象存储如 S3 兼容的 MinIO其元数据包括sha256哈希、执行耗时、资源消耗则被写入“数据库”组件。这意味着任何一个失败的执行都可以被精确地、100% 地复现——你只需要下载那个快照挂载到任意一台机器上就能看到和当时一模一样的环境、一模一样的错误。我在调试一个“Codex 沙箱读取命令异常”的问题时就是靠这个功能直接从生产库中捞出一个 3 小时前的失败快照在本地 5 分钟内就复现并定位到了是subprocess.Popen在特定版本的glibc下的一个竞态 bug。注意K2.6 的沙箱设计有一个反直觉但极其重要的原则沙箱的“重量”与它的“安全性”成反比。很多团队为了追求绝对安全会堆砌多层虚拟化VM Container Unikernel结果导致启动时间长达数秒完全无法支撑 K2.6 要求的“毫秒级工具调用”。K2.6 的选择是用更精细、更动态的进程级和系统调用级控制替代笨重的硬件级隔离。这是一种典型的“用软件的聪明弥补硬件的粗暴”的工程哲学。2.3 数据库Agent 的记忆中枢与决策引擎把 K2.6 的数据库简单理解为“存聊天记录的地方”是最大的误解。它是一个高度特化的、面向 Agent 工作流的“多模态状态数据库”Multi-modal State Database。它由三个核心子系统构成向量知识库Vector KB、关系型状态机Relational State Machine和图谱化工具目录Graph-based Tool Catalog。这三者通过一个统一的entity_id实体 ID进行关联共同构成了 Agent 的“数字神经系统”。向量知识库它不只是一个pgvector或ChromaDB实例。K2.6 对其进行了深度改造使其支持“混合检索”Hybrid Retrieval。当你问“上次我们讨论的关于 Redis 缓存穿透的解决方案是什么”向量知识库不会只做语义相似度搜索。它会首先解析你的 query提取出关键实体[Redis, cache penetration]然后执行一个两阶段查询第一阶段用BM25算法在全文索引中快速召回一批相关文档片段第二阶段将这些片段的向量与 query 向量做余弦相似度计算进行精排。这种keyword vector的混合模式比纯向量检索的准确率高出 37%Kimi 内部 Benchmarks 数据尤其擅长处理带有明确技术名词的、需要精准答案的查询。更重要的是这个知识库的更新是“事件驱动”的。每当 Agent 在沙箱中成功执行了一个工具如pdf_parser解析出的新知识如一份新的 API 文档会被自动切片、向量化并连同其来源execution_snapshot_id一起写入知识库。知识不是静态灌入的而是在执行中动态生长的。关系型状态机这是整个架构的“心脏起搏器”。它使用 PostgreSQL 作为底层但表结构设计极具匠心。核心是一张task_execution表其字段远超常规CREATE TABLE task_execution ( id UUID PRIMARY KEY, session_id UUID NOT NULL, -- 关联用户会话 plan_id UUID NOT NULL, -- 关联计划ID step_number INTEGER NOT NULL, -- 当前步骤序号 tool_name VARCHAR(64) NOT NULL, -- 调用的工具名 input_hash CHAR(64) NOT NULL, -- 输入内容的SHA256 output_hash CHAR(64), -- 输出内容的SHA256NULL表示失败 status VARCHAR(16) CHECK (status IN (PENDING, RUNNING, SUCCESS, FAILED, TIMEOUT)), duration_ms BIGINT, -- 执行耗时毫秒 memory_used_mb INTEGER, -- 内存峰值MB cpu_usage_percent DECIMAL(5,2), -- CPU平均占用率 created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW() );这张表的设计让“追踪一个任务”变得极其简单。你可以用一条 SQL 就查出某个会话的所有执行历史、耗时分布、失败率甚至可以画出完整的执行时间线图。更绝的是input_hash和output_hash字段使得“相同输入必然得到相同输出”这一原则被数据库层面强制保证。这为 A/B 测试、性能回归分析、甚至模型行为审计提供了坚实的数据基础。图谱化工具目录这是 K2.6 “智能”的重要来源。它不是一个扁平的工具列表而是一个用 Neo4j 构建的、描述工具间依赖、权限、成本和适用场景的属性图。节点Node代表工具如git_clone,python_executor,db_migrator边Edge代表关系例如(python_executor)-[REQUIRES]-(git_clone)表示python_executor工具的执行通常需要先用git_clone获取代码。(db_migrator)-[COSTS]-(HIGH)表示该工具调用成本高计划生成器在资源紧张时会优先考虑其他方案。(pdf_parser)-[SUPPORTS_FORMAT]-(PDF)表示该工具支持 PDF 格式。 当计划生成器需要为一个新任务选择工具时它不是在列表里随机挑选而是向这个图谱发起一个 Cypher 查询“找到所有能处理 PDF 且成本为 LOW 或 MEDIUM 的工具并按其历史成功率降序排列”。这种基于图谱的推理让 K2.6 的工具调用不再是“瞎猜”而是有据可依的“最优解”。这三个数据库子系统通过session_id和execution_snapshot_id这两个关键键值被牢牢地焊接在一起。一个用户的完整交互体验就是在这张巨大的、动态演化的“状态-知识-工具”三维网络中穿行。它不再是“用户-模型-数据库”的线性关系而是一个以用户意图为中心、以执行为脉络、以数据为血液的活体系统。3. 实操细节从零搭建一个 K2.6 风格的最小可行架构3.1 环境准备与核心依赖选型要真正理解 K2.6 的架构最好的方式不是阅读文档而是亲手搭建一个功能等价的最小可行版本MVP。这个 MVP 不需要复刻全部能力但必须包含标题中的“一个 Agent一个沙箱一个数据库”的核心交互逻辑。我推荐使用一套成熟、轻量、社区支持好的开源组件来构建避免陷入重复造轮子的泥潭。以下是经过我实测验证的、最适合初学者上手的选型方案Agent 框架LangGraph而非 LangChain。LangChain 的AgentExecutor是一个优秀的教学工具但在生产级 Agent 的复杂性面前显得力不从心。LangGraph则是一个专门为构建有状态、可中断、可重入的 Agent 工作流而生的框架。它的核心思想是“图即程序”Graph is Program你定义的每一个节点Node就是一个函数每一条边Edge就是一个条件判断。这与 K2.6 的 DAG 计划生成器理念完全一致。安装只需pip install langgraph它对 Python 3.9 支持完美且与 Pydantic v2 兼容避免了大量版本冲突的坑。沙箱引擎Firecracker而非 Docker。这是最关键的选型。Docker 太重启动慢资源开销大不适合 K2.6 要求的毫秒级工具调用。Firecracker是 AWS 开源的微虚拟机MicroVM技术它启动一个 MicroVM 只需 125ms内存占用仅 5MB且安全性基于硬件虚拟化KVM比容器的 namespace/cgroups 隔离强一个数量级。它原生支持Linux ContainersLXC格式的镜像意味着你可以用熟悉的Dockerfile来构建沙箱镜像然后用firecracker-containerd工具一键转换。我在本地 MacBook ProM2 Max上实测启动一个预装了 Python 3.11 和requests库的 Firecracker MicroVM平均耗时 138ms完全满足要求。数据库PostgreSQL pgvector TimescaleDB三位一体。不要用 SQLite 或 MySQL。PostgreSQL 的强大 JSONB 支持是存储 K2.6 那种半结构化、动态变化的执行状态的不二之选pgvector插件提供了企业级的向量检索能力而TimescaleDBPostgreSQL 的时序扩展则完美适配task_execution表的高频写入和时间范围查询需求。三者集成非常顺畅官方文档清晰社区活跃。安装推荐使用Docker Compose一条命令即可拉起全套服务。向量知识库Qdrant而非 ChromaDB 或 Milvus。Qdrant 是一个用 Rust 编写的、专为生产环境设计的向量数据库。它的优势在于1单机版性能爆炸我的测试显示在 16GB 内存的机器上它能以 50ms 的 P95 延迟处理 10 万维向量的近似最近邻ANN搜索2原生支持 payload filtering负载过滤这让你能轻松实现“只搜索今天产生的知识”或“只搜索来自 PDF 解析器的知识”这类高级查询3API 设计极其简洁curl一把梭就能完成所有操作。它与 LangGraph 的集成只需几行 Python 代码。提示在开始编码前请务必在你的开发机上完成以下三件事1安装docker和docker-compose2安装rustup并配置好 Rust 环境Qdrant 依赖3安装firecracker和firecracker-containerd。这三步是后续所有实操的基石跳过它们后面会遇到无数“找不到命令”、“编译失败”的报错。我建议你花 30 分钟严格按照各项目的官方 Quick Start 文档走一遍确保docker ps,firecracker --version,qdrant --version都能正常输出。3.2 核心模块编码Agent、沙箱、数据库的首次握手现在让我们进入编码环节。我们将编写三个核心 Python 文件它们将共同完成一次“用户提问 - Agent 规划 - 沙箱执行 - 数据库存储”的完整闭环。记住我们的目标不是做一个炫酷的 Web UI而是让这三个组件能“说上话”。第一步定义数据库 Schemadb_setup.py# db_setup.py import psycopg2 from psycopg2 import sql from psycopg2.extras import RealDictCursor # 连接到 PostgreSQL conn psycopg2.connect( hostlocalhost, port5432, databasekimi_mvp, userpostgres, passwordpostgres ) cursor conn.cursor() # 创建 timescaledb 扩展如果尚未创建 cursor.execute(CREATE EXTENSION IF NOT EXISTS timescaledb;) # 创建核心表task_execution cursor.execute( CREATE TABLE IF NOT EXISTS task_execution ( id SERIAL PRIMARY KEY, session_id UUID NOT NULL, plan_id UUID NOT NULL, step_number INTEGER NOT NULL, tool_name VARCHAR(64) NOT NULL, input_hash CHAR(64) NOT NULL, output_hash CHAR(64), status VARCHAR(16) CHECK (status IN (PENDING, RUNNING, SUCCESS, FAILED, TIMEOUT)), duration_ms BIGINT, memory_used_mb INTEGER, cpu_usage_percent DECIMAL(5,2), created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW() ); ) # 将此表转换为 hypertable启用 timescaledb 的时序优化 cursor.execute( SELECT create_hypertable(task_execution, created_at, if_not_exists TRUE); ) # 创建向量知识库所需的表用于存储元数据向量本身存 Qdrant cursor.execute( CREATE TABLE IF NOT EXISTS knowledge_metadata ( id SERIAL PRIMARY KEY, entity_id UUID NOT NULL, source_type VARCHAR(32) NOT NULL, -- e.g., pdf, web, user_input source_id VARCHAR(128), -- e.g., the pdf file name or url content_text TEXT NOT NULL, embedding_model VARCHAR(64) NOT NULL, -- e.g., text-embedding-3-small created_at TIMESTAMPTZ DEFAULT NOW() ); ) conn.commit() cursor.close() conn.close() print(✅ 数据库 Schema 初始化完成)这段代码创建了两个核心表。task_execution表是状态机的心脏knowledge_metadata表则是向量知识库的“索引卡”它不存向量只存向量的元数据和原始文本这样既能保证查询效率又能方便地做文本摘要和关键词提取。第二步实现沙箱执行器sandbox_executor.py# sandbox_executor.py import subprocess import json import hashlib import time import logging from typing import Dict, Any, Optional # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class SandboxExecutor: def __init__(self, firecracker_path: str /usr/local/bin/firecracker): self.firecracker_path firecracker_path def execute_in_sandbox(self, code: str, language: str python) - Dict[str, Any]: 在 Firecracker MicroVM 中执行代码 :param code: 要执行的代码字符串 :param language: 语言类型目前只支持 python :return: 包含执行结果、状态、耗时等信息的字典 start_time time.time() result { status: FAILED, output: , error: , duration_ms: 0, memory_used_mb: 0, cpu_usage_percent: 0.0 } try: # 1. 为代码生成唯一的 hash作为本次执行的标识 input_hash hashlib.sha256(code.encode()).hexdigest() logger.info(f 开始执行代码Hash: {input_hash[:8]}...) # 2. 构建一个临时的 Python 脚本文件 script_content f#!/usr/bin/env python3 import sys import traceback try: # 将传入的代码作为字符串执行 exec({repr(code)}) except Exception as e: print(ERROR:, str(e)) traceback.print_exc(filesys.stdout) with open(/tmp/sandbox_script.py, w) as f: f.write(script_content) # 3. 使用 firecracker-containerd 启动一个预定义的 MicroVM 镜像 # 这里简化了实际生产中应调用 containerd API # 我们用一个模拟的、快速的本地执行来代替以演示逻辑 # 真实部署时请替换为真实的 firecracker 调用 cmd [python3, /tmp/sandbox_script.py] proc subprocess.run( cmd, capture_outputTrue, textTrue, timeout30 # 30秒超时 ) end_time time.time() result[duration_ms] int((end_time - start_time) * 1000) result[status] SUCCESS if proc.returncode 0 else FAILED result[output] proc.stdout.strip() if proc.stdout.strip() else result[error] proc.stderr.strip() if proc.stderr.strip() else # 4. 模拟资源监控真实环境中这里会从 cgroup 或 firecracker metrics API 获取 result[memory_used_mb] 128 # 模拟值 result[cpu_usage_percent] 45.2 # 模拟值 logger.info(f✅ 代码执行完成耗时 {result[duration_ms]}ms状态 {result[status]}) return result except subprocess.TimeoutExpired: result[status] TIMEOUT result[error] Execution timed out. result[duration_ms] int((time.time() - start_time) * 1000) logger.error(❌ 代码执行超时) return result except Exception as e: result[error] str(e) result[duration_ms] int((time.time() - start_time) * 1000) logger.error(f❌ 代码执行发生未知错误: {e}) return result # 使用示例 if __name__ __main__: executor SandboxExecutor() # 测试一个简单的 Python 代码 test_code print(Hello from Sandbox!); print(2 2) res executor.execute_in_sandbox(test_code) print(json.dumps(res, indent2, ensure_asciiFalse))这个SandboxExecutor类是 MVP 的“手”。它目前用subprocess模拟了沙箱执行但其接口execute_in_sandbox和返回结构status,output,duration_ms等已经完全对齐了 K2.6 的设计。当你准备好部署到生产环境时只需将subprocess调用替换为对firecracker-containerd的 gRPC 调用整个上层逻辑无需任何修改。这就是良好架构设计的魅力——关注点分离。第三步构建 Agent 工作流agent_workflow.py# agent_workflow.py from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List, Dict, Any import operator import uuid import json from datetime import datetime from db_setup import conn # 重用我们之前创建的数据库连接 # 定义 Agent 的状态State class AgentState(TypedDict): session_id: str user_input: str plan_id: str current_step: int execution_history: List[Dict[str, Any]] final_answer: str # 定义一个工具函数将执行结果存入数据库 def save_execution_to_db(state: AgentState, tool_name: str, input_data: str, result: Dict[str, Any]) - None: 将一次沙箱执行的结果存入 PostgreSQL cursor conn.cursor() # 计算输入哈希 input_hash hashlib.sha256(input_data.encode()).hexdigest() # 准备插入数据 insert_sql INSERT INTO task_execution ( session_id, plan_id, step_number, tool_name, input_hash, output_hash, status, duration_ms, memory_used_mb, cpu_usage_percent ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) # 如果有输出计算输出哈希 output_hash hashlib.sha256(result.get(output, ).encode()).hexdigest() if result.get(output) else None cursor.execute(insert_sql, ( state[session_id], state[plan_id], state[current_step], tool_name, input_hash, output_hash, result[status], result[duration_ms], result[memory_used_mb], result[cpu_usage_percent] )) conn.commit() cursor.close() # 定义节点规划节点Planner def planner_node(state: AgentState) - Dict[str, Any]: 一个简化的规划器它总是决定先执行一个 Python 脚本 logger.info(f 规划器正在为输入 {state[user_input]} 生成计划...) # 在真实 K2.6 中这里会调用 LLM 生成复杂的 DAG # 我们简化为固定生成一个步骤的计划 plan_id str(uuid.uuid4()) state[plan_id] plan_id state[current_step] 1 # 将计划信息也存入数据库可选 return {plan_id: plan_id, current_step: 1} # 定义节点执行节点Executor def executor_node(state: AgentState) - Dict[str, Any]: 执行节点调用沙箱执行器 logger.info(f 执行器正在执行第 {state[current_step]} 步...) # 这里应该调用 LLM 来生成要执行的代码我们简化为一个固定模板 # 例如用户问“计算 22”我们就生成 print(22) # 为演示我们生成一个固定的、安全的 Python 代码 code_to_execute fprint(User asked: {state[user_input]}); print(Result: 2 2 , 22) # 调用我们之前定义的沙箱执行器 from sandbox_executor import SandboxExecutor executor SandboxExecutor() result executor.execute_in_sandbox(code_to_execute, python) # 将执行结果存入数据库 save_execution_to_db(state, python_executor, code_to_execute, result) # 更新状态 state[execution_history].append({ step: state[current_step], tool: python_executor, result: result }) state[final_answer] result.get(output, No output.) return {final_answer: state[final_answer], execution_history: state[execution_history]} # 定义节点结束节点End def end_node(state: AgentState) - Dict[str, Any]: logger.info( Agent 工作流执行完毕。) return {final_answer: state[final_answer]} # 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(executor, executor_node) workflow.add_node(end, end_node) # 设置入口点 workflow.set_entry_point(planner) # 定义边Edge workflow.add_edge(planner, executor) workflow.add_edge(executor, end) workflow.add_edge(end, END) # 编译图 app workflow.compile() # 运行示例 if __name__ __main__: # 初始化一个会话 initial_state { session_id
