SDK应用框架

框架概述

SDK应用框架基于Starlette Web框架实现，采用装饰器模式将用户函数封装为HTTP端点。这种设计使得开发者无需关心底层网络通信，只需专注于业务逻辑的实现。

核心入口类: HuaweiAgentRunApp

HuaweiAgentRunApp是SDK的核心类，继承自Starlette框架，提供AI代理部署所需的所有Web服务功能。

文件位置：src/hw_agentrun_wrapper/runtime/app.py

基本用法：

from hw_agentrun_wrapper import HuaweiAgentRunApp
app = HuaweiAgentRunApp()
@app.entrypoint
def my_agent(payload, context):
    return {"response": "Hello, World!"}
app.run(port=8080)

装饰器详解

@app.entrypoint - 主入口装饰器

用于注册主处理函数，处理所有到达/invocations端点的请求。

@app.entrypoint
def handle_request(payload, context):
    """
    处理Agent调用请求

    Args:
        payload: 请求载荷，包含用户输入
        context: 请求上下文，包含用户ID、会话ID等信息

    Returns:
        响应数据，可以是普通对象、生成器或异步生成器
    """
    prompt = payload.get("prompt")
    # 处理逻辑
    return {"response": f"处理: {prompt}"}

支持的返回类型：

• 普通对象：自动序列化为JSON响应
• 同步生成器：支持流式响应（SSE）
• 异步生成器：支持异步流式响应

@app.ping - 健康检查装饰器

用于注册自定义的健康检查处理器。

@app.ping
def custom_ping():
    """返回自定义的ping状态"""
    return "Healthy"  # 或返回 PingStatus.HEALTHY

PingStatus枚举值：

• HEALTHY：健康状态
• HEALTHY_BUSY：忙碌状态
• UNHEALTHY：不健康状态

@app.websocket - WebSocket装饰器

用于注册WebSocket处理函数，支持双向实时通信。

@app.websocket
async def handle_websocket(websocket, context):
    """处理WebSocket连接"""
    await websocket.accept()
    while True:
        try:
            data = await websocket.receive_text()
            # 处理消息
            await websocket.send_text(f"Echo: {data}")
        except Exception:
            break

@app.async_task - 异步任务装饰器

用于注册异步任务，SDK会自动追踪任务健康状态。

@app.async_task
async def background_task():
    """后台异步任务"""
    # 任务执行期间，ping状态会自动设置为HEALTHY_BUSY
    # 任务完成后，自动恢复为HEALTHY
    await asyncio.sleep(10)

请求上下文

SDK提供两种上下文对象用于获取请求信息：

AgentArtsRuntimeContext

基于contextvars的线程安全运行时上下文，用于在请求处理过程中存储和获取上下文信息。

from hw_agentrun_wrapper.runtime.context import AgentArtsRuntimeContext
# 设置上下文
AgentArtsRuntimeContext.set_user_id("user123")
AgentArtsRuntimeContext.set_session_id("session456")
AgentArtsRuntimeContext.set_workload_access_token("token789")
# 获取上下文
user_id = AgentArtsRuntimeContext.get_user_id()
session_id = AgentArtsRuntimeContext.get_session_id()
token = AgentArtsRuntimeContext.get_workload_access_token()

RequestContext

请求级别的上下文对象，包含请求的详细信息。

@app.entrypoint
def handle_request(payload, context: RequestContext):
    """
    context包含以下属性：
    - user_id: 用户ID
    - session_id: 会话ID
    - request_id: 请求ID
    - workload_access_token: 工作负载访问令牌
    """
    print(f"User: {context.user_id}")
    print(f"Session: {context.session_id}")
    return {"status": "ok"}

框架请求处理流程

HTTP请求
    │
    ▼
Starlette Router
    │
    ├─ POST /invocations ──▶ _handle_invocation()
    │                              │
    │                              ├─ 解析JSON payload
    │                              ├─ 构建RequestContext
    │                              ├─ 调用用户handler
    │                              ├─ 处理响应类型:
    │                              │   ├─ StreamingResponse (SSE)
    │                              │   ├─ JSONResponse
    │                              │   └─ 普通Response
    │                              └─ 返回响应
    │
    ├─ GET /ping ──▶ _handle_ping()
    │                     │
    │                     └─ 返回PingStatus
    │
    └─ WebSocket /ws ──▶ _handle_websocket()
                              │
                              └─ 传递给websocket handler