Fork me on GitHub

分类 Daytona 下的文章

基于 Daytona 实现代码解释器

在入门篇里,我们提到 Daytona 作为一个 AI 优先基础设施,特别适合运行由 AI 生成的代码,比如下面这些场景:

  • 代码解释器(Code Interpreter)
  • 代码智能体(Coding Agents)
  • 数据分析(Data Analysis)
  • 数据可视化(Data Visualisation)
  • AI 评估(AI Evaluations)

到目前为止,我们已经学习了 Daytona 的大多数功能,包括代码执行、沙箱管理和镜像管理等,这些知识已经足够我们应付 80% 的应用场景。在今天这篇文章中,我将带大家从零开始基于 Daytona 实现一个代码解释器。

什么是代码解释器?

代码解释器(Code Interpreter)是 OpenAI 2023 年 7 月推出的一项功能,作为 ChatGPT 的插件,它允许用户通过自然语言指令生成并执行代码,支持数据分析、文件处理、图表生成等任务:

chatgpt-code-interpreter.png

代码解释器功能推出后颇受用户欢迎,在复杂任务中它可以显著提升效率,例如从 PDF 中提取文本、分析 CSV 数据或转换文件格式等。

实现 Function Calling

代码解释器的实现依赖于 OpenAI 更早时候推出的 Function Calling 功能,它为 Chat Completions API 添加了函数调用的能力,帮助开发者通过 API 的方式实现类似于 ChatGPT 插件的数据交互能力。

简单来说,Function Calling 就是在调用大模型时提供几个工具选项,大模型判断用户的问题是否需要调用某个工具。如果不需要则直接回复,这和传统的调用没有区别;如果需要调用则返回合适的工具和对应的参数给用户,用户拿到后调用对应的工具,再将调用结果送给大模型,最后,大模型根据工具的调用结果来回答用户的问题。

根据需要,工具的调用可能有多轮,我们可以将 Function Calling 实现成下面这样一个循环:

def chat_completion_with_function_calling(query):
  # 原始用户问题
  messages=[
    {'role': 'user', 'content': query},
  ]
  while True:
    # 调用大模型
    completion = client.chat.completions.create(
      model="gpt-4o",
      messages=messages,
      tools=tools,
    )

    # 不需要工具直接返回
    message = completion.choices[0].message
    if not message.tool_calls:
      break

    # 工具调用消息
    tool_calls = message.tool_calls
    messages.append(
      {'role': 'assistant', 'tool_calls': tool_calls, 'content': ''},
    )
    
    for tool_call in tool_calls:
      # 调用工具
      function = globals().get(tool_call.function.name)
      args = json.loads(tool_call.function.arguments)
      result = function(args)

      # 工具调用结果消息
      messages.append(
        {'role': 'tool', 'content': result, 'tool_call_id': tool_call.id},
      )
  return completion

其中 tools 字段是我们提供的工具列表,比如这样:

tools = [
  {
    "type": "function",
    "function": {
      "name": "get_current_date",
      "description": "获取今天的日期信息,包括几月几号和星期几",
      "parameters": {
        "type": "object",
        "properties": {}
      }
    }
  },
  {
    "type": "function",
    "function": {
      "name": "get_weather_info",
      "description": "获取某个城市某一天的天气信息",
      "parameters": {
        "type": "object",
        "properties": {
          "city": {
            "type": "string",
            "description": "城市名",
          },
          "date": {
            "type": "string",
            "description": "日期,格式为 yyyy-MM-dd",
          },
        },
        "required": ["city", "date"],
      }
    }
  }
]

上面我们定义了两个工具 get_current_dateget_weather_info,分别用于获取日期和查询天气。这两个工具都需要我们自己实现,大模型只负责告诉我们要不要调用工具,它自己不会主动调用工具,所以调用的逻辑还是需要我们来做,比如上面代码中的这段:

function = globals().get(tool_call.function.name)
args = json.loads(tool_call.function.arguments)
result = function(args)

