SDK · Python

平行映射自 TypeScript 权威实现。接口签名一致，惯用法适配 Python。

实现状态

[TODO: volo-ai PyPI 包未发布，接口规格如下，可作 issue / contribution 起点]
SDK 内部基于 httpx（async）+ httpx-sse 实现 SSE 流处理。

---

安装

pip install volo-ai
# 或
poetry add volo-ai

环境要求：Python ≥ 3.10（依赖 asyncio.TaskGroup 等新语法）

---

初始化

import os
from volo_ai import VoloClient

client = VoloClient(
    api_key=os.environ["VOLO_API_KEY"],
    base_url="https://api.volo.ai",
    timeout=30.0,
)

---

1 · Chat · 流式对话

import asyncio
import uuid

async def main():
    async with client.chat.stream(
        session_id=str(uuid.uuid4()),
        user_message="你好，请用一句话介绍 VOLO AI",
        execution_mode="AUTO",
    ) as stream:
        async for event in stream:
            if event.type == "ASSISTANT_DELTA":
                print(event.content, end="", flush=True)
            elif event.type == "TOOL_START":
                print(f"\n[工具] {event.tool_name}...")
            elif event.type == "COMPLETE":
                print("\n[完成]")

asyncio.run(main())

同步版本

# 不需要 asyncio 时
for event in client.chat.stream_sync(
    session_id=session_id,
    user_message="你好",
):
    if event.type == "ASSISTANT_DELTA":
        print(event.content, end="")

停止 / 追加 / 恢复

await client.chat.stop(session_id=session_id)

await client.chat.append(
    session_id=session_id,
    appended_text="顺便加个夜间模式",
)

async with client.chat.resume(
    turn_id=pending_turn_id,
    response="选择 B 方案",
) as stream:
    async for event in stream:
        ...

---

2 · Memory · 记忆相册

上传

with open("/path/to/photo.jpg", "rb") as f:
    memory = await client.memory.upload(
        file=f,
        filename="photo.jpg",
        user_title="春节家宴",
        user_description="2026 春节，全家在北京聚餐",
    )

print(memory.memory_id, memory.thumbnail_url)

列出 / 搜索 / 删除

page = await client.memory.list(page=0, page_size=20)

results = await client.memory.search(
    query="春节",
    top_k=10,
    min_score=0.6,
)

await client.memory.delete(memory_id)

---

3 · Skill · 技能市场

# 列出
page = await client.skill.list(
    page=0,
    page_size=20,
    category="design",
    keyword="critique",
)

# 安装/卸载
await client.skill.install(skill_id)
await client.skill.uninstall(skill_id)

# 我的
mine = await client.skill.list_mine(page=0, page_size=50)

# Package 子技能
children = await client.skill.get_children(package_id)
await client.skill.toggle_child(
    package_id=package_id,
    child_skill_id=child_skill_id,
    enabled=False,
)

在对话中触发

async with client.chat.stream(
    session_id=session_id,
    user_message="/h-frontend-design 设计一个登录页",
    execution_mode="MAX",
) as stream:
    async for event in stream:
        ...

---

4 · Auth · 鉴权

result = await client.auth.login_by_email(
    email="user@example.com",
    code="123456",
)

# SDK 自动管理 Token 刷新；也可手动
tokens = await client.auth.refresh(result.refresh_token)

await client.auth.logout()

---

错误处理

from volo_ai.errors import (
    VoloAuthError,
    VoloRateLimitError,
    VoloQuotaError,
    VoloServerError,
    VoloStreamError,
)

try:
    async with client.chat.stream(...) as stream:
        async for event in stream:
            ...
except VoloAuthError:
    print("API Key 无效")
except VoloRateLimitError as e:
    print(f"被限流，重试 after {e.retry_after}s")
except VoloStreamError as e:
    print(f"SSE 流异常: {e}")

---

类型 / 数据类

from volo_ai.types import (
    AgentEvent,
    ChatRequest,
    ExecutionMode,  # Literal["FAST", "MAX", "AUTO"]
    EventType,      # Literal["ASSISTANT_DELTA", ...]
    MemoryVO,
    SkillVO,
)

数据类使用 pydantic.BaseModel，自动校验 + IDE 提示。

---

与 LangChain / LlamaIndex 集成

[TODO: VoloAI LangChain adapter 待发布——参考 OpenAI adapter 设计模式，提供 VoloAILLM / VoloAIChatModel 包装类]

---

相关

• SDK · TypeScript — 权威实现参考
• API Reference