1. 项目概述:为什么要在 Modal 沙箱里跑 OpenAI Agents SDK?

OpenAI Agents SDK 是一套面向开发者设计的轻量级工具集,它不是官方正式发布的生产级框架,而是 OpenAI 工程团队在内部快速验证 Agent 架构时沉淀出的一套可复用模式——核心价值在于把 LLM 调用、工具绑定、状态管理、循环执行 这四件事封装成可组合、可测试、可插拔的 Python 类。它不依赖 LangChain 或 LlamaIndex,没有抽象层嵌套,代码干净到可以直接读源码理解行为逻辑。而 Modal 是一个以“函数即服务”为底层范式的云运行平台,它的沙箱(sandbox)本质是带完整 Linux 环境、预装 CUDA、支持任意 pip 包安装、可挂载卷、能保持长时间运行的隔离容器实例。把两者结合,不是为了炫技,而是解决一个非常现实的工程痛点: 如何让基于 OpenAI 的 Agent 在真实业务流中稳定、可控、可观测地长期运行,而不是卡在本地 Jupyter Notebook 里反复 Ctrl+C 重试

我第一次用 Agents SDK 写完一个能调用天气 API + 日历写入 + 邮件摘要的会议准备 Agent 后,立刻意识到问题:本地跑一次要等 8 秒(含模型响应+工具链路),调试 5 次就心累;换到 FastAPI 部署?得自己写路由、鉴权、限流、日志追踪、错误重试、上下文超时控制——这已经不是“跑 Agent”,是在造一个微服务中间件。Modal 沙箱的价值就在这里:它天然提供进程隔离、资源配额(CPU/GPU/内存)、自动扩缩容、内置日志流、结构化错误捕获,最关键的是—— 你写的 Agent 代码几乎不用改,只需要加几行装饰器和配置,就能从 python main.py 变成 modal run main.py::run_agent ,且全程可 debug、可中断、可复现 。这不是“部署”,是“运行环境对齐”。适合三类人:一是正在做 PoC 验证 Agent 流程是否成立的产品/算法同学;二是需要把 Agent 快速接入现有工作流(比如 Slack Bot、CRM 自动跟进)的后端工程师;三是想避开 LangChain 复杂生命周期、专注业务逻辑本身的技术负责人。它不承诺“开箱即用的多跳推理”,但保证“你写的每行 Python 都在受控环境中按预期执行”。

2. 整体架构设计与选型逻辑:为什么是 Modal,而不是其他平台?

2.1 核心矛盾:Agent 运行的本质需求 vs 平台能力匹配度

Agent 不是传统 Web API。它的典型生命周期包含:接收输入 → 解析意图 → 调用 1~N 个外部工具(可能串行/并行)→ 中间状态暂存(如临时文件、缓存结果)→ 多轮 LLM 调用(需维护 conversation history)→ 输出最终结果或触发下游动作。这个过程天然具备 长时延、非确定性、状态依赖、资源波动大 四个特征。我们来横向对比主流平台对这四点的支持程度:

平台类型 典型代表 是否支持长时延(>30s) 是否支持中间状态持久化 是否支持 GPU 加速(用于本地小模型 fallback) 是否支持细粒度资源控制(如 4GB RAM + 2 vCPU) 是否支持沙箱级隔离(避免工具调用污染全局环境)
Serverless 函数 AWS Lambda / Vercel ❌ 默认 15s,最高 15m(需特殊申请) ❌ 仅支持临时 /tmp ,无跨调用持久化 ❌ 无 GPU ❌ 内存与 CPU 绑定,无法独立调节 ⚠️ 进程隔离,但 /tmp 共享,易冲突
容器编排 Kubernetes / Docker Compose ✅(通过 PVC/Redis) ✅(需自建 GPU 节点) ✅(CPU/MEM 单独设置) ✅(Pod 级隔离)
FaaS+沙箱混合 Modal / CodeSandbox ✅(默认 24h,可设 timeout) ✅( modal.Volume 挂载持久卷) ✅( gpu="any" 或指定型号) ✅( cpu=2.0 , memory=4096 精确声明) ✅(每个 sandbox 是独立容器)