这里从大模型返回的 tool_call.function 字段中提取出调用的工具名称和参数,然后调用对应的工具函数:

def get_current_date(args):
  import datetime
  today = datetime.date.today()
  weekday = today.weekday()
  weeekdays = ['一','二','三','四','五','六','日']
  return '今天是' + str(today) + ', ' + '星期' + weeekdays[weekday]

def get_weather_info(args):
  import requests
  import json
  day = args['date'].split('-')[2]
  url = 'https://query.asilu.com/weather/baidu/?city=' + args['city']
  content = requests.get(url).content
  response = json.loads(content.decode())
  for weather in response['weather']:
    if day == weather['date'].split('日')[0]:
      return weather['weather'] + "," + weather['temp'] + "," + weather['wind']
  return '未查询到天气'

Function Calling 执行流程

接下来,我们问大模型一个问题,看看整体的执行流程是什么样的:

completion = chat_completion_with_function_calling('明天合肥的天气怎么样?')
print(completion.choices[0].message.content)

首先,我们调用大模型的入参是这样:

[
  {
    "role": "user",
    "content": "明天合肥的天气怎么样?"
  }
]

大模型返回的结果是这样:

{
  "id": "chatcmpl-BWFUtwaRXeOo3v7T9dynk8DI1bevC",
  "choices": [
    {
      "message": {
        "content": null,
        "role": "assistant",
        "tool_calls": [
          {
            "id": "call_Omy5d3I2D3lOccmy1V8DjFiL",
            "function": {
              "arguments": "{}",
              "name": "get_current_date"
            },
            "type": "function"
          }
        ]
      }
    }
  ]
}

可以看到大模型希望先知道今天的日期,于是我们调用 get_current_date 函数得到今天的日期,再次调用大模型:

[
  {
    "role": "user",
    "content": "明天合肥的天气怎么样?"
  },
  {
    "role": "assistant",
    "tool_calls": [
      {
        "id": "call_Omy5d3I2D3lOccmy1V8DjFiL",
        "function": {
          "arguments": "{}",
          "name": "get_current_date"
        },
        "type": "function"
      }
    ],
    "content": ""
  },
  {
    "role": "tool",
    "content": "今天是2025-05-12, 星期一",
    "tool_call_id": "call_Omy5d3I2D3lOccmy1V8DjFiL"
  }
]

大模型返回结果如下:

{
  "id": "chatcmpl-BWFUv9hHWuFDNUJ4OrlkaoajT3VKT",
  "choices": [
    {
      "message": {
        "content": null,
        "role": "assistant",
        "tool_calls": [
          {
            "id": "call_dnxfTpHKBUwAC1dwdR45pYqx",
            "function": {
              "arguments": "{\"city\":\"合肥\",\"date\":\"2025-05-13\"}",
              "name": "get_weather_info"
            },
            "type": "function"
          }
        ]
      }
    }
  ]
}

可以看到大模型已经知道了今天的日期是 2025-05-12,并推出明天的日期是 2025-05-13,接下来需要调用天气查询了,于是我们拿着大模型给的工具名称和参数调用 get_weather_info 函数,将调用结果再次丢给大模型:

[
  {
    "role": "user",
    "content": "明天合肥的天气怎么样?"
  },
  {
    "role": "assistant",
    "tool_calls": [
      {
        "id": "call_Omy5d3I2D3lOccmy1V8DjFiL",
        "function": {
          "arguments": "{}",
          "name": "get_current_date"
        },
        "type": "function"
      }
    ],
    "content": ""
  },
  {
    "role": "tool",
    "content": "今天是2025-05-12, 星期一",
    "tool_call_id": "call_Omy5d3I2D3lOccmy1V8DjFiL"
  },
  {
    "role": "assistant",
    "tool_calls": [
      {
        "id": "call_dnxfTpHKBUwAC1dwdR45pYqx",
        "function": {
          "arguments": "{\"city\":\"合肥\",\"date\":\"2025-05-13\"}",
          "name": "get_weather_info"
        },
        "type": "function"
      }
    ],
    "content": ""
  },
  {
    "role": "tool",
    "content": "晴转多云,28/19℃,南风",
    "tool_call_id": "call_dnxfTpHKBUwAC1dwdR45pYqx"
  }
]

