We0Agent

We0Agent 的构造参数、stream / invoke / build_request 方法与 create_agent 工厂函数完整参考。

We0Agent 是 we0agent loop framework 的主入口类。它持有一次 agent 定义中的长期配置（模型、系统提示词、工具、持久化、端口、hooks），而单次请求相关的数据（session、消息、运行模式）由 stream() 或 invoke() 在调用时传入。内部会把这些参数组装成 We0TurnInput，再交给 engine 执行。

from we0agent.agent.agent import We0Agent

Mermaid

Rendering diagram...

使用顺序

We0Agent 是长期对象，通常在进程启动或业务会话初始化时构造一次。单次用户输入通过 stream() 或 invoke() 传入。

agent = We0Agent(
    name="assistant",
    model=model,
    system_prompt=prompt,
    tools=tools,
    persistence=persistence,
    hooks=hooks,
    ports=ports,
)

stream() 适合需要实时展示文本、工具调用和状态变化的界面。每次调用都传入新的 asyncio.Event()、稳定的 session_id 和本轮新增的 messages。

import asyncio

abort = asyncio.Event()

async for event in agent.stream(
    abort=abort,
    session_id="ses_001",
    messages=[user_message("请分析这个需求，并给出执行步骤。")],
):
    if event["type"] == "text-delta":
        print(event["text"], end="", flush=True)
    elif event["type"] == "tool-call":
        print(f"\n[tool] {event['name']}")
    elif event["type"] == "finish":
        print(f"\n[finish] {event.get('reason')}")

invoke() 适合后台任务、测试、结构化输出和只关心终态的接口。它会消费完整事件流并返回 We0TurnResult。

result = await agent.invoke(
    abort=asyncio.Event(),
    session_id="ses_001",
    messages=[user_message("把刚才的分析压缩成三条结论。")],
)

print(result.result)

同一个 session_id 会复用同一段会话历史。接入持久化后，服务重启后仍可继续使用相同 session_id 追问历史内容，例如「刚才说过什么」。完整续聊案例见 Persistence。

stream 和 invoke 的选择

场景	建议
页面要实时显示模型输出	使用 `stream()`，按 `event["type"]` 更新 UI。
后台任务只需要最终状态	使用 `invoke()`。
需要结构化输出	使用 `invoke()`，并配置 `We0AgentStructuredOutput`。
需要同时消费事件并获取终态	使用 `build_request()`，先消费 `runner.events()`，结束后读取 `runner.result()`。

构造参数

We0Agent.__init__ 全部使用关键字参数（*），name、model、system_prompt 为必填。

参数	类型	必填	默认	说明
`name`	`str`	是	—	agent 的稳定标识，会写入每次 turn 的运行输入和持久化记录。
`model`	`We0Model`	是	—	当前 agent 使用的模型运行时，包含模型卡片和底层 chat model。参见 `We0Model`。
`system_prompt`	`SystemPrompt`	是	—	当前 agent 的系统提示词，按请求参与模型输入投影，默认不写入会话数据库。参见 `Prompts`。
`tools`	`Sequence[AnyToolDefinition] \| None`	否	`None`（视为空集合）	可提供给模型调用的工具定义集合，传入顺序会保留到模型工具 schema 投影阶段。参见 `Tools`。
`persistence`	`Persistence \| None`	否	`MemoryPersistence()`	会话持久化端口。省略时使用内存实现，适合测试或无外部存储的短生命周期场景。参见 `Persistence`。
`tool_choice`	`str \| None`	否	`None`	透传给底层模型请求的工具选择策略。省略时由 provider 默认策略决定。
`structured_output`	`We0AgentStructuredOutput \| None`	否	`We0AgentStructuredOutput()`	结构化输出配置。设置 `output_schema` 后只能通过 `invoke()` 请求。
`config`	`We0AgentConfig \| None`	否	`We0AgentConfig()`	agent 运行配置，控制压缩、工具裁剪、流式工具执行等行为。
`ports`	`We0AgentPorts \| None`	否	`We0AgentPorts()`	外部端口集合，接入事件发布、状态跟踪、会话回滚等能力。
`hooks`	`Sequence[We0Hook] \| None`	否	`None`（空 hook 集合）	hook handler 集合，可在模型请求和工具执行前后介入。参见 `Hooks`。