结论很清晰:K8s 功能最全,但学习成本高、运维负担重;Serverless 函数太“短平快”,不适合 Agent 的“慢思考”特性;Modal 则精准卡在中间——它用 Serverless 的易用性( modal deploy 一行命令),提供了接近 K8s 的控制力(资源声明、GPU、Volume)。尤其关键的是,Modal 的 sandbox 是 有状态的、可交互的、可 attach 的 。你可以 modal shell 进去查日志、 ps aux 看进程、 curl http://localhost:8000/debug 查当前 session,这在 Lambda 里根本不可想象。

2.2 为什么不是直接用 Modal Functions?Sandbox 的不可替代性在哪?

Modal 提供两种运行单元:Function(无状态、短时、自动扩缩)和 Sandbox(有状态、长时、手动管理)。初学者容易混淆,以为“Agent 就是个函数,用 Function 就行”。这是最大的认知陷阱。我踩过坑:用 @modal.function() 包裹一个 Agent.run(),结果发现:

  • 第二轮调用时, self.history 是空的(Function 每次启动新进程,对象不保留);
  • 调用外部工具返回的大 JSON(如 5MB 的 CRM 数据导出),在 Function 的 6MB 响应体限制下直接报错;
  • 想加个 time.sleep(10) 做轮询等待异步任务完成?Function 默认超时 300s,且无法感知 sleep 中断。

Sandbox 则完全不同:它是一个 持续运行的 Linux 进程容器 。你启动它,它就一直在线,直到你主动 sandbox.stop() 或超时。这意味着:

  • Agent 实例可以作为单例常驻内存, history tool_cache session_id 全部自然保留;
  • 你可以用 requests.Session() 复用连接,用 sqlite3.connect("/vol/db.sqlite") 存状态,用 threading.Event() 做同步;
  • 所有 stdout/stderr 实时流式输出到 Modal 控制台,配合 print(f"[DEBUG] Step 3: got {len(results)} items") ,调试体验接近本地开发。

所以选型逻辑不是“哪个平台更火”,而是“哪个平台的运行模型与 Agent 的生命周期模型最匹配”。Modal Sandbox 是目前唯一把“函数式部署”和“进程级控制”无缝融合的方案。

2.3 Agents SDK 的轻量化设计如何与 Modal 形成正向增强?

Agents SDK 的源码只有 3 个核心类: Agent (主调度器)、 Tool (工具基类)、 Message (消息结构体)。它刻意回避了:

  • 不做自动 tool discovery(不扫描 tools/ 目录);
  • 不做隐式 state serialization(不自动 pickle self );
  • 不做 retry/backoff 封装(交由用户在 Tool.call() 里实现)。

这种“裸金属”设计,反而和 Modal 的哲学高度一致: 平台只提供可靠、透明、可预测的运行环境,业务逻辑的复杂度必须显式表达 。比如,Agents SDK 要求你手动定义 get_tools() 返回工具列表,Modal 就要求你手动声明 @sandbox.cls() @sandbox.method() 。二者叠加,不会产生“魔法”,但会形成极强的可追溯性:你在 Modal 日志里看到 Calling weather_tool with {'city': 'Beijing'} ,就知道这行日志必然来自 WeatherTool.call() 方法,且该方法一定在 sandbox 的 /app/tools/weather.py 里。没有中间层混淆调用栈,没有抽象泄漏。这也是为什么我坚持用 Agents SDK 而非 LangChain:当你的 Agent 需要对接企业内网数据库(需 Kerberos 认证)、调用私有 SOAP 接口(需 WSDL 解析)、处理 GB 级 Excel(需 openpyxl 流式读取)时,越少的抽象层,越高的成功率。

3. 核心细节解析与实操要点:从零搭建可运行的 Agent Sandbox

3.1 环境初始化:Modal App 结构与依赖管理

Modal 的 App 是一个 Python 对象,它定义了所有可部署的组件。对于 Agent,我们需要三个核心部分:Sandbox 类(承载 Agent 实例)、入口函数(启动 sandbox)、工具模块(具体实现)。目录结构如下:

agent_project/
├── modal_app.py          # 主 App 定义
├── agent_core.py         # Agent 类继承与业务逻辑
├── tools/
│   ├── __init__.py
│   ├── weather.py        # 天气工具
│   └── calendar.py       # 日历工具
└── requirements.txt