至此,大模型终于可以回答我们的问题了:

{
  "id": "chatcmpl-BWFUyjlIADisfQnvm8e0E7td2m6Rj",
  "choices": [
    {
      "message": {
        "content": "明天合肥的天气是晴转多云,气温为28到19摄氏度,南风。",
        "role": "assistant"
      }
    }
  ]
}

实现代码解释器

现在我们已经知道了 Function Calling 的执行流程,要实现代码解释器就很容易了,我们只需要将上面的工具列表替换成 code_interpreter

tools = [
  {
    "type": "function",
    "function": {
      "name": "code_interpreter",
      "description": "生成 Pytohn 代码并执行,用于解决用户提出的数学或代码类问题",
      "parameters": {
        "type": "object",
        "properties": {
          "query": {
            "type": "string",
            "description": "用户问题",
          }
        },
        "required": ["query"],
      }
    }
  }
]

这个工具可以生成 Pytohn 代码并执行,专门用于解决用户提出的数学或代码类问题。工具的实现如下:

def code_interpreter(args):
  result = generate_code(args['query'])
  print("生成代码:", result)
  
  import re
  code_match = re.search(r"```python\n(.*?)```", result, re.DOTALL)
  code = code_match.group(1) if code_match else result
  code = code.replace('\\', '\\\\')
  
  result = execute_code(code)
  print("代码执行结果:", result)
  return result

主要分为两个部分,第一部分为生成代码,我们直接调大模型接口来生成 Python 代码:

def generate_code(query):
  completion = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
      {
        "role": "system", 
        "content": "你是一个 Python 编码助手,请生成一段 Python 代码解决用户的问题,直接输出代码,不用解释。"
      },
      {
        "role": "user",
        "content": query,
      },
    ],
  )
  return completion.choices[0].message.content

第二部分为执行代码,注意大模型返回的内容可能包含一些 Markdown 的代码语法,需要通过正则处理下,然后通过 Daytona SDK 创建沙箱,并在沙箱中安全地执行代码:

def execute_code(code):
  params = CreateSandboxParams(
    language="python"
  )
  sandbox = daytona.create(params)
  response = sandbox.process.code_run(code)
  if response.exit_code != 0:
    return f"Error running code: {response.exit_code} {response.result}"
  else:
    return response.result

接下来,我们问大模型一个数学问题:

completion = chat_completion_with_function_calling('斐波拉契数列的第20项是多少?')
print(completion.choices[0].message.content)

执行结果如下:

daytona-code-interpreter.png

可以看到大模型正确生成了 Python 代码,通过 Daytona 的沙箱功能,我们可以放心地执行由大模型生成的代码,然后我们将执行结果丢给大模型,最终回答了用户问题。

至此,我们成功实现了一个简单的代码解释器。

小结

在本文中,我们详细讲解了如何基于 Daytona 实现一个简单的代码解释器。通过结合 OpenAI 的 Function Calling 功能和 Daytona 的沙箱管理,我们成功构建了一个能够生成并运行 Python 代码的系统。关于这个代码解释器,还有很多可以优化的地方,比如支持文件处理,生成分析图表,等等,有机会后面再实践,感兴趣的同学也可以自己尝试下。


详解 Daytona 的镜像管理

在上一篇文章中,我们学习了 Daytona 的沙箱管理功能,包括沙箱的创建、资源限制、自动停止机制以及沙箱状态的管理。今天我们将继续学习 Daytona 的镜像管理功能。可以把镜像比喻成一个模板,它包含了沙箱中所需的所有依赖项、工具和环境设置。Daytona 支持所有标准的 Docker 或 OCI 兼容镜像。

我们在之前的代码中创建沙箱时使用了一个自定义镜像 debian:12.9,细心的同学可能已经发现了,如果你直接运行那段代码是会报错的:

