Agent Runtime 进阶功能（归档）

本文档介绍 PPIO Agent Runtime 的高级功能和最佳实践。

配置文件详解

.ppio-agent.yaml 结构说明

.ppio-agent.yaml 配置文件采用 Kubernetes 风格的 YAML 格式：

apiVersion: v1
kind: Agent
metadata:
  name: my-agent              # Agent 名称（必须小写字母数字和连字符）
  version: 1.0.0              # Agent 版本（语义化版本）
  author: dev@example.com     # 作者邮箱（必填）
  description: My AI Agent    # Agent 描述（可选）
  created: '2025-10-23T10:30:00Z'  # 创建时间（自动生成）

spec:
  entrypoint: app.py          # Python 入口文件（必须是 .py 文件）
  
  # 环境变量配置（可选）
  envVars:
    MODEL_NAME: deepseek/deepseek-v3-0324
    TEMPERATURE: '0.7'
  
  # 运行时配置（可选，会体现在构建好的沙箱模板中）
  runtime:
    timeout: 300              # 启动超时秒数（1-3600，默认 300）
    memory_limit: 1Gi         # 内存限制（支持 "512Mi", "1Gi" 等格式）
    cpu_limit: '1'            # CPU 限制（支持 "1", "1000m" 等格式）

# 状态字段（由系统维护，用户不应手动修改）
status:
  phase: deployed             # 当前阶段: pending/building/deployed/failed
  agent_id: agent-xxxxx  # Agent ID（部署后自动生成）
  last_deployed: '2025-10-23T10:35:00Z'  # 最后部署时间
  build_id: build_xyz789      # 构建 ID（部署后自动生成）

修改配置

修改 CPU 和内存配置

在 .ppio-agent.yaml 的 spec.runtime 下修改资源配置：

spec:
  runtime:
    # CPU 配置
    cpu_limit: '2'        # 2 核 CPU
    # 内存配置
    memory_limit: 2Gi     # 2 GB 内存

修改环境变量

.ppio-agent.yaml 中的 spec.envVars 仅用于 CLI 的 agent invoke 命令，不会传递到部署后的沙箱模板中。在 .ppio-agent.yaml 的 spec.envVars 下修改环境变量：

spec:
  envVars:
    # LLM 配置
    MODEL_NAME: deepseek/deepseek-v3-0324
    TEMPERATURE: '0.7'

注意：

⚠️ 不要在 .ppio-agent.yaml 中存储敏感信息（如 API Key）
也可以在运行 agent invoke 命令时通过 --env 参数传递环境变量

重新部署使配置生效

修改 .ppio-agent.yaml 的资源规格配置后，需要重新部署：

# 重新部署（会创建新版本）
npx ppio-sandbox-cli agent launch

环境变量管理

向运行在沙箱实例中的 Agent 传递环境变量有以下几种方式：

方式 1: 在配置文件中定义（仅用于 CLI 调用）

在 .ppio-agent.yaml 的 spec.envVars 中定义环境变量：

spec:
  envVars:
    MODEL_NAME: deepseek/deepseek-v3-0324
    TEMPERATURE: '0.7'

方式 2: 使用 SDK 调用时动态传递

使用 SDK 的 invoke_agent_runtime 方法调用 Agent 时，通过 envVars 参数动态传递：

import os
from ppio_sandbox.agent_runtime import AgentRuntimeClient

client = AgentRuntimeClient(api_key=os.getenv("PPIO_API_KEY"))

response = await client.invoke_agent_runtime(
    agentId="agent-xxxxx",
    payload=payload,
    envVars={
        # 从环境变量读取敏感信息
        "PPIO_API_KEY": os.getenv("PPIO_API_KEY"),
        "DATABASE_PASSWORD": os.getenv("DATABASE_PASSWORD"),
        
        # 或直接传递
        "MODEL_NAME": "deepseek/deepseek-v3-0324",
        "TEMPERATURE": "0.7"
    }
)