requirements.txt 必须显式声明所有依赖,Modal 不会自动推断。Agents SDK 本身无 PyPI 包,需直接引用 OpenAI 官方 SDK 和 pydantic

openai==1.35.11
pydantic==2.7.1
requests==2.31.0
python-dateutil==2.8.2

注意:Modal 默认使用 Python 3.11,但 Agents SDK 的 Message 类依赖 pydantic.BaseModel model_dump() 方法(v2.0+),若用旧版 pydantic 会报 AttributeError 。务必锁死版本,不要写 pydantic>=2.0 。我曾因未锁定版本,在 Modal 自动升级后导致所有 sandbox 启动失败,错误日志只显示 ImportError: cannot import name 'model_dump' ,排查了 2 小时才发现是依赖漂移。

modal_app.py 是整个项目的“蓝图”。关键点在于 Sandbox 的声明方式:

from modal import App, Sandbox, Volume, Image
import os

# 创建共享存储卷,用于跨 sandbox 持久化数据
vol = Volume.from_name("agent-data", create_if_missing=True)

# 构建运行镜像:基础镜像 + 依赖安装 + 代码复制
image = (
    Image.debian_slim(python_version="3.11")
    .pip_install_from_requirements("requirements.txt")
    .copy_local_file("agent_core.py", "/root/agent_core.py")
    .copy_local_directory("tools", "/root/tools")
)

app = App("openai-agent-sandbox")

@app.function(
    image=image,
    volumes={"/data": vol},
    timeout=3600,  # 1小时超时,足够处理复杂流程
)
def run_agent(user_input: str):
    # 此函数在 sandbox 启动后执行,是 Agent 的入口
    from agent_core import MeetingPrepAgent
    agent = MeetingPrepAgent()
    result = agent.run(user_input)
    return result

这里有个极易忽略的细节: volumes={"/data": vol} 声明了挂载点,但 /data 目录在 sandbox 内部必须存在,否则启动失败。Modal 不会自动创建挂载路径。因此,必须在 Image 构建阶段显式创建:

image = (
    Image.debian_slim(python_version="3.11")
    .run_commands("mkdir -p /data")  # 关键!必须创建挂载目录
    .pip_install_from_requirements("requirements.txt")
    .copy_local_file("agent_core.py", "/root/agent_core.py")
    .copy_local_directory("tools", "/root/tools")
)

3.2 Agent 类改造:适配 Sandbox 的生命周期管理

Agents SDK 的 Agent 基类默认假设单次运行即结束。但在 sandbox 中,我们要让它“活”下来,响应多次请求。因此,不能直接 agent.run(input) ,而要将其改造为一个 状态机服务 。核心改造点有三处:

  1. run() 拆解为 process_input() get_response() 两个方法 ,分离输入解析与响应生成,便于在 sandbox 中循环调用;
  2. 引入 self.session_state 字典 ,存储当前会话的 history pending_tasks last_tool_result ,避免每次调用都重建;
  3. 增加 reset_session() 方法 ,用于清理状态,模拟新会话开始。

改造后的 MeetingPrepAgent 示例:

# agent_core.py
from openai import OpenAI
from typing import List, Dict, Any
from tools.weather import WeatherTool
from tools.calendar import CalendarTool

class MeetingPrepAgent:
    def __init__(self):
        self.client = OpenAI()  # 使用环境变量 OPENAI_API_KEY
        self.tools = [WeatherTool(), CalendarTool()]
        self.session_state = {
            "history": [],
            "pending_tasks": [],
            "last_tool_result": None
        }

    def process_input(self, user_input: str) -> List[Dict[str, Any]]:
        """解析用户输入,生成初始 message list"""
        system_msg = {"role": "system", "content": "You are a meeting preparation assistant..."}
        user_msg = {"role": "user", "content": user_input}
        self.session_state["history"] = [system_msg, user_msg]
        return self.session_state["history"]

    def get_response(self, messages: List[Dict[str, Any]]) -> str:
        """执行 Agent 循环,返回最终响应"""
        # 这里复用 Agents SDK 的核心 loop 逻辑
        while True:
            response = self.client.chat.completions.create(
                model="gpt-4-turbo",
                messages=messages,
                tools=self._get_tool_schemas(),
                tool_choice="auto"
            )
            msg = response.choices[0].message
            messages.append(msg)

            if msg.tool_calls:
                # 执行所有 tool calls
                for tool_call in msg.tool_calls:
                    tool_result = self._execute_tool(tool_call)
                    messages.append({
                        "role": "tool",
                        "content": tool_result,
                        "tool_call_id": tool_call.id
                    })
                continue
            else:
                return msg.content

    def _get_tool_schemas(self):
        return [tool.get_schema() for tool in self.tools]

    def _execute_tool(self, tool_call):
        # 根据 tool_call.function.name 查找对应工具并执行
        for tool in self.tools:
            if tool.name == tool_call.function.name:
                return tool.call(**tool_call.function.arguments)
        raise ValueError(f"Unknown tool: {tool_call.function.name}")

    def reset_session(self):
        """清空当前会话状态"""
        self.session_state = {
            "history": [],
            "pending_tasks": [],
            "last_tool_result": None
        }