易错点：所有构造参数都是关键字参数，不能按位置传入。tools 与 hooks 传 None 时会被规范化为空集合，行为与传入空列表一致。

config（We0AgentConfig）

from we0agent.domain.models.agent import We0AgentConfig

字段	类型	默认	说明
`auto_compact`	`bool`	`False`	请求模型前是否按上下文估算结果触发自动压缩。
`reactive_compact`	`bool`	`True`	上下文溢出或模型返回高 token 用量后是否触发补救压缩。
`prune_tool`	`bool`	`True`	step 成功结束后是否裁剪历史中的旧工具结果正文。
`streaming_tool_execution`	`bool`	`True`	工具输入完成后是否立即调度工具执行。关闭后会在模型流结束后统一执行工具。

structured_output（We0AgentStructuredOutput）

from we0agent.domain.models.agent import We0AgentStructuredOutput

字段	类型	默认	说明
`output_schema`	`type[BaseModel] \| None`	`None`	Pydantic schema 类型，用于要求模型返回可解析的结构化结果。
`method`	`StructuredOutputMethod \| None`	`None`	透传给底层 SDK 的结构化输出方式，可选 `function_calling`、`json_mode` 或 `json_schema`。

约束：一旦 output_schema 非空，stream() 会直接抛出 ValueError。结构化输出只能通过 invoke() 获取。

ports（We0AgentPorts）

from we0agent.domain.models.agent import We0AgentPorts。所有端口均为可选，省略时对应能力不启用。

字段	类型	默认	说明
`event_publishers`	`list[We0EventPublisher]`	`[]`	外部事件发布端口集合，接收 engine 产生的 We0 事件。基础 stream 输出由 runner 管理，与此独立。
`status_tracker`	`We0StatusTracker \| None`	`None`	状态跟踪端口，用于写入当前 session 的运行状态。
`session_revert`	`SessionRevertCapability \| None`	`None`	会话回滚增强能力，包含代码快照、变更摘要、revert、unrevert 和 cleanup 能力。参见 `Revert` 和 `Built-ins`。
`context_manager`	`We0ContextManager \| None`	`None`	上下文管理端口，提供动态 user meta 构建和 system reminder 插入能力。

请求级参数

stream()、invoke() 和 build_request() 共享同一组关键字参数，描述单次请求的运行意图。

参数	类型	必填	默认	说明
`abort`	`asyncio.Event`	是	—	当前请求的中断信号。调用方设置该事件后，engine 会按当前运行阶段执行中断收口。参见 `Abort`。
`session_id`	`str`	是	—	当前请求所属的会话标识。持久化、状态跟踪和事件都会使用该标识。
`mode`	`We0RunMode`	否	`"prompt"`	请求模式。`prompt` 追加新用户输入并运行；`resume` 基于已有会话恢复未完成的 assistant message 或工具调用。参见 `Resume`。
`messages`	`We0Messages \| None`	否*	`None`	`prompt` 模式下追加到会话的新消息（`list[MessageWithParts]`）。`resume` 模式下必须省略。
`parent_id`	`str \| None`	否	`None`	当前请求的父会话标识，通常用于 fork agent、compact 或业务侧会话派生关系。
`max_steps`	`int \| None`	否	`None`	本次请求允许执行的最大 step 数。省略时使用 agent 配置中的默认行为。
`persistence`	`Persistence \| None`	否	构造时的端口	本次请求覆盖使用的持久化端口。省略时沿用 agent 构造时的持久化端口。
`we0_source`	`str`	否	`"main"`	当前请求来源标识，例如主 agent、fork agent 或业务侧自定义来源。
`metadata`	`We0RequestMetadata \| None`	否	`None`（视为空 dict）	当前请求运行元数据，会传入状态跟踪、事件发布和 context manager 等外部端口。

*关于 messages 的校验：prompt 模式下若 messages 为 None 抛出 ValueError("messages is required in prompt mode.")；resume 模式下若传入 messages 抛出 ValueError("messages is not allowed in resume mode.")。

We0RequestMetadata 定义在 we0agent.domain.types.metadata，本质是 dict[str, JsonValue | We0Observability]，其中 We0Observability 可携带 trace_id、user_id、session_id 等可观测性字段。

