Quickstart

10 分钟跑通第一个 We0Agent：定义模型、系统提示词、工具，构造 agent，并使用 stream / invoke。

这篇文档按 SDK 使用顺序组织：先定义 We0Model、SystemPrompt 和 AnyToolDefinition，再构造 We0Agent，最后使用 stream() 和 invoke() 发起请求。

前置条件

Python >=3.12
已安装 uv
本机能访问私有 GitLab 仓库
至少一个模型 provider 的 API key。默认示例使用 OPENAI_API_KEY

首次调用流程

Mermaid

Rendering diagram...

1. 定义 We0Model

We0Model 来自 ProviderCard、ModelCard 和 LlmFactory。模型卡片声明能力和默认参数，provider 负责连接信息。

import os

from we0agent.domain.models.model import We0Model, ModelCard, ProviderCard
from we0agent.sdk.langchain.llm.llm_factory import LlmFactory


provider = ProviderCard(
    id="openai",
    name="OpenAI",
    family="openai",
    api_key=os.environ["OPENAI_API_KEY"],
    models=[
        ModelCard(
            id="gpt-5.4-mini",
            name="GPT-5.4 Mini",
            context=400_000,
            max_tokens=128_000,
            temperature=1.0,
        ),
    ],
)

model: We0Model = LlmFactory.create_model(provider.find_model("gpt-5.4-mini"))

2. 定义 SystemPrompt

from we0agent.prompts.core import SystemPrompt, SystemPromptBlock


system_prompt = SystemPrompt(
    blocks=[
        SystemPromptBlock(
            text="你是一个简洁、准确的中文助手。需要查看文件时，可以调用工具。",
        )
    ]
)

3. 定义 AnyToolDefinition

工具由参数 schema、执行函数和 ToolDefinitionInit 组成。下面定义一个只读文件工具：

from pathlib import Path

from pydantic import Field, BaseModel

from we0agent.domain.types.tools import AnyToolDefinition
from we0agent.domain.models.tools import ToolContext, ToolExecuteResult
from we0agent.tools.core.definition import ToolDefinitionInit, define
from we0agent.tools.core.policy import ToolExecutionPolicy


class ReadFileParameters(BaseModel):
    path: str = Field(description="Path of the UTF-8 text file to read.")


async def read_file(
    parameters: ReadFileParameters,
    context: ToolContext,
) -> ToolExecuteResult:
    del context
    path = Path(parameters.path).expanduser()
    return ToolExecuteResult(
        title=f"Read {path}",
        metadata={"path": str(path)},
        output=path.read_text(encoding="utf-8"),
    )


read_file_tool: AnyToolDefinition = await define(
    "read_file",
    ToolDefinitionInit(
        description="Read a UTF-8 text file from the local filesystem.",
        parameters=ReadFileParameters,
        execution_policy=ToolExecutionPolicy(
            is_concurrency_safe=True,
            has_side_effects=False,
        ),
        execute_handler=read_file,
    ),
).init()

tools: list[AnyToolDefinition] = [read_file_tool]

4. 构造 We0Agent

把 model、system_prompt 和 tools 传给 We0Agent：

from we0agent.agent.agent import We0Agent

agent = We0Agent(
    name="demo",
    model=model,
    tools=tools,
    system_prompt=system_prompt,
)

5. 使用 stream

stream() 返回事件流，适合聊天 UI 和长任务进度展示。

import asyncio

from we0agent.utils.time import current_time_ms
from we0agent.domain.session.part import TextPart
from we0agent.domain.session.message import We0UserMessage, MessageWithParts, We0UserMessageTime


def user_message(text: str) -> MessageWithParts:
    return MessageWithParts(
        info=We0UserMessage(time=We0UserMessageTime(created=current_time_ms())),
        parts=[TextPart(text=text)],
    )

async for event in agent.stream(
    abort=asyncio.Event(),
    session_id="demo-session",
    messages=[user_message("用三句话介绍 we0agent 的运行方式。")],
):
    if event["type"] == "text-delta":
        print(event["text"], end="", flush=True)
    elif event["type"] == "finish":
        print(f"\n[finish] {event.get('reason')}")

事件怎么读

stream() 返回的是 AsyncIterator[We0Event]。事件是 TypedDict，用 event["type"] 判断类型：

async for event in agent.stream(...):
    if event["type"] == "text-delta":
        print(event["text"], end="", flush=True)
    elif event["type"] == "tool-call":
        print(f"[tool] {event['name']} {event['input']}")
    elif event["type"] == "finish":
        print(f"[finish] {event.get('reason')}")

不要写 event.type。事件不是对象属性模型。

6. 使用 invoke

invoke() 适合命令行任务、测试和后端接口。它不逐条返回事件，而是在完整 run 结束后返回 We0TurnResult。

result = await agent.invoke(
    abort=asyncio.Event(),
    session_id="demo-session",
    messages=[user_message("把刚才的介绍整理成一句话。")],
)

print(result.result)

同一个 session_id 会沿用同一段历史；跨进程继续对话需要接入 Persistence。

运行官方示例

uv sync
export OPENAI_API_KEY="sk-..."

stream 示例运行命令见 Examples: stream_agent。示例会创建干净工作区，产物位于仓库忽略的 workspace/ 目录下。

典型输出如下：

[session step started] index=0 agent=demo model=...
[step start] index=0
...
[finish] reason=stop

无输出时查看 Troubleshooting: stream 没有输出。

第一次可以改什么

想改的东西	改哪里
换模型	看 We0Model，用业务自己的 provider 配置构造 `We0Model`。
换系统提示词	`SystemPromptBlock(text=...)`
换用户输入	`user_message("...")`
增加工具	看 AnyToolDefinition，定义业务工具的参数 schema 和执行函数。
改事件输出	看 Streaming，按 `event["type"]` 分发到 UI、日志或 WebSocket。
保存会话	看 Persistence

常见约束

每次运行都必须传新的 asyncio.Event() 作为 abort。
mode="prompt" 必须传 messages。
mode="resume" 不能传 messages。
配置结构化输出时不能用 stream()，要用 invoke()。
demo 里的 bash 工具只适合本地示例；生产环境要加权限和沙箱。

下一步

跑通后继续阅读：

看 Concepts，理解对象关系。
看 AnyToolDefinition，添加你的第一个业务工具。
看 Streaming，把事件接到 UI。
看 Examples，运行持久化、resume、snapshot 和 abort 示例。

Quickstart

On this page