daytona_sdk.common.errors.DaytonaError: 
    Failed to create sandbox: Image debian:12.9 not found or not accessible

这是因为在 Daytona 中使用自定义镜像,我们必须得提前创建它。

默认镜像

在创建镜像之前,我们先来看看 Daytona 的默认镜像长啥样,前面提到过,当使用默认镜像时,Daytona 会自动从池中获取,所以创建速度极快,如果用自定义镜像,就没有这个优势了,所以我们应该当且仅当默认镜像满足不了我们的需求时,才使用自定义镜像。

Daytona 默认镜像中预装了一些有用的工具,例如 python、node、pip 以及一些常见的 pip 包,如下:

daytona-default-image.png

镜像的创建

当我们确实需要使用自定义镜像时,可以进入 Dashboard -> Images 页面,点击 Create Image 按钮创建新镜像:

daytona-create-image.png

这里可以填写 Docker Hub 上的任何公共镜像,注意名称和标签都必须填写,比如我们这里使用的 debian:12.9,而且建议不要写 latest 或 stable 这样的标签。也可以填写来自其他公共仓库上的镜像,只要在前面加上仓库地址即可,比如来自 Github 的镜像仓库的 ghcr.io/graalvm/jdk-community:21 镜像。

另外,为了确保你的镜像运行后不会立即退出,一定要有一个长时间运行的入口点(entrypoint),如果不确定,可以直接填上 sleep infinity 命令。

镜像一旦被创建,就会被自动拉取,然后自动验证,验证通过后就会进入 Active 状态,此时我们才可以在程序中使用它:

daytona-images.png

私有仓库镜像

Daytona 也支持使用来自私有仓库的镜像,我们需要先在 Dashboard -> Registries 页面配置私有仓库:

daytona-add-registry.png

在这里输入仓库名称、URL、用户名、密码以及项目名称即可,确保 Daytona 能访问你的私有仓库。然后就可以创建私有仓库的镜像了,注意镜像名称包括私有仓库的 URL 和项目名称,比如 my-private-registry.com/<my-project>/alpine:3.21

本地镜像

除了公共镜像和私有仓库镜像,还有一种情况是,我们在本地开发和构建的镜像。我们可能没有自己的私有镜像仓库,也不想推送到公共镜像仓库中,这时,就可以使用 Daytona 提供的命令行工具将本地镜像推送到 Daytona 来使用。

首先,安装 Daytona CLI:

$ brew install daytonaio/cli/daytona

然后,通过 daytona login 登录:

$ daytona login

Select Authentication Method

  › Login with Browser
    Authenticate using OAuth in your browser

    Set Daytona API Key
    Authenticate using Daytona API key

Daytona 会显示两种登录方式:浏览器登录,或输入 API KEY。这里我们选择浏览器,登录成功:

 Opening the browser for authentication ...
                                           
 If opening fails, visit:
                         
https://daytonaio.us.auth0.com/authorize?audience=...
                        
 Successfully logged in!

接着就可以通过 daytona image push 推送本地镜像了:

$ daytona image push test-image:1.0
The push refers to repository [harbor-transient.internal.daytona.app/daytona/test-image]
08000c18d16d: Pushed 
1.0: digest: sha256:ec1b05d1eac264d9204a57f4ad9d4dc35e9e756e9fedaea0674aefc7edb1d6a4 size: 527
                                              
 Successfully pushed test-image:1.0 to Daytona
                                              
 ✓  Use 'harbor-transient.internal.daytona.app/daytona/test-image:1.0' to create a new sandbox using this image

这个命令将本地的 test-image:1.0 镜像成功推送到 Daytona 的镜像仓库中了,注意在使用这个镜像时,需要指定全路径。除了推送本地已有镜像,Daytona 也支持通过 Dockerfile 自动构建并推送镜像:

$ daytona image push test-image:1.0 --dockerfile ./Dockerfile

最后还有一点要注意的是,Daytona 目前仅支持 AMD 架构的镜像,还不支持 ARM 架构的镜像,否则会报错:

