类class agentscope.sop

SOPStepBase

class SOPStepBase(ABC)

源码Source agentscope/sop/_schema.py:65–162

SOP 步骤基类，把复杂任务拆分为可执行步骤并驱动流程。位于 agentscope.sop 包，继承自 ABC。核心成员包括 reply_stream、record。

Subclasses put whatever they like in :meth:reply_stream — the contract is only that one call is one attempt, and that it either parks or files a verdict on the state it was handed.

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(subject: str, description: str, max_attempts: int = 3) -> None`	初始化 `SOPStepBase` 实例。 Initialize the step.

属性摘要Attribute Summary

属性Attribute	类型Type	说明Description	值Value
`state_type`	ClassVar[type[SOPStepRunState]]	状态类型字段。	`SOPStepRunState`

方法摘要Method Summary

方法Method	说明Description
`def reply_stream(inputs:` `Msg \| list[Msg] \| UserConfirmResultEvent \| UserInterruptEvent \| ExternalExecutionResultEvent \|…`abstractabstractmethod	流式回复输入消息，并逐个产出事件或消息。 Make one attempt at this step, streaming what happens.
`def record(state:` `SOPStepRunState, passed: bool, message: str = '', verifier: str = '') -> None`	执行记录相关操作。 File a verdict on the current attempt.

构造器详情Constructor Detail

SOPStepBase

def __init__(subject: str, description: str, max_attempts: int = 3) -> None

初始化 SOPStepBase 实例。

Initialize the step.

参数Parameters

名称Name	类型Type	说明Description
`subject`	str	主题（类型 `str`）。A brief, actionable name.
`description`	str	描述（类型 `str`）。What this step must achieve — the destination, not the route.
`max_attempts`默认default `3`	int	最大尝试次数（类型 `int`）。How many refusals before the run gives up on it. Enforced by the engine, not here.

源码Source agentscope/sop/_schema.py:78–98

属性详情Attribute Detail

state_type

ClassVar[type[SOPStepRunState]] = SOPStepRunState

状态类型字段。声明为 ClassVar[type[SOPStepRunState]]。

The state type field.

源码Source agentscope/sop/_schema.py:73

方法详情Method Detail

reply_stream

def reply_stream(inputs: Msg | list[Msg] | UserConfirmResultEvent | UserInterruptEvent | ExternalExecutionResultEvent | None, state: SOPStepRunState) -> AsyncGenerator[AgentEvent | Msg, None]

abstractabstractmethod

流式回复输入消息，并逐个产出事件或消息。

Make one attempt at this step, streaming what happens.

参数Parameters

名称Name	类型Type	说明Description
`inputs`	Msg \| list[Msg] \| UserConfirmResultEvent \| UserInterruptEvent \| ExternalExecutionResultEvent \| None	输入消息（类型 `Msg \| list[Msg] \| UserConfirmResultEvent \| UserInterruptEvent \| ExternalExecutionResultEvent \| None`）。What the engine handed over for a fresh attempt, or the answer a parked attempt was waiting for.
`state`	SOPStepRunState	状态对象（类型 `SOPStepRunState`）。This step's record in the run. Read it to pick up where the last call stopped; write its phase, submission and verdicts as the attempt goes.

返回Returns AsyncGenerator[AgentEvent | Msg, None]

以异步流的形式返回AgentEvent|消息,None。`AgentEvent | Msg`:

源码Source agentscope/sop/_schema.py:101–125

record

def record(state: SOPStepRunState, passed: bool, message: str = '', verifier: str = '') -> None

执行记录相关操作。

A refusal clears the submission, so the next attempt starts from the work rather than from the judging.

参数Parameters

名称Name	类型Type	说明Description
`state`	SOPStepRunState	状态对象（类型 `SOPStepRunState`）。The step's record in the run, which this writes to.
`passed`	bool	passed（类型 `bool`）。Whether the attempt is accepted.
`message`默认default `''`	str	消息（类型 `str`）。Why it was refused, handed to the executor verbatim on the next attempt.
`verifier`默认default `''`	str	校验器（类型 `str`）。Who decided.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/sop/_schema.py:127–162