stream

async def stream(*, abort, session_id, mode="prompt", messages=None,
                 parent_id=None, max_steps=None, persistence=None,
                 we0_source="main", metadata=None) -> AsyncIterator[We0Event]

以事件流方式运行一次请求，返回 AsyncIterator[We0Event]。迭代过程中会依次产出模型流事件、工具事件、step 状态事件、session 状态事件以及最终收口事件。事件结构参见 Events 与 Streaming。

import asyncio

abort = asyncio.Event()

async for event in agent.stream(
    abort=abort,
    session_id="demo-session",
    messages=[user_message("用三句话介绍 we0agent 的运行方式。")],
):
    # 事件是 TypedDict，用 event["type"] 区分形态
    if event["type"] == "text-delta":
        print(event["text"], end="", flush=True)
    elif event["type"] == "tool-call":
        print(f"\n[tool] {event['name']} {event['input']}")
    elif event["type"] == "finish":
        print(f"\n[finish] {event.get('reason')}")

约束：当 agent 配置了 structured_output.output_schema 时，stream() 会抛出 ValueError("Structured output is only supported by invoke, not stream.")。需要结构化结果请改用 invoke()。

invoke

async def invoke(*, abort, session_id, mode="prompt", messages=None,
                 parent_id=None, max_steps=None, persistence=None,
                 we0_source="main", metadata=None) -> We0TurnResult

以最终结果方式运行一次请求。内部会消费完整事件流，最后返回 We0TurnResult。适合不需要实时进度、只关心终态的同步式工作流，以及结构化输出场景。

import asyncio

result = await agent.invoke(
    abort=asyncio.Event(),
    session_id="demo-session",
    messages=[user_message("把上面的结论整理成一段摘要。")],
)

print(result.result)  # "stop" 或 "compact"

We0TurnResult 定义在 we0agent.engine.query，对外暴露一个字段：

字段	类型	说明
`result`	`We0TurnResultType`	本次 run 的终态原因，取值为 `"stop"`（正常结束）或 `"compact"`（因压缩流程结束）。

build_request

def build_request(*, abort, session_id, mode="prompt", messages=None,
                  parent_id=None, max_steps=None, persistence=None,
                  we0_source="main", metadata=None) -> We0TurnRunner

构建单次请求运行器并返回 We0TurnRunner，而不立即开始消费。stream() 和 invoke() 都是它的薄封装：前者等价于 build_request(...).events()，后者等价于 build_request(...).result()。当你需要在同一个 runner 上既订阅事件流、又获取最终结果时，可直接使用它。

import asyncio

runner = agent.build_request(
    abort=asyncio.Event(),
    session_id="demo-session",
    messages=[user_message("开始任务")],
)

# 既消费事件流，又在结束后取终态结果
async for event in runner.events():
    if event["type"] == "text-delta":
        print(event["text"], end="")

result = await runner.result()

约束：We0TurnRunner 的事件流只能消费一次，重复调用 events() 会抛出 RuntimeError。events() 与 result() 共享同一个底层 producer，二者可在同一 runner 上配合使用。

create_agent

函数式工厂入口，等价于直接构造 We0Agent，区别在于 model、system_prompt、tools 为位置参数，其余为关键字参数，且 name 默认值为 "assistant"。

from we0agent.agent.agent import create_agent

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

参数	形式	默认	说明
`model`	位置	—	当前 agent 使用的模型运行时。
`system_prompt`	位置	—	当前 agent 的系统提示词。
`tools`	位置	`None`	工具定义集合。
`name`	关键字	`"assistant"`	agent 的稳定标识。
`tool_choice`	关键字	`None`	工具选择策略。
`persistence`	关键字	`None`	会话持久化端口。
`structured_output`	关键字	`None`	结构化输出配置。
`config`	关键字	`None`	agent 运行配置。
`ports`	关键字	`None`	外部端口集合。
`hooks`	关键字	`None`	hook handler 集合。

create_agent 与 We0Agent(...) 唯一可观察的差异是 name 的默认值：直接构造时 name 必填，工厂函数省略时为 "assistant"。

We0Agent

On this page