进阶功能

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（部署后自动生成）

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

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

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

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

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"
    }
)

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"}

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"}

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"}

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)

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

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

from ppio_sandbox.agent_runtime import AgentRuntimeApp

app = AgentRuntimeApp()

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

@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

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}")  # 应该回答"张三"