$ daytona image push test-image:1.0
FATA[0000] image 'test-image:1.0' is not compatible with AMD architecture

镜像管理

上面提到,我们可以在 Dashboard -> Images 页面对镜像进行管理,比如创建、禁用、删除等。我们也可以通过 daytona image 命令对镜像进行管理,比如:

  • daytona image create - 创建镜像
  • daytona image push - 推送本地镜像
  • daytona image delete - 删除镜像
  • daytona image list - 显示镜像列表

下面通过 daytona image list 命令列出所有镜像:

daytona-image-list.png

小结

好了,今天的内容就这么多。在本文中,我们详细探索了 Daytona 的镜像管理功能,包括镜像的创建、使用及管理。我们学习了默认镜像、创建自定义镜像以及如何使用私有仓库镜像和本地镜像,相信大家对 Daytona 的镜像功能有了更深入的了解。


详解 Daytona 的沙箱管理

在上一篇文章中,我们学习了 Daytona 的代码执行功能,特别是如何在沙箱中执行 Python 代码、Shell 脚本和长耗时任务。关于代码执行,还有一个小尾巴没有讲完,那就是代码执行的环境。我们知道,执行代码之前首先要创建沙箱,代码是在沙箱中被安全地执行,那么这个沙箱环境里都有什么东西呢?我们到底能在沙箱中执行哪些代码呢?我们今天就来看看这些问题。

沙箱的创建

回顾下前面的内容,创建沙箱的代码如下:

sandbox = daytona.create(CreateSandboxParams(language="python"))

我们是通过 Daytona 的 create() 方法创建沙箱的,其中 CreateSandboxParams 用于配置沙箱环境,该类的定义如下:

daytona-create-sandox-params.png

可以看到,除了编程语言还有不少的配置选项,包括镜像、环境变量、标签、资源限制、自动停止时间等等,下面是一个较复杂的例子:

params = CreateSandboxParams(
    language="python",
    image="debian:12.9",
    env_vars={"DEBUG": "true"},
    labels={"ENV": "dev"},
    resources=SandboxResources(cpu=2, memory=4),
    auto_stop_interval=0
)

在这个例子里,我们指定了基础镜像为 debian:12.9,学过 Docker 的同学应该很快就能理解这里的沙箱其实就是容器,而容器都是由镜像创建而来,我们指定不同的镜像,也就是为沙箱配置不同的环境,关于镜像的使用,我们后面再看,这里先看另几个配置。

值得注意的是,Daytona 针对默认镜像提前创建了一些沙箱放在池中,当用户使用默认镜像创建沙箱时,就会直接从池中取出一个可用的沙箱,从而将创建时间缩短到毫秒。所以,当使用自定义镜像时,你会发现创建沙箱的速度会变慢,因为自定义的镜像没有提前池化。

对沙箱进行资源限制

我们在用 Docker 启动一个容器的时候,可以通过 --memory--cpus 等参数限制容器能使用的系统资源:

$ docker run --memory=512m --cpus=1 ubuntu

Daytona 也提供了类似的机制,通过 SandboxResources 可以控制沙箱能使用的资源上限,该类定义如下:

daytona-sandbox-resources.png

下面是一个示例,限制 2 个 CPU 核心,1 个 GPU,4G 内存 和 20G 磁盘:

resources = SandboxResources(
    cpu=2,
    gpu=1,
    memory=4,
    disk=20
)

沙箱的自动停止

在前面的学习中,我们每次使用完沙箱后都是通过 daytona.remove(sandbox) 将沙箱删除掉。如果你不手工删除,Daytona 为了避免资源浪费,会自动检测沙箱的活动状态,对于 15 分钟内不活跃的沙箱,将自动停止。

这个 15 分钟就是默认的自动停止时间,我们可以在创建沙箱时通过 auto_stop_interval 参数修改这个值,如果我们不希望沙箱被自动停止,可以将其设置为 0:

sandbox = daytona.create(CreateSandboxParams(
    language = "python",
    auto_stop_interval = 0
))

操作沙箱

Daytona 对象提供了一系列和沙箱有关的操作,包括:

  • create() - 创建沙箱
  • delete() - 删除沙箱
  • get_current_sandbox() - 根据 ID 获取沙箱
  • find_one() - 根据 ID 或标签获取沙箱
  • list() - 列出所有沙箱
  • start() - 启动沙箱
  • stop() - 停止沙箱

这些操作都比较简单,和 Docker 操作容器也很类似,此处略过。

沙箱的状态

对沙箱的操作必然导致沙箱状态的变动,Daytona 的沙箱可以有三种状态:

  • 运行中:运行中的沙箱会占用 CPU、内存和磁盘存储,每种资源都按使用时间计费,所以当沙箱不再被主动使用时,建议将其停止。可以选择手动停止,也可以通过设置自动停止时间自动停止;
  • 已停止:已停止的沙箱不会占用 CPU 和内存,只占用磁盘存储,当你需要使用沙箱时可以再启动它;
  • 已归档:当沙箱被归档时,整个文件系统会迁移到更廉价的对象存储,这样可以更加节约成本,但是启动一个已归档的沙箱需要更多时间,你需要根据实际情况来权衡是停止还是归档你的沙箱。

小结

在本篇文章中,我们详细探讨了 Daytona 的沙箱管理,包括沙箱的创建、资源限制、自动停止机制以及沙箱状态的管理。了解沙箱的配置选项,不仅能提高代码的执行效率,还能有效控制成本,避免资源的浪费。自动停止功能进一步提升了资源管理的灵活性,使得开发者无需担心长时间未使用的沙箱占用系统资源。

沙箱和镜像是两个密不可分的概念,在后续的内容中,我们将进一步学习 Daytona 的镜像管理功能,希望大家继续关注!


详解 Daytona 的代码执行

在上一篇文章中,我们学习了 Daytona 的核心特性和基本用法,然后通过 Python SDK 创建沙箱并执行代码。回顾下之前学过的内容,可以将 Daytona 执行代码分成四步:

  1. 初始化 Daytona 客户端;
  2. 通过客户端创建沙箱,指定沙箱环境,Daytona 支持执行 Python 和 JS 代码;
  3. 在沙箱执行代码,返回执行结果;
  4. 销毁沙箱;

其中第三步执行代码是 Daytona 最核心的功能,我们今天继续深入学习它。

执行代码和运行命令

Daytona 的 Sandbox 类含有不少属性,如下:

daytona-sandbox.png

其中 process 属性是我们关注的焦点,通过它的 code_run() 方法可以执行代码:

response = sandbox.process.code_run('print("Hello")')

这个方法支持传入一些额外参数,比如环境变量或命令行参数等:

params = CodeRunParams(
    argv=[
        "-h", "-i"
    ], 
    env={
        'DEMO_ENV': 'xxx'
    }
)
response = sandbox.process.code_run('''
import os
import sys

print("环境变量:DEMO_ENV = ", os.getenv("DEMO_ENV"))
print("命令行参数:", *[f"{arg}" for _, arg in enumerate(sys.argv)])
''', params=params)

在代码中可以通过 os.getenv()sys.argv 来获取,调用结果如下:

环境变量:DEMO_ENV =  xxx
命令行参数: -c -h -i

这里的命令行参数比较有意思,可以看到我们只传入了 -h -i 两个参数,但是输出却多了一个 -c,这说明 Daytona 很有可能是通过 python -c 去执行我们的 Python 代码的:

python -c "print('Hello')"

process 属性还有一个 exec() 方法,可以运行 Shell 命令:

response = sandbox.process.exec('echo "Hello"')

这个方法同样支持传入一些额外参数,比如设置当前工作目录:

response = sandbox.process.exec('ls', cwd="/")

设置环境变量:

response = sandbox.process.exec('echo $DEMO_ENV', env={"DEMO_ENV": "xxx"})

运行长耗时任务