实操心得:不要在 __init__() 里做耗时操作(如加载大模型、连接数据库)。Modal sandbox 启动时会先实例化 class,再调用 function。如果 __init__() requests.get("https://api.xxx.com/init") 超时,整个 sandbox 启动失败,且错误日志只会显示 Sandbox failed to start ,非常难定位。正确做法是把初始化延迟到 process_input() get_response() 的首次调用时,并加 if not hasattr(self, '_initialized'): 判断。

3.3 工具(Tool)编写规范:安全、可观测、可重试

Agents SDK 的 Tool 是 Agent 的“手脚”,其质量直接决定 Agent 的鲁棒性。在 Modal sandbox 中,工具必须遵循三条铁律:

  1. 所有网络请求必须带超时和重试 :sandbox 虽然稳定,但外部 API(如天气、日历)可能抖动。 requests.get(url, timeout=10) 是底线,推荐用 tenacity 库:
# tools/weather.py
from tenacity import retry, stop_after_attempt, wait_exponential
import requests

class WeatherTool:
    name = "get_weather"
    
    def get_schema(self):
        return {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Get current weather for a city",
                "parameters": {
                    "type": "object",
                    "properties": {"city": {"type": "string"}},
                    "required": ["city"]
                }
            }
        }

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
    def call(self, city: str) -> str:
        resp = requests.get(
            f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={os.getenv('OPENWEATHER_API_KEY')}",
            timeout=10
        )
        resp.raise_for_status()
        data = resp.json()
        return f"Weather in {city}: {data['weather'][0]['description']}, {data['main']['temp']}K"
  1. 敏感信息必须通过 Modal Secret 注入 :绝对禁止在代码里硬编码 API Key。Modal 提供 Secret 抽象,可在部署时注入环境变量:
modal secret create openweather-api-key --env OPENWEATHER_API_KEY=your_key_here

然后在 modal_app.py 中引用:

@app.function(
    image=image,
    secrets=[Secret.from_name("openweather-api-key")],  # 关键!注入 secret
    volumes={"/data": vol},
    timeout=3600,
)
def run_agent(user_input: str):
    ...
  1. 工具输出必须结构化、可解析 :Agent 的下一步决策依赖工具返回内容。避免返回 "Success" 这种模糊字符串,而应返回 JSON-like 字符串,包含明确字段:
# 错误示范
return "Created event on 2024-05-20"

# 正确示范
return json.dumps({
    "status": "success",
    "event_id": "evt_abc123",
    "start_time": "2024-05-20T14:00:00Z",
    "calendar_url": "https://calendar.google.com/event/abc123"
})

这样,当 Agent 下次调用邮件工具时,可以直接 json.loads(tool_result).get("calendar_url") 提取链接,无需正则匹配。

4. 实操过程与核心环节实现:从本地测试到线上部署全流程

4.1 本地开发与调试:用 Modal Sandbox 模拟真实环境

Modal 最大的优势是“本地即线上”。你不需要在本地搭一整套 K8s 才能测试 sandbox 行为。Modal CLI 提供 modal shell modal run 两条路径:

路径一: modal run 快速验证单次执行

# 在 agent_project/ 目录下
modal run modal_app.py::run_agent --user_input "Prepare meeting for team sync tomorrow at 10am in Beijing"