流式响应

使用同步生成器实现流式响应

使用 Python 生成器（Generator）实现流式响应：

from ppio_sandbox.agent_runtime import AgentRuntimeApp

app = AgentRuntimeApp()

@app.entrypoint
def streaming_agent(request: dict):
    """同步流式响应"""
    prompt = request.get("prompt", "")

    # 使用生成器逐块返回
    for i, chunk in enumerate(generate_response(prompt)):
        yield {
            "chunk": chunk,
            "type": "content",
            "index": i
        }
    # 发送结束标记
    yield {"chunk": "", "type": "end"}

使用异步生成器实现流式响应

使用 Python 异步生成器（Async Generator）：

import asyncio

@app.entrypoint
async def async_streaming_agent(request: dict):
    """异步流式响应"""
    prompt = request.get("prompt", "")

    async for chunk in async_generate_response(prompt):
        yield {
            "chunk": chunk,
            "type": "content"
        }
    yield {"chunk": "", "type": "end"}

LangChain 流式响应示例

使用 LangChain 实现流式响应的完整示例：

import os
from langchain_openai import ChatOpenAI
from langchain.callbacks.base import BaseCallbackHandler
from ppio_sandbox.agent_runtime import AgentRuntimeApp

app = AgentRuntimeApp()

class StreamingHandler(BaseCallbackHandler):
    """流式回调处理器"""
    def __init__(self):
        self.tokens = []
    
    def on_llm_new_token(self, token: str, **kwargs):
        self.tokens.append(token)

@app.entrypoint
def langchain_streaming_agent(request: dict):
    """LangChain 流式响应"""
    prompt = request.get("prompt", "")

    # 创建支持流式的 LLM
    llm = ChatOpenAI(
        api_key=os.getenv("PPIO_API_KEY"),
        streaming=True
    )
    
    # 流式调用
    for chunk in llm.stream(prompt):
        if chunk.content:
            yield {
                "chunk": chunk.content,
                "type": "content"
            }
    yield {"chunk": "", "type": "end"}

调用流式 Agent

使用 SDK 调用流式 Agent：

import asyncio
import json
import os
from ppio_sandbox.agent_runtime import AgentRuntimeClient

async def call_streaming_agent():
    client = AgentRuntimeClient(api_key=os.getenv("PPIO_API_KEY"))
    
    payload = json.dumps({
        "prompt": "Tell me a story"
    }).encode()
    
    response = await client.invoke_agent_runtime(
        agentId="agent-xxxxx",
        payload=payload
    )
    
    # 处理流式响应
    print("Streaming response:")
    print(response)

多版本管理

部署新版本 Agent

修改版本号并部署新版本：

# 修改版本号
npx ppio-sandbox-cli agent configure --agent-version 1.1.0

# 部署新版本
npx ppio-sandbox-cli agent launch

部署成功后会生成新的 agent_id。Agent 的实际版本由 agent_id 控制

健康检查

默认健康检查端点

AgentRuntimeApp 自动提供 /ping 健康检查端点：

from ppio_sandbox.agent_runtime import AgentRuntimeApp

app = AgentRuntimeApp()

# 默认健康检查会自动响应 {"status": "Healthy"}

自定义健康检查

使用 @app.ping 装饰器自定义健康检查逻辑：

@app.ping
def custom_health_check():
    """自定义健康检查"""
    # 检查依赖服务
    db_ok = check_database_connection()
    llm_ok = check_llm_service()
    
    if db_ok and llm_ok:
        return {"status": "Healthy"}
    elif db_ok or llm_ok:
        return {"status": "HealthyBusy"}  # 部分可用
    else:
        return {"status": "Unhealthy"}  # 不可用

def check_database_connection():
    """检查数据库连接"""
    try:
        # 模拟数据库检查
        return True
    except:
        return False