对于一些长耗时任务,为避免超时,可以在运行时加上 timeout 参数:

response = sandbox.process.exec("sleep 5", timeout=10)

除此之外,Daytona 还提供了一种推荐做法:通过后台会话运行长耗时任务

sandbox.process 属性提供了一系列和会话相关的方法:

  • create_session() - 创建会话
  • get_session() - 获取会话
  • list_sessions() 列出所有会话
  • delete_session() - 删除会话
  • execute_session_command() - 在会话中执行命令
  • get_session_command() - 获取会话中的命令
  • get_session_command_logs() - 获取会话中的命令日志
  • get_session_command_logs_async() - 异步获取会话中的命令日志

我们首先选择一个唯一标志作为会话 ID,然后通过它创建一个会话:

session_id = "my-session"
sandbox.process.create_session(session_id)

接着就可以使用 execute_session_command() 在会话中执行命令,要执行的命令需要放在一个 SessionExecuteRequest 对象中:

req = SessionExecuteRequest(
    run_async=True,
    command="for ((i=0; i<10; i++)); do echo $i; sleep 1; done"
)
response = sandbox.process.execute_session_command(session_id, req)

注意这里的 run_async=True 表示异步执行,调用完 execute_session_command() 方法后会立即返回,不会阻塞后面的代码。返回的 response 有一个 cmd_id 字段,通过 get_session_command() 方法可以检查命令的实时状态以及命令输出的日志。下面的代码示例中每隔 1 秒检查一次命令状态:

while(True):
    time.sleep(1)
    command = sandbox.process.get_session_command(session_id, response.cmd_id)
    if command.exit_code is None:
        print(f"Command is running ...")
        continue
    if command.exit_code == 0:
        print(f"Command {command.command} completed successfully")    
    else:
        print(f"Command {command.command} failed")
    logs = sandbox.process.get_session_command_logs(session_id, response.cmd_id)
    print(f"{logs}")
    break

当命令处于运行中状态时,exit_code 一直等于 None,当 exit_code 为 0 时表示命令运行成功,否则运行失败,再通过 get_session_command_logs() 方法获取命令的完整日志。

有时候我们不想等命令运行结束才打印日志,而是希望实时地输出日志,这对调试和监控都非常有意义。这时我们可以使用 get_session_command_logs_async() 方法:

await sandbox.process.get_session_command_logs_async(
    session_id,
    response.cmd_id,
    lambda chunk: print(f"Log chunk: {chunk}")
)

该方法接受一个 callback 函数,当有日志输出时就回调该函数。

小结

在本文中,我们深入探讨了 Daytona 的代码执行功能,特别是如何在沙箱中有效地执行代码和长耗时任务。我们学习了 Daytona 的 process 属性以及其便捷的 code_run()exec() 方法,这些方法支持环境变量、命令行参数等设置,让我们能够灵活地执行 Python 和 Shell 代码。

此外,对于长耗时任务,我们介绍了如何通过后台会话进行处理,实现非阻塞的命令执行。我们可以创建会话、执行异步命令,并实时获取日志输出,这为任务的监控和调试提供了极大的便利。通过 get_session_command_logs_async() 方法,我们甚至能够实现实时日志的流式处理,进一步增强了我们对代码执行过程的可控性。

以上就是今天的全部内容了,基本上涉及到了和 Daytona 代码执行相关的各个细节,不过还有一点非常重要,那就是代码执行的环境,我们下一篇继续学习。


Daytona 快速入门

在当今的技术浪潮中,人工智能的飞速发展正在重新定义我们的工作方式,尤其是在软件开发领域。随着大模型生成代码能力的不断提升,越来越多的开发者开始探索如何利用这些强大的工具提高工作效率。然而,伴随而来的是一个重要的问题:生成的代码的安全性和可靠性能否得到保障?我们今天要介绍一个新的开源项目 Daytona,就是为了解决这一挑战。

daytona-homepage.png