这条命令会:

  • 自动构建镜像(如果本地无缓存);
  • 启动一个 sandbox 实例;
  • 在 sandbox 内执行 run_agent() 函数;
  • 将 stdout/stderr 实时打印到你的终端;
  • 函数返回后,sandbox 自动销毁。

这是验证 Agent 逻辑是否正确的最快方式。我习惯先写一个 test_simple.py ,用固定输入反复跑,观察 print() 日志是否符合预期。例如,看到 Calling get_weather with {'city': 'Beijing'} 和紧接着的 Got weather: Clear sky, 298K ,就说明工具链路通了。

路径二: modal shell 进入交互式调试

modal run 报错,或者你想 inspect 运行时状态时, modal shell 是神器:

# 启动一个长期运行的 sandbox(不执行函数,只启动环境)
modal sandbox create --image-name my-agent-image --timeout 3600

# 获取 sandbox ID(如 sbx_abc123)
modal sandbox list

# 进入 shell
modal sandbox exec sbx_abc123 -- bash

进入后,你拥有一个完整的 Debian 环境:

  • ls /root/ 查看代码是否正确复制;
  • cat /root/requirements.txt 确认依赖版本;
  • python -c "import openai; print(openai.__version__)" 验证 SDK;
  • cd /root && python agent_core.py 手动运行 Agent 类(需稍作修改,加 if __name__ == "__main__": );
  • df -h 查看 /data 卷挂载是否成功;
  • ps aux | grep python 查看进程是否存在。

提示:Modal sandbox 默认不开启 SSH, modal shell 是唯一的交互入口。如果你需要 vim 编辑文件,记得在 Image 构建时加上 .apt_install("vim")

4.2 关键参数配置详解:超时、资源、日志的精确控制

Modal sandbox 的行为由 @function 装饰器的参数精确控制。以下是生产环境必须关注的 5 个参数及其取值逻辑:

参数 类型 推荐值 为什么这么设 实测影响
timeout int (秒) 3600 (1小时) Agent 处理复杂流程(如下载大附件+OCR+总结)可能耗时较长。设太短会导致 sandbox 强制终止,状态丢失。 设为 600 (10分钟)时,一个处理 12MB PDF 的 OCR Agent 在第 8 分钟被 kill,日志只显示 Sandbox timed out ,无任何中间状态。
cpu float 2.0 Modal 的 CPU 单位是 vCPU 的 1/1000。 2.0 ≈ 2 vCPU,足够支撑 GPT-4 Turbo 的 token 流式解析 + 多个并发工具调用。 cpu=0.5 时, openai.ChatCompletion.create() 响应延迟从 1.2s 升至 4.7s,因为模型推理线程被严重抢占。
memory int (MB) 4096 工具可能加载大文件(如 pandas.read_csv() 读取 500MB CSV),需要足够内存。Modal 内存与 CPU 独立配置,可灵活调整。 memory=1024 时, openpyxl.load_workbook() 加载 100MB Excel 直接 OOM,sandbox 退出码 137 (OOM Killer 触发)。
gpu str "any" "a10g" 如果 Agent 需要本地小模型(如 llama.cpp 做 RAG embedding),必须显式声明 GPU。 "any" 表示可用即可,Modal 自动分配; "a10g" 指定型号,确保性能一致。 未声明 gpu 时, torch.cuda.is_available() 返回 False ,所有 CUDA 代码跳过,降级为 CPU 模式,速度慢 15 倍。
concurrency_limit int 1 Sandbox 默认并发为 1,即一个 sandbox 同时只处理一个请求。这是 Agent 的最佳实践——避免 self.session_state 被多个线程污染。 设为 5 时,两个并发请求同时修改 self.session_state["history"] ,导致 IndexError: list index out of range ,因为 history 被交叉写入。

这些参数不是拍脑袋定的。我的做法是:先用 timeout=3600, cpu=2.0, memory=4096, gpu="any" 作为 baseline,然后用 modal run 跑 10 次典型用例,记录平均耗时、内存峰值( modal sandbox logs -f 中的 memory_usage_mb 字段)、CPU 利用率( modal sandbox exec sbx_xxx -- top -b -n1 \| grep python ),再根据数据微调。例如,如果 10 次运行中内存峰值最高 3200MB,那 memory=4096 就很稳妥;如果平均耗时 280s,那 timeout=3600 就绰绰有余。