def check_llm_service():
    """检查 LLM 服务"""
    try:
        # 模拟 LLM 服务检查
        return True
    except:
        return False

支持的健康检查状态

Agent 可以返回以下健康状态：

状态	说明	HTTP 状态码
`Healthy`	Agent 完全可用	200
`HealthyBusy`	Agent 部分可用（如正在处理大量请求）	200
`Unhealthy`	Agent 不可用	503

多轮对话

使用会话 ID 保持多轮对话

通过 runtimeSessionId 参数将多次请求发送到同一个沙箱实例：

import uuid
import json
import os
from ppio_sandbox.agent_runtime import AgentRuntimeClient

async def multi_turn_conversation():
    runtime_session_id = str(uuid.uuid4())
    client = AgentRuntimeClient(api_key=os.getenv("PPIO_API_KEY"))
    agent_id = "agent-xxxxx"
    
    # 第一轮对话
    response1 = await client.invoke_agent_runtime(
        agentId=agent_id,
        payload=json.dumps({"prompt": "你好，我叫张三"}).encode(),
        runtimeSessionId=runtime_session_id,
    )
    print(f"AI: {response1}")
    
    # 第二轮对话（发送到同一个沙箱实例，Agent 记住了上下文）
    response2 = await client.invoke_agent_runtime(
        agentId=agent_id,
        payload=json.dumps({"prompt": "我叫什么名字？"}).encode(),
        runtimeSessionId=runtime_session_id,
    )
    print(f"AI: {response2}")  # 应该回答"张三"

示例项目

我们提供了基于 LangGraph 的完整的示例项目，展示如何使用 PPIO Agent Runtime 构建真实的 AI 应用。

项目地址

🔗 https://github.com/PPIO/agent-runtime-example

产品简介

快速开始

最佳实践

Sandbox

Template

Snapshot

文件系统

Secret

Volume

工具集成

CLI 功能

旧版归档

进阶功能

目录

配置文件详解

.ppio-agent.yaml 结构说明

修改配置

修改 CPU 和内存配置

修改环境变量

重新部署使配置生效

环境变量管理

方式 1: 在配置文件中定义（仅用于 CLI 调用）

方式 2: 使用 SDK 调用时动态传递

流式响应

使用同步生成器实现流式响应

使用异步生成器实现流式响应

LangChain 流式响应示例

调用流式 Agent

多版本管理

部署新版本 Agent

健康检查

默认健康检查端点

自定义健康检查

支持的健康检查状态

多轮对话

使用会话 ID 保持多轮对话

示例项目

项目地址

​目录

​配置文件详解

​.ppio-agent.yaml 结构说明

​修改配置

​修改 CPU 和内存配置

​修改环境变量

​重新部署使配置生效

​环境变量管理

​方式 1: 在配置文件中定义（仅用于 CLI 调用）

​方式 2: 使用 SDK 调用时动态传递

​流式响应

​使用同步生成器实现流式响应

​使用异步生成器实现流式响应

​LangChain 流式响应示例

​调用流式 Agent

​多版本管理

​部署新版本 Agent

​健康检查

​默认健康检查端点

​自定义健康检查

​支持的健康检查状态

​多轮对话

​使用会话 ID 保持多轮对话

​示例项目

​项目地址

目录

配置文件详解

.ppio-agent.yaml 结构说明

修改配置

修改 CPU 和内存配置

修改环境变量

重新部署使配置生效

环境变量管理

方式 1: 在配置文件中定义（仅用于 CLI 调用）

方式 2: 使用 SDK 调用时动态传递

流式响应

使用同步生成器实现流式响应

使用异步生成器实现流式响应

LangChain 流式响应示例

调用流式 Agent

多版本管理

部署新版本 Agent

健康检查

默认健康检查端点

自定义健康检查

支持的健康检查状态

多轮对话

使用会话 ID 保持多轮对话

示例项目

项目地址