zhuye (Python SDK)

让 Python 项目直接驱动 Zhuye Agent 的轻量 SDK。底层通过 spawn zhuye serve 子进程、走 JSONL stdio 协议,零运行时依赖(仅 stdlib)。

安装

pip install zhuye

需要本机有 zhuye 二进制在 PATH(或通过 ZHUYE_BIN 指向)。从 GitHub Releases 下载,或从本仓库 make build 产出 target/release/zhuye。

用法

from zhuye import ZhuyeAgent, ToolOutput, ContentBlock

agent = ZhuyeAgent.spawn()

# 注册客户端自定义工具(LLM 调用时由本地函数返回结果)
def get_weather(args):
    return ToolOutput(
        content=[ContentBlock(type="text", text=f"{args['city']} 今天晴 26℃")],
        is_error=False,
        metadata={},
        elapsed_ms=0,
    )

agent.register_tool(
    name="get_weather",
    description="查询城市天气",
    parameters={
        "type": "object",
        "properties": {"city": {"type": "string"}},
        "required": ["city"],
    },
    handler=get_weather,
)

# 高层:聚合 delta,返回最终文本
print(agent.prompt("北京天气如何?"))

# 低层:遍历 turn 事件
for ev in agent.submit("第二轮"):
    msg = ev["msg"]
    if msg["type"] == "agent_message_delta":
        print(msg["delta"], end="", flush=True)
    if msg["type"] == "turn_complete":
        break

agent.close()

全局事件

next_global_event(timeout_secs=5) 读取 id="" 的会话事件,与 submit() 的 按 submission 路由的事件迭代器独立。超时抛 queue.Empty,进程 stdout 关闭后返回 None;未知变体原样返回。

handshake = agent.next_global_event(timeout_secs=5)

环境变量

测试

pip install -e .[dev]
pytest -ra

mock serve 进程跑在 Node 上(Node ≥ 18),不依赖真 zhuye 二进制 —— 测试完全离线,也不需要网络/API key。

端到端测试(对真 zhuye serve 子进程的集成)在仓库 CI 里跑;在仓库根配置好 ZHUYE_BIN 后执行 ./tests/run_all.sh sdk-python。

协议

SDK 与 zhuye serve 之间的 wire 协议见 ../PROTOCOL.md(同仓库 sdks/PROTOCOL.md)。Rust 核心新增协议变体时,SDK 自动降为 UnknownEventMsg,不会中断下游。