4.3 部署与上线:从 modal run modal deploy

modal run 是开发态, modal deploy 是生产态。二者区别在于: run 启动临时 sandbox, deploy 创建一个长期存在的、可通过 HTTP/WebSocket 调用的 endpoint。

要让 Agent 可被外部系统(如 Slack App、Webhook)调用,需添加一个 Web Endpoint:

# modal_app.py 续写
from fastapi import FastAPI
from modal import asgi_app

web_app = FastAPI()

@web_app.post("/agent")
async def handle_agent_request(payload: dict):
    user_input = payload.get("input", "")
    if not user_input:
        return {"error": "Missing input"}
    
    # 调用 sandbox 函数
    result = await run_agent.remote.aio(user_input)  # 注意:remote.aio 是异步调用
    return {"response": result}

@app.function(
    image=image,
    volumes={"/data": vol},
    timeout=3600,
)
@asgi_app()
def fastapi_app():
    return web_app

部署命令极其简单:

modal deploy modal_app.py

执行后,Modal 返回一个 URL,如 https://your-app-name.modal.run/agent 。之后,任何系统都可以 POST JSON 到这个地址:

curl -X POST https://your-app-name.modal.run/agent \
  -H "Content-Type: application/json" \
  -d '{"input": "Summarize the Q2 sales report"}'

Modal 会自动:

  • 为每个请求启动一个新的 sandbox(或复用空闲 sandbox);
  • 将请求 body 传入 run_agent()
  • 将函数返回值作为 HTTP 响应体;
  • 记录所有请求的 latency、status code、error trace。

注意事项: @asgi_app() 装饰的函数必须返回一个 ASGI app(如 FastAPI 实例),且 Modal 会自动处理 CORS、gzip 压缩、HTTPS 终止。你不需要写任何网络层代码。但必须确保 fastapi_app() 函数内不包含耗时阻塞操作(如 time.sleep(10) ),否则会阻塞整个 endpoint。所有耗时逻辑必须放在 run_agent() 里,用 await run_agent.remote.aio() 异步调用。

5. 常见问题与排查技巧实录:我在真实项目中踩过的 7 个坑

5.1 问题速查表:高频报错与根因分析

