作者：江昱在构建 Agent 应用时，凭证管理是一个容易被忽视但又极其重要的问题。一个典型的 Agent 应用会面临两个方向的凭证需求：向内，用户如何安全地调用你的 Agent？向外，Agent 如何安全地调用外部服务？传统做法存在诸多问题。硬编码在代码里容易泄露且难以更新，存在配置文件中同样有安全风险，每次都手动传递不仅麻烦还容易出错，让大模型处理凭证更是巨大的安全隐患。更棘手的是，当凭证需要更新时（比如 API Key 过期、权限变更），如何在不重启服务的情况下动态更新？函数计算 AgentRun 的凭证管理系统就是为了解决这些问题而生。入站凭证与出站凭证：双向安全保障函数计算 AgentRun 的凭证管理分为两个维度，分别解决“谁能调用我”和“我能调用谁”的问题。入站凭证：控制谁能访问你的 Agent入站凭证用于控制外部用户或系统如何访问你的 Agent 应用。当你创建一个 Agent 并对外提供服务时，需要确保只有授权的用户才能调用。函数计算 AgentRun 提供了灵活的入站凭证管理，可以为不同的调用方生成独立的凭证，设置不同的权限和配额，控制每个凭证能访问哪些 Agent、调用频率限制、有效期等。由于所有请求都经过函数计算 AgentRun 网关，入站凭证可以实现真正的动态更新。 比如你的 Agent 对外提供客服能力，可以为不同的业务部门生成不同的入站凭证，每个部门只能访问各自授权的 Agent。当某个部门的凭证泄露时，可以立即撤销并重新生成，所有变更在网关层实时生效，不影响其他部门的使用，也无需重启任何服务。出站凭证：安全调用外部服务出站凭证用于 Agent 访问外部服务时的身份认证。Agent 应用通常需要调用各种外部服务：大模型 API（OpenAI、Claude、Qwen 等）、数据库、第三方工具、企业内部系统等，每个服务都需要相应的凭证。传统方式下，开发者要么把这些凭证硬编码在代码里，要么通过环境变量传递，不仅不安全，更新时还需要重启服务。函数计算 AgentRun 采用了一套巧妙的定时查询与缓存机制来管理出站凭证。 所有出站凭证统一存储在加密的凭证库中，代码里不再出现任何敏感信息。Agent 启动时会从凭证库拉取所需的所有凭证并缓存到本地，运行过程中直接使用本地缓存，避免频繁的网络请求带来的性能开销。同时，系统会定期进行健康检查，主动查询凭证是否有更新，发现变更时只更新发生变化的凭证。如果健康检查失败，会自动重试，确保凭证始终可用。这种定时查询方案带来了多重价值。 从性能角度看，本地缓存避免了每次调用都查询凭证库，大幅降低了延迟和网络开销；从可用性角度看，即使凭证服务短暂不可用，缓存的凭证仍然可用，不会影响 Agent 的正常运行；从安全性角度看，定时健康检查确保凭证泄露或过期时能在几分钟内完成更新，而不需要等到下次部署。最关键的是，整个更新过程对 Agent 代码完全透明，开发者无需编写任何凭证更新逻辑，专注于业务实现即可。这种最终一致性的设计在实践中被证明是最优的平衡：既保证了性能和可用性，又实现了凭证的动态更新能力。相比于每次都实时查询（性能差）或者只在启动时加载（更新不及时），定时查询方案在三者之间找到了最佳平衡点。实际应用：工具和模型的凭证配置函数计算 AgentRun 的凭证管理在两个关键场景发挥作用，展示了从理论到实践的完整闭环。场景一：大模型调用的凭证管理当你的 Agent 需要调用多个大模型时，每个模型都需要各自的 API Key。以前你可能需要在代码里硬编码这些 Key，或者通过环境变量传递，但这样做存在安全风险且更新困难。有了函数计算 AgentRun 的凭证管理，你只需要在平台上配置各个模型的出站凭证，给每个凭证命名（如 openai_key、qwen_key），然后在 Agent 配置中引用这些凭证名称。运行时系统会自动注入实际的 Key，你的代码里完全看不到任何敏感信息。当某个模型的 Key 过期需要更新时，只需在凭证管理界面更新，几分钟后所有使用该凭证的 Agent 会通过定时健康检查自动获取新的 Key，无需修改代码或重启服务。这种体验就像是有一个智能管家在后台默默地帮你管理所有的钥匙，你只需要告诉他你要开哪扇门。# Agent 配置示例（伪代码） models: - name: gpt-4 credential: ${credentials.openai_key} # 引用凭证名称，不暴露实际Key - name: qwen-max credential: ${credentials.qwen_key}场景二：工具调用的凭证注入回到之前提到的 FunctionQ 案例，这是一个更复杂但也更能体现凭证管理价值的场景。Agent 需要通过 MCP 调用 CLI 工具查询用户的函数计算资源，这些工具需要用户的 AccessKey 和 SecretKey。关键问题是：如何在不暴露凭证给大模型的前提下，让工具能够正确调用 API？函数计算 AgentRun 通过前置 Hook 实现了优雅的动态凭证注入。 用户在平台上配置自己的出站凭证后，Agent 调用工具时请求中只携带用户 ID，不包含任何凭证信息。前置 Hook 拦截请求，根据用户 ID 从凭证库获取对应的凭证，然后将凭证注入到环境变量或请求参数中。工具使用注入的凭证执行实际操作，后置 Hook 再清理敏感信息并记录审计日志。整个过程中，凭证从未暴露给大模型，也不会出现在 Agent 的代码中，真正做到了安全可控。核心价值：让开发者专注业务逻辑函数计算 AgentRun 的凭证管理系统带来的价值远不止“管理凭证”这么简单。从安全性角度看，凭证不再出现在代码和日志中，集中加密存储大幅降低泄露风险，即使某个凭证泄露也可以快速撤销和更换。从开发效率角度看，开发者不需要关心凭证如何存储、如何传递、如何更新，只需在配置中引用凭证名称，系统自动处理剩下的事情。从运维角度看，凭证更新不需要修改代码、不需要重新部署、不需要重启服务，在管理界面更新后通过定时机制自动生效。更重要的是，凭证管理让 Agent 应用从“能用”变成“敢用” 。企业不再担心凭证泄露的风险，不再为凭证更新而头疼，不再因为安全问题而犹豫是否将 Agent 应用部署到生产环境。这种信心的建立，才是凭证管理最大的价值所在——它消除了企业拥抱 AI Agent 的最后一道顾虑，让技术真正为业务创造价值。立即体验函数计算 AgentRun函数计算 AgentRun 的无代码到高代码演进能力，现已开放体验：查看更多产品详情：https://www.aliyun.com/product/fc/agentrun快速创建：访问控制台（https://functionai.console.aliyun.com/cn-hangzhou/agent/explore），60 秒创建你的第一个 Agent深度定制：当需要更复杂功能时，一键转换为高代码持续演进：利用函数计算 AgentRun 的基础设施能力，持续优化你的 Agent从想法到上线，从原型到生产，函数计算 AgentRun 始终是你最好的伙伴。欢迎加入“函数计算 AgentRun 客户群”，钉钉群号：134570017218。快速了解函数计算 AgentRun：一句话介绍： 函数计算 AgentRun 是一个以高代码为核心的一站式 Agentic AI 基础设施平台。秉持生态开放和灵活组装的理念，为企业级 Agent 应用提供从开发、部署到运维的全生命周期管理。函数计算 AgentRun 架构图函数计算 AgentRun 运行时基于阿里云函数计算 FC 构建，继承了 Serverless 计算极致弹性、按量付费、零运维的核心优势。通过深度集成 AgentScope、LangChain、RAGFlow、Mem0 等主流开源生态。函数计算 AgentRun 将 Serverless 的极致弹性、零运维和按量付费的特性与 AI 原生应用场景深度融合，助力企业实现成本与效率的极致优化，平均 TCO 降低 60% 。 让开发者只需专注于 Agent 的业务逻辑创新，无需关心底层基础设施，让 Agentic AI 真正进入企业生产环境。
探秘 AgentRun｜基于 Serverless 的 AI Agent 沙箱工程化之路
阿里云函数计算 AgentRun 全新发布后，我们整理了“探秘 AgentRun”系列文章，本系列将梳理企业落地Agent 常见难题，给出具体解法，助力 Agentic AI 快速走进生产级环境。欢迎加入“函数计算 AgentRun 客户群”与我们交流，钉钉群号：134570017218。AI Agent 时代的沙箱需求从 Copilot 到 Agent：执行能力的质变在生成式 AI 的早期阶段，应用主要以“Copilot”形式存在，AI 仅作为辅助生成建议。然而，随着 AutoGPT、BabyAGI 以及 OpenAI Code Interpreter（现为 Advanced Data Analysis）的出现，AI 开始扮演“Agent”的角色。Agent 被赋予了目标，并能自主规划步骤、使用工具来达成目标。这种质变的核心在于代码执行（Code Execution）。为了回答“分析这层楼的销售数据并绘制趋势图”这样的请求，LLM 不再只是生成一段 Python 代码文本，而是需要在一个真实的 Python 环境中运行这段代码，并获取绘图结果。同样，为了“帮我预订一张去东京的机票”，Agent 可能需要在一个无头浏览器（Headless Browser）中模拟用户点击。不可信代码的安全隐患当 LLM 生成代码并执行时，这段代码在本质上是不可信的（Untrusted）。如果直接在应用服务器或用户的本地设备上运行，将面临灾难性的安全风险：系统破坏：AI 生成的代码可能无意或恶意地包含 rm -rf / 等破坏性指令，或者修改关键系统配置文件。数据泄露：代码可能尝试读取环境变量中的 API Key，或者扫描内网数据库，将敏感数据发送到外部服务器。资源耗尽：死循环或内存泄漏代码可能导致宿主机崩溃，影响其他租户的服务。网络攻击：恶意 Prompt 注入（Prompt Injection）可能诱导 AI 将执行环境作为跳板（Jump Box），对内部网络发起 DDoS 攻击或端口扫描。Agent 场景面临的独特挑战除了基础的安全性，AI Agent 的交互特性还给沙箱环境带来了前所未有的工程挑战，这也是传统沙箱（如简单的 Docker 容器或虚拟机）难以应对的：状态保持：与传统的“请求-响应”模式不同，Agent 往往需要进行多轮对话。上一轮定义的变量（如 df = load_data()）需要在下一轮（df.plot()）中继续可用。这就要求沙箱环境必须具备上下文记忆能力，而非每次请求都重置环境。极速启动：用户无法忍受每次交互都等待数秒甚至数十秒的虚拟机启动时间。为了保证流畅的对话体验（Time to First Token），沙箱必须具备毫秒级的冷启动能力。环境依赖多样性：不同的 Agent 任务可能需要完全不同的依赖库（如 Pandas、Scipy 用于数据分析，Puppeteer 用于网页操作）。沙箱需要支持灵活的自定义镜像或动态依赖加载，同时不能影响启动速度。资源成本控制：Agent 的调用往往具有稀疏性和突发性（例如一天只用几次，但一次用很久）。长期运行独占的虚拟机（VM）成本高昂且资源利用率低，而传统的 FaaS 虽然便宜但往往缺乏状态保持能力。如何在低成本和高性能之间找到平衡点，是一个巨大的挑战。因此，构建一个沙箱（Sandbox）——一个与宿主机、内网以及其他用户数据严格隔离，同时具备高性能、低成本、有状态的封闭执行环境——成为了 AI Agent 沙箱落地的前提条件。AgentRun Sandbox：专为 Agent 设计的工程化方案为了解决上述挑战，我们推出了 AgentRun Sandbox。这是一个以高代码为核心，开放生态、灵活组装的一站式 Agentic AI 基础设施平台。AgentRun 并非从零构建传统的虚拟机集群，而是基于阿里云函数计算（FC）这一强大的 Serverless 底座构建。通过充分利用 Serverless 的按需付费、极致弹性以及免运维（NoOps） 特性，AgentRun 解决了一直困扰沙箱领域的成本与效率难题，并在此基础上通过工程化封装，提供了面向 Agent 场景的专业能力。为什么选择函数计算作为 Sandbox Infra在构建 Agent 沙箱时，我们坚定地选择了函数计算（FC）作为底层基础设施，这主要基于以下核心优势的考量：强安全隔离： 沙箱的核心诉求是安全。函数计算底层采用神龙裸金属与 RunD 安全容器技术，每个执行环境都运行在独立的 MicroVM 中。这种基于虚拟化技术的内核级隔离，相比传统的 Docker 容器隔离具有更高的安全性，能有效防止恶意代码逃逸，为不可信代码执行提供了坚实屏障。极致弹性与冷启动优化： Agent 的调用往往具有突发性。函数计算具备毫秒级的弹性伸缩能力，结合 RunD 技术对启动速度的极致优化，使得沙箱能够在数秒甚至毫秒内完成创建和启动。这不仅满足了高并发场景下的需求，也保证了 Agent 交互的流畅性，避免了传统虚拟机启动慢带来的延迟感。成本效益：自建虚拟机集群通常需要为峰值流量预留资源，导致低谷期资源浪费。函数计算采用按需付费（Pay-as-you-go）模式，且 AgentRun 利用了 FC 的空闲自动回收机制，真正做到了“有请求才计费”。对于稀疏调用的 Agent 场景，这种模式能显著降低基础设施成本。免运维： 基于 Serverless 架构，开发者无需关心底层服务器的操作系统补丁、网络配置及集群维护。AgentRun 团队可以将精力集中在沙箱的核心逻辑与业务体验上，而非底层基础设施的繁琐运维。会话能力：函数计算围绕 AI Agent Sandbox 场景推出了会话亲和、隔离以及管理能力。在一次会话生命周期内，相同会话的请求均会被亲和路由到同一个实例中，并独占该实例，保证了会话交互的连续性、上下文完整性以及多租安全性，同时提供完整的管理接口来主动对会话生命周期进行控制，降低了开发门槛。AgentRun 的核心运行机制传统的 Serverless 通常是无状态的，难以满足 Code Interpreter 这类需要上下文保持的场景。AgentRun 借助函数计算的会话产品能力，在无状态的计算底座上构建了有状态、会话级的沙箱体验。1. 沙箱请求亲和AgentRun 允许开发者显式地创建一个具有生命周期的执行环境，解决了传统 Serverless“用完即走”导致的上下文丢失问题。会话亲和：AgentRun 依赖函数计算会话亲和机制。当开发者创建沙箱后，AgentRun 会维护一个唯一的 SessionID。后续所有携带该 ID 的请求，都会被精准路由到同一个底层的计算实例。这意味着用户在第一步定义的 df = pd.read_csv(...) 对象，在第二步 df.plot() 时依然存在于内存中，完美复刻本地开发体验。MCP 协议原生支持：针对模型上下文协议（Model Context Protocol, MCP），AgentRun 提供了 MCP SSE 及 MCP Streamable HTTP 会话亲和支持。AgentRun 可以直接作为 MCP 网关，让 LLM 与外部工具的交互更加顺滑。2. 多层次安全隔离在多租户 SaaS 平台中，安全性是 AgentRun 的基石。计算隔离：AgentRun 利用底层基础设施的神龙裸金属与 RunD 安全容器技术，确保每个沙箱实例在内核级别进行隔离。通过强制将会话并发度设置为 1，AgentRun 保证租户 A 的进程空间、内存数据与租户 B 物理分离，防止容器逃逸。网络隔离：网络隔离完全由用户控制。用户可以根据安全需求灵活配置，选择开启或关闭沙箱的公网访问权限，或者将沙箱接入指定的 VPC 网络环境，从而在满足业务连通性的同时，防止恶意代码对内网发起攻击。3. 灵活的生命周期控制AgentRun 通过函数计算的会话能力，接管了底层计算资源的生命周期，为上层应用提供精细化管理：自动闲置回收（Idle Timeout）：为了通过 Serverless 架构降低成本，AgentRun 支持设置空闲超时（例如 5 分钟）。如果 Agent 在这段时间内没有新指令，底层实例会自动销毁并停止计费，完美适配 AI 交互“突发性强、稀疏度高”的特点。状态暂停与恢复（即将上线）：针对长时间的任务间歇，AgentRun 能够将沙箱的内存与磁盘状态快照保存，在用户回归时通过快照快速恢复现场，既节省成本又保留了上下文。4. 会话粒度存储隔离（即将上线）代码执行需要隔离，数据存储更需要隔离。AgentRun 创新性地规划了会话粒度存储粘性。动态绑定：AgentRun 允许用户为每个沙箱环境中动态分配一个存储挂载点的专属子目录。逻辑沙箱：通过底层的挂载技术，沙箱内部只能看到属于自己的 /workspace，物理上无法访问其他租户的文件（如 ../../tenant-b/secret.txt），从文件系统层面根除了数据交叉风险。AgentRun 开箱即用的沙箱能力AgentRun 不仅提供了底层隔离环境，还预置了经过工程化调优的标准化模版，让开发者开箱即用：Code Interpreter（代码解释器）：预装 Python/Node.js/Java 等环境，支持文件上传下载、数据分析、图表绘制及命令行操作。Browser User（浏览器沙箱）：提供基于 CDP over WebSocket 协议的浏览器环境，兼容 Puppeteer / Playwright，让 Agent 能够安全地访问互联网进行网页操作。All In One：集成了代码解释器与浏览器环境的全能型沙箱，满足复杂 Agent 任务需求。这些模版镜像具备高度的灵活性，AgentRun 未来将开放镜像定义，允许用户基于标准镜像定制私有依赖库或安全策略。AgentRun 沙箱架构详解AgentRun 网关这是 AgentRun 的门户，负责接收来自 AI Agent（如 LangChain 应用、ChatGPT Plugin）的 HTTP 请求，除了标准的身份验证、鉴权以及协议转换（如将 HTTP 转为 WebSocket）之外，其核心能力便是沙箱管理以及沙箱请求路由的功能，它屏蔽了底层 Serverless 基础设施的复杂性，实现了如下能力：沙箱管理：管理沙箱资源，维护业务层沙箱 ID 与底层计算资源 SessionID 的映射关系状态维护：监控沙箱的活跃状态，基于沙箱超时配置以及底层资源情况及时对状态进行更新资源调度：根据用户指定的计算规格（CPU、Memory），向底层申请相应的资源。函数计算沙箱环境主要由函数计算作为底层算力来承载沙箱的运行。AgentRun 利用函数计算提供的极致弹性能力，实现在分钟内启动成三万个独立的沙箱环境，每个环境都运行在独立的 MicroVM 中，搭配自研开箱即用的沙箱镜像模版，在功能以及性能上为用户提供了双重保障。典型工作流：从指令到结果以“用户让 Agent 根据上传的 Excel 文件绘制图表”为例，AgentRun 的工作流程如下。阶段一：模板创建用户请求：Agent 接收到用户指令后，由 LLM 决策使用 Python 来实现该需求。Agent 工具调用：AI Agent 会向 AgentRun 网关发送 Code Interpreter 沙箱模板的创建请求。模板创建：AgentRun 网关会调用函数计算接口创建一个 Code 沙箱模板函数，镜像配置为前文提到的自研 Code Interpreter 沙箱模板，该函数需要同时配置会话亲和以及会话隔离。阶段二：沙箱创建Agent 工具调用：模板创建完成后，Agent 继续进行沙箱创建，创建时传入已有的模板 ID，标识沙箱实例运行时的配置和镜像沙箱创建：AgentRun 收到沙箱创建请求后，会调用 FC 的 CreateSession 接口来创建一个沙箱实例，该沙箱会有一个合适的闲置超时时间，最长可存活 24h创建完成：AgentRun 会保存 FC 返回的会话 ID，并生成沙箱业务 ID 与之对应，最终将沙箱业务 ID 返回给用户阶段三：任务执行上传文件：Agent 通过 Code Interpreter 的文件上传接口，将 Excel 文件上传。若想将该文件持久化，可以在创建沙箱时配置持久化存储 NAS，将其挂到沙箱中，并将文件上传到 NAS 挂载的目录上。绘制图表：Agent 生成代码 import pandas as pd; df = pd.read_excel('data.xlsx')，并调用 Code Interpreter 的 run_code 接口执行代码。会话亲和：Agent 所有发往 Code Interpreter 的请求中，都必须带上对应的沙箱 ID 才能保证请求都路由到同一个沙箱实例。内存驻留：代码执行完毕，变量 df 驻留在内存中.二次代码执行：Agent 根据数据列名生成绘图代码 df.plot()。再次发送代码运行请求上下文复用：请求再次到达同一实例，直接使用内存中的 df 对象进行绘图，生成图片文件。结果回传：图片被写入 NAS，下载链接返回给 Agent。阶段四：资源销毁空闲检测：Agent 完成任务，不再发送请求。自动回收：达到 SessionIdleTimeout（如 5 分钟）后，函数计算会自动销毁该沙箱实例，此时除了持久化到 NAS 上的数据，其余环境相关数据均被销毁。文件回收：如果 NAS 上的文件是会话隔离的，当用户会话结束后，NAS 上文件需要进行主动或者定时自动清除。工作时序图AgentRun 的核心设计原则AgentRun 的工程化实践遵循以下五大核心原则，这构成了其安全、高效、可扩展的基石：原则一：配置即代码AgentRun 将沙箱环境定义（环境变量、资源规格、健康检查等）封装为标准化模版。这种设计实现了沙箱配置的版本化管理，使得 Agent 环境可以像代码一样进行复制和回滚。原则二：会话即沙箱AgentRun 将“会话”作为沙箱的唯一实体。通过 SessionID 绑定底层的计算实例与上下文状态，实现了真正的按需分配与状态保持。沙箱的创建与销毁完全独立于底层物理设施，对用户透明。原则三：生命周期可编程AgentRun 不仅提供创建（Create）和删除（Delete）接口，还引入了“暂停”、“恢复”和“自动超时”机制。这种可编程性让上层应用能根据业务价值最大化资源利用率，实现成本与性能的最优平衡。原则四：网络接入标准化AgentRun 抹平了底层网络的差异，提供标准化的 HTTP/WebSocket 接口，并支持 Server-Sent Events（SSE）。无论底层如何升级，上层 Agent 沙箱始终通过标准的 Header 或 Cookie 携带 SessionID 进行交互，降低了集成复杂度。原则五：存储隔离细粒度化（即将上线）AgentRun 不仅支持模版粒度的文件系统共享，同时也能够配置沙箱粒度目录级动态挂载。每个沙箱单独挂载一个目录，从根源上杜绝了多租户环境下的数据越权访问风险。总结与展望AgentRun Sandbox 是 Serverless 技术在 AI Agent 领域的最佳工程化实践。通过将阿里云函数计算（FC）在 RunD 安全虚拟化（解决隔离与启动速度）、会话亲和性（解决状态保持）以及 动态 NAS 挂载（解决数据隔离）等方面的底层技术创新，封装为面向业务的 AgentRun 平台，我们成功降低了企业构建 AI Agent 的门槛。对于构建下一代智能体应用的企业而言，选择 AgentRun Sandbox 不仅是选择了一个沙箱工具，更是选择了一套兼顾安全性、用户体验与商业效率的弹性基础设施。未来，AgentRun Sandbox 将继续在启动延迟优化、状态秒级快照恢复以及更多样化的存储支持上深耕，致力于成为 AI Agent 时代最佳的沙箱基座。立即体验函数计算 AgentRun函数计算 AgentRun 的无代码到高代码演进能力，现已开放体验：查看更多产品详情：https://www.aliyun.com/product/fc/agentrun1.快速创建：访问控制台（https://functionai.console.aliyun.com/cn-hangzhou/agent/explore)，60秒创建你的第一个 Agent2.深度定制：当需要更复杂功能时，一键转换为高代码3.持续演进：利用函数计算 AgentRun 的基础设施能力，持续优化你的 Agent从想法到上线，从原型到生产，函数计算 AgentRun 始终是你最好的伙伴。欢迎加入“函数计算 AgentRun 客户群”，钉钉群号：134570017218。快速了解函数计算 AgentRun一句话介绍：函数计算 AgentRun 是一个以高代码为核心的一站式 Agentic AI 基础设施平台。秉持生态开放和灵活组装的理念，为企业级 Agent 应用提供从开发、部署到运维的全生命周期管理。函数计算 AgentRun 架构图AgentRun 运行时基于阿里云函数计算 FC 构建，继承了 Serverless 计算极致弹性、按量付费、零运维的核心优势。通过深度集成 AgentScope、LangChain、RAGFlow、Mem0 等主流开源生态。函数计算 AgentRun 将 Serverless 的极致弹性、零运维和按量付费的特性与 AI 原生应用场景深度融合，助力企业实现成本与效率的极致优化，平均 TCO 降低 60%。让开发者只需专注于 Agent 的业务逻辑创新，无需关心底层基础设施，让 Agentic AI 真正进入企业生产环境。
探秘 AgentRun丨为什么应该把 LangChain 等框架部署到函数计算 AgentRun
作者：江昱阿里云函数计算 AgentRun 全新发布后，我们整理了“探秘 AgentRun”系列文章，本系列将梳理企业落地 Agent 常见难题，给出具体解法，助力 Agentic AI 快速走进生产级环境。欢迎加入“函数计算 AgentRun 客户群”与我们交流，钉钉群号： 134570017218 。当你已经用 LangChain、AgentScope、LangGraph 等框架开发了 Agent 应用，如何让它们享受函数计算 AgentRun 提供的 Serverless 运行时、企业级 Sandbox、模型高可用、全链路可观测等能力？好消息是，你几乎不需要改动现有代码，只需要简单的适配就可以迁移到函数计算 AgentRun。这篇文章将通过真实的代码示例，展示如何将不同框架的 Agent 应用部署到函数计算 AgentRun 上，以及如何充分利用函数计算 AgentRun 的各种能力。为什么要部署到函数计算 AgentRun？在讨论具体的集成方案前，让我们先明确一个问题：如果你的 Agent 应用已经在本地或自建服务器上运行良好，为什么还要迁移到函数计算 AgentRun？答案很简单：从开发环境到生产环境，有一道巨大的鸿沟。 本地运行只需要考虑功能实现，但生产环境需要考虑性能、稳定性、成本、安全、可观测等一系列问题。函数计算 AgentRun 提供的不是又一个 Agent 框架，而是让你的 Agent 能够以企业级标准运行的完整基础设施。具体来说，部署到函数计算 AgentRun 后，你能获得：零运维的 Serverless 运行时（自动扩缩容、按量付费），企业级的 Sandbox 环境（高性能、安全隔离），模型高可用保障（自动熔断、多模型 Fallback），全链路可观测（完整的 Trace、成本归因），以及统一的工具和 MCP 管理。快速上手：5 分钟部署你的第一个 LangChain Agent让我们从最流行的 LangChain 框架开始，通过一个完整的例子展示如何将 LangChain Agent 部署到函数计算 AgentRun。第一步：安装 Serverless Devs函数计算 AgentRun 使用 Serverless Devs 作为部署工具。如果你有 Node.js 环境，一行命令即可安装：npm i -g @serverless-devs/s第二步：创建项目使用脚手架快速创建项目（注意：需要 Python 3.10 及以上版本）：# 初始化模板 s init agentrun-quick-start-langchain # 进入代码目录 cd agentrun-quick-start-langchain/code # 初始化虚拟环境并安装依赖 uv venv && uv pip install -r requirements.txt第三步：配置认证信息通过环境变量（建议使用 .env 文件）配置你的 AgentRun 访问凭证：export AGENTRUN_ACCESS_KEY_ID="your-access-key-id" export AGENTRUN_ACCESS_KEY_SECRET="your-access-key-secret" export AGENTRUN_ACCOUNT_ID="your-account-id" export AGENTRUN_REGION="cn-hangzhou"第四步：理解集成方式这是最关键的部分。打开生成的代码，你会看到集成非常简单：from agentrun.integration.langchain import model, sandbox_toolset from agentrun.server import AgentRunServer # 使用 AgentRun 的模型（自动享受高可用、熔断等能力） llm = model("<your-model-name>") # 使用 AgentRun 的 Sandbox 工具 tools = sandbox_toolset( template_name="<your-sandbox-name>", template_type=TemplateType.CODE_INTERPRETER, sandbox_idle_timeout_seconds=300, ) # 创建 LangChain Agent（和原来的代码完全一样） agent = create_agent( model=llm, tools=tools, system_prompt="你是一个智能助手" ) # 定义调用函数 def invoke_agent(request): result = agent.invoke({"messages": request.messages}) return result["messages"][-1].content # 启动 HTTP Server（提供 OpenAI 兼容的 API） AgentRunServer(invoke_agent=invoke_agent).start()核心要点：model() 函数返回的是 LangChain 可以直接使用的模型对象sandbox_toolset() 返回的是 LangChain Tools 列表你的 Agent 创建代码完全不需要改动AgentRunServer 自动处理 HTTP 请求，提供标准的 OpenAI API第五步：本地测试启动服务后，可以通过 HTTP 请求测试：curl 127.0.0.1:9000/v1/chat/completions \ -X POST \ -H "content-type: application/json" \ -d '{"messages": [{"role": "user", "content": "通过代码查询现在是几点?"}], "stream":true}'第六步：部署到生产环境项目中已经包含了 s.yaml 配置文件。你只需要修改其中的 role 字段为你的阿里云角色：role: acs:ram::{您的阿里云主账号 ID}:role/{您的阿里云角色名称}配置部署密钥：s config add # 按照引导输入 Access Key ID 和 Secret，记住密钥对名称（如 agentrun-deploy）执行部署：s deploy -a agentrun-deploy部署完成后，你会得到一个 HTTPS URL，就可以在生产环境调用你的 Agent 了。不同框架的集成案例函数计算 AgentRun 不仅支持 LangChain，还深度集成了主流的 Agent 开发框架。所有框架都遵循同样的理念：通过简单的适配层，让你的代码无缝迁移到函数计算 AgentRun，享受企业级能力。LangGraph：工作流编排LangGraph 是 LangChain 团队推出的工作流编排框架，适合构建复杂的多步骤 Agent。集成方式和 LangChain 类似：from agentrun.integration.langgraph import model, tools from langgraph.graph import StateGraph, MessagesState from langgraph.prebuilt import ToolNode # 使用 AgentRun 的模型和工具 llm = model("<your-model-name>").to_langgraph() agent_tools = tools() # 构建 LangGraph 工作流（和原来的代码一样） def call_model(state: MessagesState): messages = state["messages"] response = llm.invoke(messages) return {"messages": [response]} workflow = StateGraph(MessagesState) workflow.add_node("agent", call_model) workflow.add_node("tools", ToolNode(agent_tools)) workflow.set_entry_point("agent") # 定义条件边... app = workflow.compile() # 调用 result = app.invoke({"messages": [HumanMessage(content="查询上海天气")]})LangGraph 的优势是可以精确控制 Agent 的执行流程，比如条件分支、循环、并行执行等。部署到函数计算 AgentRun 后，这些复杂的工作流都能自动享受弹性伸缩和可观测能力。AgentScope：多智能体协作AgentScope 是阿里达摩院开源的多智能体框架，特别适合构建多 Agent 协作场景。集成方式：from agentrun.integration.agentscope import model, tools from agentscope.agent import ReActAgent from agentscope.tool import Toolkit # 使用 AgentRun 的模型和工具 llm = model("<your-model-name>").to_agentscope() agent_tools = tools() # 注册工具到 Toolkit toolkit = Toolkit() for tool in agent_tools: toolkit.register_tool_function(tool) # 创建 Agent（和原来的代码一样） agent = ReActAgent( name="assistant", sys_prompt="你是一个智能助手", model=llm, toolkit=toolkit, ) # 调用 result = await agent.reply(Msg(name="user", content="查询上海天气", role="user"))AgentScope 的优势是对多 Agent 系统的原生支持，包括 Agent 之间的通信、协调、记忆共享等。部署到函数计算 AgentRun 后，每个 Agent 都在独立的隔离环境中运行，确保安全性。PydanticAI：类型安全的 Agent 框架PydanticAI 是一个新兴框架，强调类型安全和结构化输出。集成方式：from agentrun.integration.pydantic_ai import model, tools from pydantic_ai import Agent # 使用 AgentRun 的模型和工具 llm = model("<your-model-name>").to_pydantic_ai() agent_tools = tools() # 创建 Agent agent = Agent( llm, instructions="Be concise, reply with one sentence.", tools=agent_tools, ) # 同步调用 result = agent.run_sync("上海的天气如何？") # 异步调用 result = await agent.run("上海的天气如何？")PydanticAI 的优势是强类型和结构化输出，特别适合需要严格数据验证的企业场景。充分利用函数计算 AgentRun 的核心能力将 Agent 部署到函数计算 AgentRun 后，你不仅获得了 Serverless 运行环境，还可以深度利用平台提供的各种企业级能力。模型高可用：告别单点故障（搭配 AI 网关）部署到函数计算 AgentRun 后，你的 Agent 自动享受模型高可用能力。当你配置的主模型出现故障、限流或超时时，系统会自动切换到备用模型，整个过程对你的代码完全透明。在函数计算 AgentRun 控制台配置模型时可以和 AI 网关进行联动，可以设置：主模型（如 GPT-4），备用模型列表（如 Claude-3、Qwen-Max），熔断策略（错误率阈值、超时时间），负载均衡策略（轮询、权重、最少连接）。你的代码完全不需要改动，只需要在创建模型时使用函数计算 AgentRun 的模型名称，所有的容错、切换、负载均衡都由平台自动处理。企业级 Sandbox：安全执行代码函数计算 AgentRun 提供的 Sandbox 不是简单的代码执行环境，而是企业级的安全隔离沙箱。每个 Sandbox 实例都是独立隔离的，支持多种执行类型：Code Interpreter 支持 Python、Node.js、Java、Bash 等语言，可以执行数据分析、文件处理等任务。Browser Tool 提供浏览器自动化能力，支持网页爬取、表单填写、截图等操作。All In One 集成了代码解释器和浏览器工具，提供更丰富的交互能力。使用时，通过 sandbox_toolset() 函数就可以获取相应的工具集合，这些工具会自动转换为你使用的框架所需的格式。工具和 MCP：标准化集成函数计算 AgentRun 提供统一的工具管理和 MCP（Model Context Protocol）机制。你可以从工具市场选择现成的工具，也可以自定义工具并发布到市场。更强大的是 MCP 的 Hook 机制。通过前置 Hook，可以在工具调用前自动注入用户凭证、记录请求日志、校验参数合法性。通过后置 Hook，可以对结果进行转换、记录审计日志、处理异常情况。这些通用逻辑不需要在每个工具中重复实现，大大提升了开发效率。全链路可观测：不再是黑盒这是函数计算 AgentRun 最强大的能力之一。你的代码不需要做任何改动，平台会自动记录 Agent 的完整执行链路。在可观测平台上，你可以看到：Agent 接收到用户请求的时间和内容，调用了哪个模型、使用了多少 Token、花费了多少钱，调用了哪些工具、每个工具的执行时间和结果，访问了哪些知识库、检索了多少数据，每个环节的耗时分布，完整的调用链 Trace。这些能力都是平台自动提供的，通过探针注入实现，无论是高代码还是低代码创建的 Agent，都自动享受这些可观测能力。记忆和知识库：数据不出域函数计算 AgentRun 深度集成了 RAGFlow、Mem0 等开源项目，提供灵活的记忆和知识库管理。你可以选择一键托管模式，由平台统一管理部署运维，享受 Serverless 的弹性和按量付费优势。也可以选择绑定模式，将 Agent 连接到已经部署在企业 VPC 或 IDC 内的实例，数据完全不出企业内网。这种灵活性让你可以根据数据的敏感级别选择不同的策略：核心业务数据私有化部署，一般数据托管上云，在安全性和便利性之间找到最佳平衡。立即体验函数计算 AgentRun函数计算 AgentRun 的无代码到高代码演进能力，现已开放体验：快速创建： 访问控制台（ https://functionai.console.aliyun.com/cn-hangzhou/agent/explore ），60 秒创建你的第一个 Agent深度定制： 当需要更复杂功能时，一键转换为高代码持续演进： 利用函数计算 AgentRun 的基础设施能力，持续优化你的 Agent从想法到上线，从原型到生产，函数计算 AgentRun 始终是你最好的伙伴。欢迎加入“函数计算 AgentRun 客户群”，钉钉群号：134570017218。快速了解函数计算 AgentRun：一句话介绍：函数计算 AgentRun 是一个以高代码为核心的一站式 Agentic AI 基础设施平台。秉持生态开放和灵活组装的理念，为企业级 Agent 应用提供从开发、部署到运维的全生命周期管理。函数计算 AgentRun 架构图AgentRun 运行时基于阿里云函数计算 FC 构建，继承了 Serverless 计算极致弹性、按量付费、零运维的核心优势。通过深度集成 AgentScope、LangChain、RAGFlow、Mem0 等主流开源生态。函数计算 AgentRun 将 Serverless 的极致弹性、零运维和按量付费的特性与 AI 原生应用场景深度融合，助力企业实现成本与效率的极致优化，平均 TCO 降低 60% 。 让 开发者只需专注于 Agent 的业务逻辑创新，无需关心底层基础设施， 让 Agentic AI 真正进入企业生产环境。
作者：辰泉前言在 Agentic AI 时代，智能体需要与真实世界交互，而浏览器是连接虚拟世界与现实世界的重要桥梁。AgentRun Browser Sandbox 为智能体提供了安全、高性能、免运维的浏览器执行环境，让 AI Agent 真正具备“上网”的能力——从网页抓取、信息提取到表单填写、自动化操作，一切皆可实现。AgentRun Browser Sandbox 介绍什么是 Browser Sandbox?Browser Sandbox 是 AgentRun 平台提供的云原生无头浏览器沙箱服务，基于阿里云函数计算（FC）构建。它为智能体提供了一个安全隔离的浏览器执行环境，支持通过标准的 Chrome DevTools Protocol (CDP) 远程控制浏览器实例。核心特性无头浏览器能力内置 Chromium/Chrome 浏览器，支持完整的 Web 标准原生兼容 Puppeteer、Playwright 等主流自动化框架支持通过 CDP 协议进行精细化控制实时可视化内置 VNC 服务，支持实时查看浏览器界面提供操作录制功能，方便调试和回放支持通过 noVNC 客户端在网页中直接交互安全与隔离每个沙箱实例运行在独立的容器环境中文件系统和进程空间完全隔离支持 WSS 加密传输，确保数据安全Serverless 架构按需创建，按量付费，无需提前预置资源快速弹性伸缩，支持高并发场景零运维，无需管理服务器和浏览器依赖主要应用场景AI Agent 赋能： 为大模型提供“眼睛”和“手”，执行网页浏览、信息提取、在线操作等任务自动化测试： 在云端运行端到端（E2E）测试和视觉回归测试数据采集： 稳定、高效地进行网页抓取，应对动态加载和反爬虫挑战内容生成： 自动化生成网页截图或 PDF 文档上手使用 AgentRun Browser SandboxAgentRun SDK 快速介绍后续的内容将基于 AgentRun SDK 进行，因此我们先对 SDK 进行简要介绍。Agentrun SDK 是一个开源的开发者工具包，本期介绍 Python 版本。其旨在简化智能体与 AgentRun 平台各种服务（包括 Browser Sandbox）的集成。它提供了统一的接口，让您可以用几行代码就将沙箱能力集成到现有的 Agent 框架中。SDK 的核心功能如下：统一集成接口提供对 LangChain、AgentScope 等主流框架的开箱即用支持统一的模型代理接口，简化多模型管理标准化的工具注册机制Sandbox 生命周期管理自动创建和销毁沙箱实例支持会话级别的状态保持灵活的资源配置和超时控制安装 AgentRun SDKpip install agentrun-sdk[playwright,server]注意： 确保您的 Python 环境版本在 3.10 及以上。基本使用示例以下是使用 AgentRun SDK 创建和管理 Browser Sandbox 的核心代码：from agentrun.sandbox import Sandbox, TemplateType from playwright.sync_api import sync_playwright # 创建 Browser Sandbox sandbox = Sandbox.create( template_type=TemplateType.BROWSER, template_name="your-template-name", sandbox_idle_timeout_seconds=300 ) # 获取 CDP URL（用于 Playwright 连接） cdp_url = sandbox.get_cdp_url() # 使用 Playwright 连接并操作 with sync_playwright() as p: browser = p.chromium.connect_over_cdp(cdp_url) page = browser.contexts[0].pages[0] page.goto("https://www.example.com") page.screenshot(path="screenshot.png") browser.close() # 销毁 Sandbox sandbox.delete()关键概念：template_name： 控制台创建的浏览器环境模板cdp_url： 用于 Playwright/Puppeteer 连接vnc_url： 用于实时查看浏览器画面（可通过 sandbox.get_cdp_url() 获取）注意： 由于所有浏览器操作都在云端进行，您无需在本地安装浏览器。Playwright 仅用于通过 CDP 协议连接到云端的浏览器实例。如何创建 Sandbox 模板使用 Browser Sandbox 需要新建 Sandbox 模板，您需要访问 AgentRun 控制台网站 [ 1] ，并按照如下步骤创建模板：在顶部菜单栏选择“运行时与沙箱”；在左侧边栏选择“Sandbox 沙箱”；点击右上角“创建沙箱模板”；选择“浏览器”；在弹出的抽屉对话框中填写和选择您的模板的规格、网络等配置，并复制模板名称；点击“创建浏览器”等待其就绪即可。从零开始用 LangChain 创建 Browser Sandbox 智能体本教程将指导您从零开始创建一个完整的 Browser Sandbox 智能体项目。基于 LangChain 集成 Browser Sandbox本教程将详细讲解如何使用 LangChain 创建 Browser Sandbox 相关的 Tools 并集成到 Agent 中。项目结构为了保持代码的内聚性和可维护性，我们将代码拆分为以下模块：模块职责划分：sandbox_manager.py：负责 Sandbox 的创建、管理和销毁，提供统一的接口langchain_agent.py：负责创建 LangChain Tools 和 Agent，集成 VNC 信息main.py：作为入口文件，演示如何使用上述模块步骤 1：创建项目并安装依赖首先创建项目目录（如果还没有）：mkdir -p langchain-demo cd langchain-demo创建 requirements.txt 文件，内容如下：# LangChain 核心库 langchain>=0.1.0 langchain-openai>=0.0.5 langchain-community>=0.0.20 # AgentRun SDK agentrun-sdk[playwright,server]>=0.0.8 # 浏览器自动化 playwright>=1.40.0 # 环境变量管理 python-dotenv>=1.0.0然后安装依赖：pip install -r requirements.txt主要依赖说明：langchain 和 langchain-openai：LangChain 核心库agentrun-sdk[playwright,server]：AgentRun SDK，用于 Sandbox 管理playwright：浏览器自动化库python-dotenv：环境变量管理步骤 2：配置环境变量在项目根目录创建 .env 文件，配置以下环境变量：# 阿里云百炼平台的 API Key，用于调用大模型能力 # 请前往 https://bailian.console.aliyun.com/?tab=app#/api-key 创建和查看 DASHSCOPE_API_KEY=sk-your-bailian-api-key # 阿里云账号的访问密钥 ID 和访问密钥 Secret，用于 AgentRun SDK 鉴权 ALIBABA_CLOUD_ACCESS_KEY_ID=your-ak ALIBABA_CLOUD_ACCESS_KEY_SECRET=your-sk ALIBABA_CLOUD_ACCOUNT_ID=your-main-account-id ALIBABA_CLOUD_REGION=cn-hangzhou # browser sandbox 模板的名称，可以在 https://functionai.console.aliyun.com/cn-hangzhou/agent/runtime/sandbox 控制台创建 BROWSER_TEMPLATE_NAME=sandbox-your-template-name # agentrun 的控制面和数据面的 API 端点请求地址，默认cn-hangzhou AGENTRUN_CONTROL_ENDPOINT=agentrun.cn-hangzhou.aliyuncs.com AGENTRUN_DATA_ENDPOINT=https://${your-main-account-id}.agentrun-data.cn-hangzhou.aliyuncs.com步骤 3：创建 Sandbox 生命周期管理模块创建 sandbox_manager.py 文件，负责 Sandbox 的创建、管理和销毁。核心代码如下：""" Sandbox 生命周期管理模块 负责 AgentRun Browser Sandbox 的创建、管理和销毁。 提供统一的接口供 LangChain Agent 使用。 """ import os from typing import Optional, Dict, Any from dotenv import load_dotenv # 加载环境变量 load_dotenv() class SandboxManager: """Sandbox 生命周期管理器""" def __init__(self): self._sandbox: Optional[Any] = None self._sandbox_id: Optional[str] = None self._cdp_url: Optional[str] = None self._vnc_url: Optional[str] = None def create( self, template_name: Optional[str] = None, idle_timeout: int = 3000 ) -> Dict[str, Any]: """ 创建或获取一个浏览器 sandbox 实例 Args: template_name: Sandbox 模板名称，如果为 None 则从环境变量读取 idle_timeout: 空闲超时时间（秒），默认 3000 秒 Returns: dict: 包含 sandbox_id, cdp_url, vnc_url 的字典 Raises: RuntimeError: 创建失败时抛出异常 """ try: from agentrun.sandbox import Sandbox, TemplateType # 如果已有 sandbox，直接返回 if self._sandbox is not None: return self.get_info() # 从环境变量获取模板名称 if template_name is None: template_name = os.getenv( "BROWSER_TEMPLATE_NAME", "sandbox-browser-demo" ) # 创建 sandbox self._sandbox = Sandbox.create( template_type=TemplateType.BROWSER, template_name=template_name, sandbox_idle_timeout_seconds=idle_timeout ) self._sandbox_id = self._sandbox.sandbox_id self._cdp_url = self._get_cdp_url() self._vnc_url = self._get_vnc_url() return self.get_info() except ImportError as e: print(e) raise RuntimeError( "agentrun-sdk 未安装，请运行: pip install agentrun-sdk[playwright,server]" ) except Exception as e: raise RuntimeError(f"创建 Sandbox 失败: {str(e)}") def get_info(self) -> Dict[str, Any]: """ 获取当前 sandbox 的信息 Returns: dict: 包含 sandbox_id, cdp_url, vnc_url 的字典 Raises: RuntimeError: 如果没有活动的 sandbox """ if self._sandbox is None: raise RuntimeError("没有活动的 sandbox，请先创建") return { "sandbox_id": self._sandbox_id, "cdp_url": self._cdp_url, "vnc_url": self._vnc_url, } def get_cdp_url(self) -> Optional[str]: """获取 CDP URL""" return self._sandbox.get_cdp_url() def get_vnc_url(self) -> Optional[str]: """获取 VNC URL""" return self._sandbox.get_vnc_url() def get_sandbox_id(self) -> Optional[str]: """获取 Sandbox ID""" return self._sandbox_id def destroy(self) -> str: """ 销毁当前的 sandbox 实例 Returns: str: 操作结果描述 """ if self._sandbox is None: return "没有活动的 sandbox" try: sandbox_id = self._sandbox_id # 尝试销毁 sandbox if hasattr(self._sandbox, 'delete'): self._sandbox.delete() elif hasattr(self._sandbox, 'stop'): self._sandbox.stop() elif hasattr(self._sandbox, 'destroy'): self._sandbox.destroy() # 清理状态 self._sandbox = None self._sandbox_id = None self._cdp_url = None self._vnc_url = None return f"Sandbox 已销毁: {sandbox_id}" except Exception as e: # 即使销毁失败，也清理本地状态 self._sandbox = None self._sandbox_id = None self._cdp_url = None self._vnc_url = None return f"销毁 Sandbox 时出错: {str(e)}" def is_active(self) -> bool: """检查 sandbox 是否活跃""" return self._sandbox is not None def __enter__(self): """上下文管理器入口""" return self def __exit__(self, exc_type, exc_val, exc_tb): """上下文管理器退出，自动销毁""" self.destroy() return False # 全局单例（可选，用于简单场景） _global_manager: Optional[SandboxManager] = None def get_global_manager() -> SandboxManager: """获取全局 SandboxManager 单例""" global _global_manager if _global_manager is None: _global_manager = SandboxManager() return _global_manager def reset_global_manager(): """重置全局 SandboxManager""" global _global_manager if _global_manager: _global_manager.destroy() _global_manager = None关键功能：创建 Sandbox： 使用 AgentRun SDK 创建浏览器 Sandbox获取连接信息： 自动获取 CDP URL 和 VNC URL，支持多种属性名兼容生命周期管理： 提供销毁方法，确保资源正确释放步骤 4：创建 LangChain Tools 和 Agent创建 langchain_agent.py 文件，定义 LangChain Tools 并创建 Agent。核心代码如下：""" LangChain Agent 和 Tools 注册模块 负责创建 LangChain Agent，注册 Sandbox 相关的 tools，并集成 VNC 可视化。 本模块使用 sandbox_manager.py 中封装的 SandboxManager 来管理 sandbox 生命周期。 """ import os from dotenv import load_dotenv from langchain.tools import tool from langchain_openai import ChatOpenAI from langchain.agents import create_agent from pydantic import BaseModel, Field # 导入 sandbox 管理器 from sandbox_manager import SandboxManager # 加载环境变量 load_dotenv() # 全局 sandbox 管理器实例（单例模式） _sandbox_manager: SandboxManager | None = None def get_sandbox_manager() -> SandboxManager: """获取 sandbox 管理器实例（单例模式）""" global _sandbox_manager if _sandbox_manager is None: _sandbox_manager = SandboxManager() return _sandbox_manager # ============ LangChain Tools 定义 ============ @tool def create_browser_sandbox( template_name: str = None, idle_timeout: int = 3000 ) -> str: """创建或获取一个浏览器 sandbox 实例。 当需要访问网页、执行浏览器操作时，首先需要创建 sandbox。 创建成功后，会返回 sandbox 信息，包括 VNC URL 用于可视化。 Args: template_name: Sandbox 模板名称，如果不提供则从环境变量 BROWSER_TEMPLATE_NAME 读取 idle_timeout: 空闲超时时间（秒），默认 3000 秒 Returns: Sandbox 信息字符串，包括 ID、CDP URL、VNC URL """ try: manager = get_sandbox_manager() # 如果 template_name 为空字符串，转换为 None 以便从环境变量读取 if template_name == "": template_name = None info = manager.create(template_name=template_name, idle_timeout=idle_timeout) result = f"""✅ Sandbox 创建成功！ 📋 Sandbox 信息: - ID: {info['sandbox_id']} - CDP URL: {info['cdp_url']} """ vnc_url = info.get('vnc_url') if vnc_url: result += f"- VNC URL: {vnc_url}\n\n" result += "提示: VNC 查看器应该已自动打开，您可以在浏览器中实时查看浏览器操作。" else: result += "\n警告: 未获取到 VNC URL，可能无法使用可视化功能。" return result except Exception as e: return f" 创建 Sandbox 失败: {str(e)}" @tool def get_sandbox_info() -> str: """获取当前 sandbox 的详细信息，包括 ID、CDP URL、VNC URL 等。 当需要查看当前 sandbox 状态或获取 VNC 连接信息时使用此工具。 Returns: Sandbox 信息字符串 """ try: manager = get_sandbox_manager() info = manager.get_info() result = f"""📋 当前 Sandbox 信息: - Sandbox ID: {info['sandbox_id']} - CDP URL: {info['cdp_url']} """ if info.get('vnc_url'): result += f"- VNC URL: {info['vnc_url']}\n\n" result += "您可以使用 VNC URL 在浏览器中实时查看操作过程。\n" result += " 推荐使用 vnc.html 文件或 noVNC 客户端。" return result except RuntimeError as e: return f" {str(e)}" except Exception as e: return f" 获取 Sandbox 信息失败: {str(e)}" class NavigateInput(BaseModel): """浏览器导航输入参数""" url: str = Field(description="要访问的网页 URL，必须以 http:// 或 https:// 开头") wait_until: str = Field( default="load", description="等待页面加载的状态: load, domcontentloaded, networkidle" ) timeout: int = Field( default=30000, description="超时时间（毫秒），默认 30000" ) @tool(args_schema=NavigateInput) def navigate_to_url(url: str, wait_until: str = "load", timeout: int = 30000) -> str: """使用 sandbox 中的浏览器导航到指定 URL。 当用户需要访问网页时使用此工具。导航后可以在 VNC 中实时查看页面。 Args: url: 要访问的网页 URL wait_until: 等待页面加载的状态（load/domcontentloaded/networkidle） timeout: 超时时间（毫秒） Returns: 导航结果描述 """ try: manager = get_sandbox_manager() if not manager.is_active(): return " 错误: 请先创建 sandbox" # 验证 URL if not url.startswith(("http://", "https://")): return f" 错误: 无效的 URL 格式: {url}" cdp_url = manager.get_cdp_url() if not cdp_url: return " 错误: 无法获取 CDP URL" # 使用 Playwright 连接浏览器并导航 try: from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.connect_over_cdp(cdp_url) pages = browser.contexts[0].pages if browser.contexts else [] if pages: page = pages[0] else: page = browser.new_page() page.goto(url, wait_until=wait_until, timeout=timeout) title = page.title() return f"已成功导航到: {url}\n📄 页面标题: {title}\n💡 您可以在 VNC 中查看页面内容。" except ImportError: return f"导航指令已发送: {url}\n💡 提示: 安装 playwright 以启用实际导航功能 (pip install playwright)" except Exception as e: return f" 导航失败: {str(e)}" except Exception as e: return f" 操作失败: {str(e)}" @tool("browser_screenshot", description="在浏览器 sandbox 中截取当前页面截图") def take_screenshot(filename: str = "screenshot.png") -> str: """截取浏览器当前页面的截图。 Args: filename: 截图文件名，默认 "screenshot.png" Returns: 操作结果 """ try: manager = get_sandbox_manager() if not manager.is_active(): return " 错误: 请先创建 sandbox" cdp_url = manager.get_cdp_url() if not cdp_url: return " 错误: 无法获取 CDP URL" try: from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.connect_over_cdp(cdp_url) pages = browser.contexts[0].pages if browser.contexts else [] if pages: page = pages[0] else: return " 错误: 没有打开的页面" page.screenshot(path=filename) return f"截图已保存: {filename}" except ImportError: return " 错误: 需要安装 playwright (pip install playwright)" except Exception as e: return f" 截图失败: {str(e)}" except Exception as e: return f" 操作失败: {str(e)}" @tool("destroy_sandbox", description="销毁当前的 sandbox 实例，释放资源。注意：仅在程序退出或明确需要释放资源时使用，不要在一轮对话后销毁。") def destroy_sandbox() -> str: """销毁当前的 sandbox 实例。 重要提示：此工具应该仅在以下情况使用： - 程序即将退出 - 明确需要释放资源 - 用户明确要求销毁 不要在一轮对话完成后就销毁 sandbox，因为 sandbox 可以在多轮对话中复用。 Returns: 操作结果 """ try: manager = get_sandbox_manager() result = manager.destroy() return result except Exception as e: return f" 销毁失败: {str(e)}" # ============ Agent 创建 ============ def create_browser_agent(system_prompt: str = None): """ 创建带有 sandbox 工具的 LangChain Agent Args: system_prompt: 自定义系统提示词，如果为 None 则使用默认提示词 Returns: LangChain Agent 实例 """ # 配置 DashScope API api_key = os.getenv("DASHSCOPE_API_KEY") if not api_key: raise ValueError("请设置环境变量 DASHSCOPE_API_KEY") base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1" model_name = os.getenv("QWEN_MODEL", "qwen-plus") # 创建 LLM model = ChatOpenAI( model=model_name, api_key=api_key, base_url=base_url, temperature=0.7, ) # 创建工具列表 tools = [ create_browser_sandbox, get_sandbox_info, navigate_to_url, take_screenshot, destroy_sandbox, ] # 默认系统提示词 if system_prompt is None: system_prompt = """你是一个浏览器自动化助手，可以使用 sandbox 来访问和操作网页。 当用户需要访问网页时，请按以下步骤操作： 1. 首先创建或获取 sandbox（如果还没有） 2. 使用 navigate_to_url 导航到目标网页 3. 执行用户请求的操作 4. 如果需要，可以截取截图 重要提示： - 创建 sandbox 后，会返回 VNC URL，用户可以使用它实时查看浏览器操作 - 所有操作都会在 VNC 中实时显示，方便调试和监控 - sandbox 可以在多轮对话中复用，不要在一轮对话完成后就销毁 - 只有在用户明确要求销毁时才使用 destroy_sandbox 工具 - 不要主动建议用户销毁 sandbox，除非用户明确要求 - 请始终用中文回复，确保操作准确、高效。""" # 创建 Agent agent = create_agent( model=model, tools=tools, system_prompt=system_prompt, ) return agent def get_available_tools(): """获取所有可用的工具列表""" return [ create_browser_sandbox, get_sandbox_info, navigate_to_url, take_screenshot, destroy_sandbox, ]关键要点：Tool 定义： 使用 @tool 装饰器定义 LangChain Tools类型提示： 所有参数必须有类型提示，用于生成工具 schema文档字符串： 详细的文档字符串帮助 LLM 理解何时使用工具**单例模式： 使用全局管理器实例确保 Sandbox 在会话中复用**步骤 5：创建主入口文件创建 main.py 文件，作为程序入口。核心代码如下：""" LangChain + AgentRun Browser Sandbox 集成示例 主入口文件，演示如何使用 LangChain Agent 与 AgentRun Browser Sandbox 集成。 """ import os import sys import signal import webbrowser import urllib.parse import threading import http.server import socketserver from pathlib import Path from dotenv import load_dotenv from langchain_agent import create_browser_agent, get_sandbox_manager # 加载环境变量 load_dotenv() # 全局 HTTP 服务器实例 _http_server = None _http_port = 8080 # 全局清理标志，用于防止重复清理 _cleanup_done = False def start_http_server(): """启动一个简单的 HTTP 服务器来提供 vnc.html""" global _http_server if _http_server is not None: return _http_port try: current_dir = Path(__file__).parent.absolute() class VNCRequestHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(current_dir), **kwargs) def log_message(self, format, *args): # 静默日志，避免输出过多信息 pass # 尝试启动服务器 for port in range(_http_port, _http_port + 10): try: server = socketserver.TCPServer(("", port), VNCRequestHandler) server.allow_reuse_address = True # 在后台线程中运行服务器 def run_server(): server.serve_forever() thread = threading.Thread(target=run_server, daemon=True) thread.start() _http_server = server return port except OSError: continue return None except Exception as e: print(f"启动 HTTP 服务器失败: {str(e)}") return None def open_vnc_viewer(vnc_url: str): """ 自动打开 VNC 查看器并设置 VNC URL Args: vnc_url: VNC WebSocket URL """ if not vnc_url: return try: # 获取当前文件所在目录 current_dir = Path(__file__).parent.absolute() vnc_html_path = current_dir / "vnc.html" # 检查文件是否存在 if not vnc_html_path.exists(): print(f"警告: vnc.html 文件不存在: {vnc_html_path}") print_vnc_info(vnc_url) return # 启动 HTTP 服务器 port = start_http_server() if port: # 编码 VNC URL 作为 URL 参数 encoded_url = urllib.parse.quote(vnc_url, safe='') # 构建 HTTP URL http_url = f"http://localhost:{port}/vnc.html?url={encoded_url}" # 打开浏览器 print(f"\n正在打开 VNC 查看器...") print(f"HTTP 服务器运行在: http://localhost:{port}") print(f"VNC URL: {vnc_url[:80]}...") print(f"完整 URL: {http_url[:100]}...") webbrowser.open(http_url) print(f"VNC 查看器已打开") print(f"VNC URL 已通过 URL 参数自动设置，页面加载后会自动连接") else: # 如果 HTTP 服务器启动失败，尝试使用 file:// 协议 print(f"HTTP 服务器启动失败，尝试使用文件协议...") encoded_url = urllib.parse.quote(vnc_url, safe='') file_url = f"file://{vnc_html_path}?url={encoded_url}" webbrowser.open(file_url) print(f"VNC 查看器已打开（使用文件协议）") print(f"提示: 如果无法自动连接，请手动复制 VNC URL 到输入框") except Exception as e: print(f"自动打开 VNC 查看器失败: {str(e)}") print_vnc_info(vnc_url) def print_vnc_info(vnc_url: str): """打印 VNC 连接信息""" if not vnc_url: return print("\n" + "=" * 60) print("VNC 可视化连接信息") print("=" * 60) print(f"\nVNC URL: {vnc_url}") print("\n使用方式:") print(" 1. 使用 noVNC 客户端连接") print(" 2. 或在浏览器中访问 VNC 查看器页面") print(" 3. 实时查看浏览器操作过程") print("\n" + "=" * 60 + "\n") def cleanup_sandbox(): """ 清理 sandbox 资源 这个函数可以被信号处理器、异常处理器和正常退出流程调用 """ global _cleanup_done # 防止重复清理 if _cleanup_done: return _cleanup_done = True try: manager = get_sandbox_manager() if manager.is_active(): print("\n" + "=" * 60) print("正在清理 sandbox...") print("=" * 60) result = manager.destroy() print(f"清理结果: {result}\n") else: print("\n没有活动的 sandbox 需要清理\n") except Exception as e: print(f"\n清理 sandbox 时出错: {str(e)}\n") def signal_handler(signum, frame): """ 信号处理器，处理 Ctrl+C (SIGINT) 和其他信号 Args: signum: 信号编号 frame: 当前堆栈帧 """ print("\n\n收到中断信号，正在清理资源...") cleanup_sandbox() print("清理完成") sys.exit(0) def main(): """主函数""" global _cleanup_done # 重置清理标志 _cleanup_done = False # 注册信号处理器，处理 Ctrl+C (SIGINT) signal.signal(signal.SIGINT, signal_handler) # 在 Windows 上，SIGBREAK 也可以处理 if hasattr(signal, 'SIGBREAK'): signal.signal(signal.SIGBREAK, signal_handler) print("=" * 60) print("LangChain + AgentRun Browser Sandbox 集成示例") print("=" * 60) print() try: # 创建 Agent print("正在初始化 LangChain Agent...") agent = create_browser_agent() print("Agent 初始化完成\n") # 示例查询 queries = [ "创建一个浏览器 sandbox", "获取当前 sandbox 的信息，包括 VNC URL", "导航到 https://www.aliyun.com", "截取当前页面截图", ] # 执行查询 for i, query in enumerate(queries, 1): print(f"\n{'=' * 60}") print(f"查询 {i}: {query}") print(f"{'=' * 60}\n") try: result = agent.invoke({ "messages": [{"role": "user", "content": query}] }) # 提取最后一条消息的内容 output = result.get("messages", [])[-1].content if isinstance(result.get("messages"), list) else result.get("output", str(result)) print(f"\n结果:\n{output}\n") # 如果是创建 sandbox，自动打开 VNC 查看器 if i == 1: try: # 等待一下确保 sandbox 完全创建 import time time.sleep(1) manager = get_sandbox_manager() if manager.is_active(): info = manager.get_info() vnc_url = info.get('vnc_url') if vnc_url: print(f"\n检测到 VNC URL: {vnc_url[:80]}...") open_vnc_viewer(vnc_url) print_vnc_info(vnc_url) else: print("\n警告: 未获取到 VNC URL，请检查 sandbox 创建是否成功") except Exception as e: print(f"打开 VNC 查看器时出错: {str(e)}") import traceback traceback.print_exc() # 如果是获取信息，显示 VNC 信息 elif i == 2: try: manager = get_sandbox_manager() if manager.is_active(): info = manager.get_info() if info.get('vnc_url'): print_vnc_info(info['vnc_url']) except: pass except Exception as e: print(f"查询失败: {str(e)}\n") import traceback traceback.print_exc() # 交互式查询 print("\n" + "=" * 60) print("进入交互模式（输入 'quit' 或 'exit' 退出，Ctrl+C 或 Ctrl+D 中断）") print("=" * 60 + "\n") while True: try: user_input = input("请输入您的查询: ").strip() except EOFError: # 处理 Ctrl+D (EOF) print("\n\n检测到输入结束 (Ctrl+D)，正在清理资源...") cleanup_sandbox() print("清理完成") break except KeyboardInterrupt: # 处理 Ctrl+C (在 input 调用期间) print("\n\n检测到中断信号 (Ctrl+C)，正在清理资源...") cleanup_sandbox() print("清理完成") break if not user_input: continue if user_input.lower() in ['quit', 'exit', '退出']: print("\nBye") # 退出前清理 sandbox cleanup_sandbox() break try: result = agent.invoke({ "messages": [{"role": "user", "content": user_input}] }) output = result.get("messages", [])[-1].content if isinstance(result.get("messages"), list) else result.get("output", str(result)) print(f"\n结果:\n{output}\n") # 检查是否需要打开或显示 VNC 信息 user_input_lower = user_input.lower() if "创建" in user_input_lower and "sandbox" in user_input_lower: # 如果是创建 sandbox，自动打开 VNC 查看器 try: # 等待一下确保 sandbox 完全创建 import time time.sleep(1) manager = get_sandbox_manager() if manager.is_active(): info = manager.get_info() vnc_url = info.get('vnc_url') if vnc_url: print(f"\n检测到 VNC URL: {vnc_url[:80]}...") open_vnc_viewer(vnc_url) print_vnc_info(vnc_url) else: print("\n警告: 未获取到 VNC URL，请检查 sandbox 创建是否成功") except Exception as e: print(f"打开 VNC 查看器时出错: {str(e)}") import traceback traceback.print_exc() elif "sandbox" in user_input_lower or "vnc" in user_input_lower: # 其他情况只显示信息 try: manager = get_sandbox_manager() if manager.is_active(): info = manager.get_info() if info.get('vnc_url'): print_vnc_info(info['vnc_url']) except: pass except Exception as e: print(f"查询失败: {str(e)}\n") import traceback traceback.print_exc() # 清理资源（仅在程序正常退出时） cleanup_sandbox() except KeyboardInterrupt: # 处理顶层 KeyboardInterrupt (Ctrl+C) print("\n\n检测到中断信号 (Ctrl+C)，正在清理资源...") cleanup_sandbox() print("清理完成") sys.exit(0) except EOFError: # 处理顶层 EOFError (Ctrl+D) print("\n\n检测到输入结束 (Ctrl+D)，正在清理资源...") cleanup_sandbox() print("清理完成") sys.exit(0) except ValueError as e: print(f"配置错误: {str(e)}") print("\n提示: 请确保已设置以下环境变量:") print(" - DASHSCOPE_API_KEY: DashScope API Key") print(" - ALIBABA_CLOUD_ACCOUNT_ID: 阿里云账号 ID") print(" - ALIBABA_CLOUD_ACCESS_KEY_ID: 访问密钥 ID") print(" - ALIBABA_CLOUD_ACCESS_KEY_SECRET: 访问密钥 Secret") print(" - ALIBABA_CLOUD_REGION: 区域（默认: cn-hangzhou）") except Exception as e: print(f"发生错误: {str(e)}") import traceback traceback.print_exc() # 发生错误时也尝试清理 cleanup_sandbox() if __name__ == "__main__": main()关键功能：VNC 自动打开： 创建 Sandbox 后自动打开 VNC 查看器信号处理： 捕获 Ctrl+C，确保资源正确清理交互模式： 支持持续对话，复用 Sandbox 实例VNC 可视化集成VNC（Virtual Network Computing）功能允许您实时查看和监控浏览器在 Sandbox 中的操作过程，这对于调试和监控 Agent 行为非常有用。获取 VNC URL：创建 Sandbox 后，可以通过 get_sandbox_info tool 获取 VNC URL：# 通过 Agent 调用 result = agent.invoke({ "messages": [{"role": "user", "content": "获取 sandbox 信息"}] }) # 或直接通过管理器获取 manager = get_sandbox_manager() info = manager.get_info() vnc_url = info['vnc_url']自动打开 VNC 查看器：在 main.py 中，我们实现了自动打开 VNC 查看器的功能：import webbrowser import urllib.parse from pathlib import Path def open_vnc_viewer(vnc_url: str): """自动打开 VNC 查看器""" current_dir = Path(__file__).parent.absolute() vnc_html_path = current_dir / "vnc.html" if vnc_html_path.exists(): # 通过 URL 参数传递 VNC URL encoded_url = urllib.parse.quote(vnc_url, safe='') file_url = f"file://{vnc_html_path}?url={encoded_url}" webbrowser.open(file_url)VNC HTML 页面：vnc.html 页面会从 URL 参数中读取 VNC URL，并自动连接到 VNC 服务器。页面包含以下核心功能：noVNC 库加载： 从 CDN 动态加载 noVNC 客户端库自动连接： 读取 URL 参数中的 VNC URL 并自动连接状态显示： 显示连接状态（连接中、已连接、已断开）手动控制： 支持手动输入 VNC URL、断开重连等操作核心 JavaScript 代码片段：// 从 URL 参数获取 VNC URL const urlParams = new URLSearchParams(window.location.search); const vncUrl = urlParams.get('url'); // 加载 noVNC 库 async function loadNoVNC() { const module = await import('https://cdn.jsdelivr.net/gh/novnc/noVNC@v1.4.0/core/rfb.js'); return module.default; } // 连接 VNC async function connectVNC(url) { const RFB = await loadNoVNC(); rfb = new RFB(vncScreen, url, { shared: true, credentials: { password: '' } }); rfb.addEventListener('connect', () => { console.log('VNC 连接成功'); }); }完整的 vnc.html 文件可以在示例代码仓库中获取。手动使用 VNC 查看器：如果自动打开失败，您也可以手动使用 VNC 查看器：1. 使用 noVNC 在线客户端：访问 noVNC 在线客户端 [ 2]在连接设置中填入 VNC URL点击连接2. 使用本地 VNC HTML 页面：打开 vnc.html输入 VNC URL点击连接按钮实时监控功能：所有浏览器操作都会在 VNC 中实时显示可以看到 Agent 的每一步操作（导航、点击、输入等）方便调试和监控 Agent 行为支持交互式操作（在 VNC 中直接操作浏览器）运行和测试python main.py程序会自动：创建 Browser Sandbox打开 VNC 查看器（实时查看浏览器操作）执行预设查询进入交互模式工作原理为了更好地理解系统架构，我们将工作流程拆分为两个部分：LangChain Agent 工作流程和 SandboxManager 生命周期管理。1. LangChain Agent 工作流程下图展示了 LangChain Agent 如何处理用户请求并调用相应的 Tools：Agent 工作流程说明：请求接收： 用户发起自然语言请求（如“访问淘宝首页并截图”）意图分析： Agent 分析用户意图，决定需要调用哪些 ToolsTool 调用： 根据任务需求，顺序或组合调用多个 ToolsManager 交互： 所有 Tools 都通过 SandboxManager 单例实例操作 Sandbox结果处理： Agent 将 Tool 返回的结果整合成用户友好的响应多轮对话： Sandbox 在整个会话中保持活跃，支持多轮对话5 个核心 Tools 的职责：2. SandboxManager 生命周期管理下图展示了 SandboxManager 如何管理 Sandbox 的完整生命周期：SandboxManager 工作流程说明：1. 单例管理：首次调用时创建 Manager 实例后续调用复用同一个实例确保整个会话只有一个 Sandbox2. Sandbox 创建：调用 AgentRun SDK 的 Sandbox.create()SDK 通过阿里云 API 与函数计算 FC 通信FC 服务创建独立的容器实例，包含：Chromium 浏览器 VNC 服务必要的运行环境3. 连接信息获取：CDP URL： WebSocket 地址，用于 Playwright/Puppeteer 远程控制浏览器VNC URL： WebSocket 地址，用于实时查看浏览器画面**4. 浏览器操作：Playwright 通过 CDP URL 连接到远程浏览器执行各种浏览器操作（导航、点击、截图等）VNC 同步显示操作过程，用户可实时监控5. 资源清理：调用 destroy() 方法销毁 Sandbox清理 Manager 内部状态通过 SDK 释放云端资源3. Agent 与 Manager 的协作关系交互模式：用户请求 → Agent → Tool → SandboxManager → AgentRun SDK → 云端 Sandbox ↓ 用户响应 ← Agent ← Tool ← SandboxManager ← 操作结果关键设计理念：分层架构：用户层：自然语言交互Agent 层：意图理解和任务分解Tool 层：功能封装和参数验证Manager 层：资源管理和状态维护SDK 层：云服务通信云端层：实际的 Sandbox 环境单例模式：SandboxManager 使用单例模式保证整个会话中只有一个 Sandbox 实例避免资源浪费和状态冲突状态复用：Sandbox 在多轮对话中保持活跃减少创建和销毁的开销提供更流畅的用户体验双通道设计：CDP 通道：Agent 通过 Playwright 控制浏览器VNC 通道：用户通过 VNC 查看器实时监控解耦设计：Tools 不直接操作 SDK，通过 Manager 统一管理便于扩展和维护统一的错误处理和资源管理典型使用场景示例：# 第 1 轮对话 用户: "创建一个 sandbox 并访问淘宝首页" → Agent 调用: create_browser_sandbox → navigate_to_url → Manager: 创建 Sandbox → Playwright 导航 → 结果: "Sandbox 已创建，已访问淘宝首页" # 第 2 轮对话（复用 Sandbox） 用户: "截取当前页面" → Agent 调用: take_screenshot → Manager: 使用现有 Sandbox → Playwright 截图 → 结果: "截图已保存" # 第 3 轮对话（复用 Sandbox） 用户: "访问京东首页" → Agent 调用: navigate_to_url → Manager: 使用现有 Sandbox → Playwright 导航 → 结果: "已访问京东首页"通过这种设计，Agent 专注于理解用户意图和任务编排，而 Manager 专注于 Sandbox 的生命周期管理，实现了清晰的职责分离。工作原理总结：工具注册：使用 @tool 装饰器将 Sandbox 功能封装为 LangChain Tools生命周期管理： SandboxManager 负责 Sandbox 的创建、管理和销毁状态保持：使用单例模式管理 Sandbox 实例，确保同一会话内复用VNC 集成：自动获取并返回 VNC URL，方便用户实时查看错误处理：所有工具都包含完善的错误处理机制扩展和定制添加自定义 Tools：@tool def extract_table_data(url: str) -> str: """从网页中提取表格数据""" from playwright.sync_api import sync_playwright manager = get_sandbox_manager() cdp_url = manager.get_info()['cdp_url'] with sync_playwright() as p: browser = p.chromium.connect_over_cdp(cdp_url) page = browser.contexts[0].pages[0] page.goto(url) tables = page.query_selector_all("table") return f"找到 {len(tables)} 个表格"自定义提示词：custom_prompt = """你是一个专业的网页数据提取助手。 在执行任务前，请先创建 sandbox，然后使用浏览器工具完成任务。""" agent = create_browser_agent(system_prompt=custom_prompt) 最佳实践模块化设计：将 Sandbox 管理和 Agent 创建分离，提高代码可维护性错误处理：所有工具都应包含完善的错误处理资源清理：使用信号处理器确保资源正确清理VNC 提示：在工具返回中包含 VNC URL，方便用户使用单例模式：确保 Sandbox 实例在会话中复用，避免重复创建前端集成可视化监控（VNC）VNC 集成架构下图展示了前端如何集成 VNC 实现实时监控：轻量级 HTML 页面集成创建一个简单的 vnc-viewer.html 文件： Browser Sandbox VNC 查看器 const params = new URLSearchParams(window.location.search); const vncUrl = params.get('url'); if (!vncUrl) { alert('请提供 VNC URL 参数'); } else { const module = await import('https://cdn.jsdelivr.net/gh/novnc/noVNC@v1.4.0/core/rfb.js'); const RFB = module.default; const rfb = new RFB( document.getElementById('vnc-container'), vncUrl, { shared: true, credentials: { password: '' } } ); rfb.scaleViewport = true; } 使用方式：import webbrowser import urllib.parse vnc_url = sandbox.vnc_url encoded_url = urllib.parse.quote(vnc_url, safe='') viewer_url = f"file:///path/to/vnc-viewer.html?url={encoded_url}" webbrowser.open(viewer_url)React 应用集成核心组件代码：import React, { useEffect, useRef } from 'react'; interface VNCViewerProps { vncUrl: string; onConnect?: () => void; onDisconnect?: () => void; } export const VNCViewer: React.FC<VNCViewerProps> = ({ vncUrl, onConnect, onDisconnect }) => { const containerRef = useRef<HTMLDivElement>(null); useEffect(() => { let rfb: any; const initVNC = async () => { if (!containerRef.current || !vncUrl) return; const { default: RFB } = await import('@novnc/novnc/core/rfb'); rfb = new RFB(containerRef.current, vncUrl, { shared: true, credentials: { password: '' } }); rfb.scaleViewport = true; rfb.addEventListener('connect', () => onConnect?.()); rfb.addEventListener('disconnect', () => onDisconnect?.()); }; initVNC(); return () => { if (rfb) rfb.disconnect(); }; }, [vncUrl, onConnect, onDisconnect]); return ( ); };使用示例：import React, { useState, useEffect } from 'react'; import { VNCViewer } from './VNCViewer'; function App() { const [vncUrl, setVncUrl] = useState<string>(''); useEffect(() => { fetch('/api/sandbox/create', { method: 'POST' }) .then(res => res.json()) .then(data => setVncUrl(data.vnc_url)); }, []); return ( Browser Sandbox 实时监控 {vncUrl ? ( <VNCViewer vncUrl={vncUrl} onConnect={() => console.log('已连接')} onDisconnect={() => console.log('已断开')} /> ) : ( 正在初始化... )} ); }Puppeteer 和 Playwright 直接集成如果您更熟悉传统的浏览器自动化库，也可以直接使用 Puppeteer 或 Playwright 连接到 Browser Sandbox。使用 Playwrightfrom playwright.sync_api import sync_playwright from agentrun.sandbox import Sandbox, TemplateType # 创建 Sandbox sandbox = Sandbox.create( template_type=TemplateType.BROWSER, template_name="your-template-name", sandbox_idle_timeout_seconds=3000 ) # 使用 Playwright 连接 with sync_playwright() as p: browser = p.chromium.connect_over_cdp(sandbox.cdp_url) page = browser.contexts[0].pages[0] # 执行操作 page.goto("https://www.example.com") page.screenshot(path="screenshot.png") content = page.content() browser.close() # 清理 sandbox.delete() 使用 Puppeteer（Node.js）const puppeteer = require('puppeteer-core'); // CDP URL 从 Sandbox 获取 const cdpUrl = 'wss://your-account.funagent-data-pre.cn-hangzhou.aliyuncs.com/sandboxes/xxx/ws/automation'; (async () => { const browser = await puppeteer.connect({ browserWSEndpoint: cdpUrl, defaultViewport: null }); const page = (await browser.pages())[0]; await page.goto('https://www.example.com'); await page.screenshot({ path: 'screenshot.png' }); await browser.close(); })();总结通过本教程，您已经学会了：AgentRun SDK 基础： 如何使用 SDK 创建和管理 Browser SandboxLangChain 集成： 如何将 Sandbox 封装为 LangChain ToolsVNC 可视化： 如何在前端集成 VNC 实现实时监控直接集成： 如何使用 Puppeteer/Playwright 直接连接 Sandbox相关链接：[1] Agentrun 控制台网站https://functionai.console.aliyun.com/cn-hangzhou/agent/runti...[2] noVNC 在线客户端https://novnc.com/noVNC/vnc.html