Daytona 是一个旨在提供安全且弹性的基础设施的 SaaS 服务,专门用于在隔离环境中运行 AI 生成的代码。这样的设计理念不仅能确保代码在执行过程中的安全性,还能为开发者们提供一个稳定和可靠的运行平台。自从 Daytona 在开源社区推出以来,它受到了广泛关注,Github Star 数量已接近 20k,更在 Product Hunt 榜单上夺得了 Top 1 的佳绩。这一切都表明,Daytona 已成为开发者们的新宠,正引领着未来代码生成与执行的潮流。接下来,就让我们快速学习下 Daytona 的核心特性和用法。

核心特性

Daytona 的核心特性如下:

  • 极速基础设施:快速创建沙箱,低于 90 毫秒;
  • 独立与隔离的运行时:以零风险执行 AI 生成的代码,保护你的基础设施;
  • 大规模并行化以支持并发 AI 工作流程:支持沙箱文件系统和内存状态的分叉;

daytona-features.png

此外,还有很多对开发人员友好的特性:

  • 编程控制:支持文件、Git、LSP 和 执行 API;
  • 无限持久化:支持将沙箱设置为永久存在;
  • 原生 SSH 访问:支持通过 SSH 或 IDE 访问沙箱;
  • OCI/Docker 兼容性:支持使用任何 OCI/Docker 镜像来创建沙箱;
  • 开源透明:代码完全开源,支持自托管;

Daytona 被称为 AI 优先基础设施(AI-First Infrastructure),针对大模型、智能体和 AI 评估专门优化,通过 Daytona 可以实现下面这些功能:

  • 代码解释器(Code Interpreter)
  • 代码智能体(Coding Agents)
  • 数据分析(Data Analysis)
  • 数据可视化(Data Visualisation)
  • AI 评估(AI Evaluations)

准备 API Key

为了体验 Daytona 的功能,我们需要做一些环境准备。首先登录 Daytona 官网,注册一个账号:https://app.daytona.io

daytona-signup.png

然后,进入 Dashboard -> Keys 页面,创建一个 API Key:

daytona-create-apikey.png

注意这个 API Key 只会出现一次,务必记下这个 API Key,后面会用到。

然后就可以使用 Daytona 的各种功能了,Daytona 为免费用户提供了 10 个沙箱、20 核 CPU、40G 内存、50G 磁盘,以及 5 个镜像、10G 镜像大小的限额:

daytona-usage.png

使用 Python SDK

接下来我们使用 Python SDK 简单体验下 Daytona 的基础用法。首先安装依赖:

$ pip install daytona-sdk

然后编写代码如下:

from daytona_sdk import Daytona, DaytonaConfig, CreateSandboxParams

# Initialize the Daytona client
daytona = Daytona(DaytonaConfig(api_key=DAYTONA_API_KEY))

# Create the Sandbox instance
sandbox = daytona.create(CreateSandboxParams(language="python"))

# Run code securely inside the Sandbox
response = sandbox.process.code_run('print("Sum of 3 and 4 is " + str(3 + 4))')
if response.exit_code != 0:
    print(f"Error running code: {response.exit_code} {response.result}")
else:
    print(response.result)

# Clean up the Sandbox
daytona.remove(sandbox)

整个代码结构非常清晰,就是四大步:

  1. 初始化 Daytona 客户端:传入刚刚在 Dashboard 页面生成的 API Key;
  2. 创建沙箱:指定沙箱环境为 python 语言,这里也可以传入 typescript,Daytona 也支持执行 JS 代码;
  3. 在沙箱执行代码:这里传入 Python 代码,代码在沙箱中安全地被执行,并返回执行结果;
  4. 销毁沙箱:如果沙箱不需要了,可以销毁掉;也可以不销毁,这时我们可以在 Dashboard -> Sandboxes 页面查看沙箱状态,甚至通过 SSH 进到沙箱环境中排查问题;

daytona-sandboxes.png

最后的运行结果如下:

Sum of 3 and 4 is 7

至此我们对 Daytona 已经有了一个基本了解,下一篇我们将学习更高级的用法。