错误现象 Modal 日志关键词 根本原因 解决方案
Sandbox failed to start Failed to start sandbox 镜像构建失败(如 pip install 报错)、挂载目录不存在、Secret 名称错误 1. modal run 时加 --show-progress 看构建日志;2. 在 Image 中加 .run_commands("ls -la /data") 确认挂载点;3. modal secret list 核对 secret 名称
ImportError: No module named 'tools' ModuleNotFoundError copy_local_directory("tools", ...) 路径错误,或 tools/__init__.py 缺失 1. 确保 tools/ 目录下有 __init__.py (哪怕为空);2. modal run ls tools/ 确认文件存在;3. 在 sandbox 内 ls /root/tools 验证复制结果
Sandbox timed out Sandbox timed out after XXX seconds timeout 参数小于 Agent 实际运行时间,或工具调用卡死(如未设 requests.timeout 1. 将 timeout 设为预估最大耗时的 2 倍;2. 所有 requests 调用必须带 timeout= ;3. 在 get_response() 循环中加 if time.time() - start_time > 0.8 * timeout: raise TimeoutError()
ConnectionResetError ConnectionResetError: [Errno 104] Connection reset by peer 外部 API(如 OpenAI)主动断连,通常因请求头缺失或频率超限 1. 在 OpenAI() 初始化时加 default_headers={"X-Modal-Source": "agent"} ;2. 用 tenacity 重试;3. Modal 控制台查看 RateLimitError 是否频发,考虑降级到 gpt-3.5-turbo
OSError: [Errno 28] No space left on device No space left on device /tmp 目录写满(Modal sandbox 的 /tmp 默认 512MB),常见于处理大文件 1. 所有大文件操作改用 /data 卷( /data/tmp/file.pdf );2. 在 Image 中加 .run_commands("rm -rf /tmp/*") 清理;3. modal volume ls 查看 /data 使用量
ValueError: Invalid tool call Invalid tool call Tool.call() 返回非字符串,或包含非法 JSON 字符(如未转义的双引号) 1. Tool.call() 必须 return str(result) ;2. 若返回 JSON,用 json.dumps(result, ensure_ascii=False) ;3. 在 call() 结尾加 print(f"[TOOL] Returning: {repr(return_value)}") 日志检查
Sandbox exited with code 137 Exit code 137 OOM(Out of Memory),系统强制 kill 进程 1. modal sandbox logs -f 中搜索 memory_usage_mb ,确认峰值;2. 降低 memory 参数前先优化代码(如 pandas.read_csv(chunksize=1000) 流式读取);3. 用 psutil.virtual_memory() 在 sandbox 内实时监控

5.2 独家避坑技巧:提升稳定性的 3 个硬核实践

技巧一:用 modal.Volume 做 Agent 的“硬盘”,而非“U盘”

很多教程教大家用 Volume 存配置文件,这远远不够。我把 /data 卷当作 Agent 的“本地硬盘”:

  • /data/logs/ :存放结构化日志(JSON Lines 格式),每天一个文件,方便后续用 modal volume get 导出分析;
  • /data/cache/ :存放工具返回的缓存(如天气数据缓存 1 小时,用 cache_key = f"weather_{city}_{int(time.time()//3600)}" );
  • /data/sessions/ :存放活跃会话的序列化状态( pickle.dump(session_state, open(f"/data/sessions/{session_id}.pkl", "wb")) ),即使 sandbox 重启,也能恢复上下文。

这样做的好处是:Agent 的“记忆”不再依赖进程内存,而是落盘持久化。我有一个客户,Agent 需要连续 3 天跟踪一个采购审批流程,中间 sandbox 因 Modal 维护重启了 2 次,但 /data/sessions/ 里的状态完好,用户完全无感知。

技巧二:在 get_response() 循环中加入“心跳检测”

Agents SDK 的 loop 是纯 Python,一旦卡在某个 tool.call() 里(如第三方 API 永远不响应),整个 sandbox 就假死。我在循环里加了心跳:

def get_response(self, messages: List[Dict[str, Any]]) -> str:
    start_time = time.time()
    heartbeat_interval = 30  # 每30秒打一次心跳
    last_heartbeat = start_time

    while True:
        # ... LLM 调用逻辑 ...

        if time.time() - last_heartbeat > heartbeat_interval:
            print(f"[HEARTBEAT] Elapsed: {int(time.time() - start_time)}s, history_len: {len(messages)}")
            last_heartbeat = time.time()

        # ... tool 调用逻辑 ...
        
        # 检查总耗时,防死循环
        if time.time() - start_time > self.timeout * 0.9:
            raise RuntimeError(f"Agent exceeded 90% of timeout ({self.timeout}s)")

这个心跳日志在 Modal 控制台清晰可见,一旦看到 HEARTBEAT 停止刷新,就知道卡在了哪一步,无需猜。

技巧三:用 modal.Secret 管理“动态凭证”,而非静态 Key

有些工具(如企业微信、飞书)的 Access Token 有 2 小时有效期,不能硬编码。我创建了一个 token-manager service,专门负责刷新和分发 token:

# token_manager.py
from modal import App, Function
import requests
import time

app = App("token-manager")

@app.function(
    secrets=[Secret.from_name("feishu-credentials")],
    schedule=Periodic(hours=1)  # 每小时刷新一次
)
def refresh_feishu_token():
    # 调用飞书 API 刷新 token,存入 Volume
    vol = Volume.from_name("agent-data")
    with vol.open("feishu_token.json", "w") as f:
        f.write(json.dumps({"access_token": new_token, "expires_at": time.time() + 7200}))

然后在 WeatherTool.call() 里:

def call(self, city: str) -> str:
    # 从 Volume 读取最新 token
    vol = Volume.from_name("agent-data")
    with vol.open("feishu_token.json", "r") as f:
        token_data = json.load(f)
    if time.time() > token_data["expires_at"]:
        # 触发刷新(异步,不阻塞当前调用)
        refresh_feishu_token.spawn()
        raise RuntimeError("Token expired, refreshing...")
    # 使用 token 调用 API
    ...

这样,凭证管理完全

更多推荐