> ## Documentation Index
> Fetch the complete documentation index at: https://ppio.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# 进阶功能

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

***

## 目录

* [配置文件详解](#配置文件详解)
* [环境变量管理](#环境变量管理)
* [流式响应](#流式响应)
* [多版本管理](#多版本管理)
* [健康检查](#健康检查)
* [多轮对话](#多轮对话)
* [示例项目](#示例项目)

***

## 配置文件详解

### .ppio-agent.yaml 结构说明

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

```yaml theme={null}
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` 下修改资源配置：

```yaml theme={null}
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` 下修改环境变量：

```yaml theme={null}
spec:
  envVars:
    # LLM 配置
    MODEL_NAME: deepseek/deepseek-v3-0324
    TEMPERATURE: '0.7'
```

**注意**：

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

#### 重新部署使配置生效

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

```bash theme={null}
# 重新部署（会创建新版本）
npx ppio-sandbox-cli agent launch
```

***

## 环境变量管理

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

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

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

```yaml theme={null}
spec:
  envVars:
    MODEL_NAME: deepseek/deepseek-v3-0324
    TEMPERATURE: '0.7'
```

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

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

```python theme={null}
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）实现流式响应：

```python theme={null}
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）：

```python theme={null}
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 实现流式响应的完整示例：

```python theme={null}
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：

```python theme={null}
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

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

```bash theme={null}
# 修改版本号
npx ppio-sandbox-cli agent configure --agent-version 1.1.0

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

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

***

## 健康检查

### 默认健康检查端点

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

```python theme={null}
from ppio_sandbox.agent_runtime import AgentRuntimeApp

app = AgentRuntimeApp()

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

### 自定义健康检查

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

```python theme={null}
@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` 参数将多次请求发送到同一个沙箱实例：

```python theme={null}
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](https://github.com/PPIO/agent-runtime-example)