多智能体编排

    max_round: int = 10
    admin_name: str = "Admin"
    func_call_filter: bool = True
    speaker_selection_method: Literal["auto", "manual", "random", "round_robin"] | Callable[..., Any] = "auto"
    max_retries_for_selecting_speaker: int = 2
    allow_repeat_speaker: bool | list[Agent] | None = None
    allowed_or_disallowed_speaker_transitions: dict[str, Any] | None = None
    speaker_transitions_type: Literal["allowed", "disallowed", None] = None
    enable_clear_history: bool = False
    send_introductions: bool = False
    select_speaker_message_template: str = """You are in a role play game. The following roles are available:
                {roles}.
                Read the following conversation.

logger = logging.getLogger(__name__)


class SendMessage(FunctionTool):
    """
    Use this tool to facilitate direct, synchronous communication between specialized agents within your agency.

    When you send a message using this tool, you receive a response exclusively from the designated recipient
    agent. To continue the dialogue, invoke this tool again with the desired recipient agent and your follow-up
    message. Remember, communication here is synchronous; the recipient agent won't perform any tasks post-response.

    You are responsible for relaying the recipient agent's responses back to the user, as the user does not have
    direct access to these replies. Keep engaging with the tool for continuous interaction until the task is fully

        fields = output_model.model_fields
        field_descriptions = [f"{field}: {fields[field]}" for field in fields]
        schema = "\n".join(field_descriptions)
        response = self.ApiClient.prompt_agent(
            agent_id=self.agent_id,
            prompt_name="Convert to JSON",
            prompt_args={
                "user_input": input_string,
                "schema": schema,
                "conversation_name": "AGiXT Terminal",
            },
        )
        response = str(response).split("```json")[1].split("```")[0].strip()

            **kwargs,
        )

    async def call(self, target: str, *args, **kwargs) -> dict:
        """
        Initiates a cross-agent call to another reasoner or skill via the AgentField execution gateway.

        This method allows agents to seamlessly communicate and utilize reasoners/skills
        deployed on other agent nodes within the AgentField ecosystem. It properly propagates
        workflow tracking headers and maintains execution context for DAG building.

        **Return Type**: Always returns JSON/dict objects, similar to calling any REST API.
        No automatic schema conversion is performed - developers can convert to Pydantic

        """
        skillbook = skillbook or Skillbook()
        steps: list[StepProtocol[ACEStepContext]] = [
            AgentStep(agent, skillbook),
            EvaluateStep(environment),
            *learning_tail(
                reflector,
                skill_manager,
                skillbook,
                dedup_manager=dedup_manager,
                dedup_interval=dedup_interval,
                checkpoint_dir=checkpoint_dir,
                checkpoint_interval=checkpoint_interval,

        finally:
            pass

    async def observe(self, msgs: Msg | list[Msg] | None = None) -> None:
        """Receive external observation message(s) and save them into
        context."""
        await self._handle_incoming_messages(msgs)

    async def compress_context(
        self,
        context_config: ContextConfig | None = None,
    ) -> None:
        """Compress the agent's context if the token count exceeds the

# class Rule(BaseModel):
class SimulationRule(BaseRule):
    """
    Rule for the environment. It controls the speaking order of the agents
    and maintain the set of visible agents for each agent.
    """

    order: BaseOrder
    visibility: BaseVisibility
    selector: BaseSelector
    updater: BaseUpdater

        return DefaultNodeExecutionStrategy()


class WorkflowEngine(BaseModel):
    """
    Main workflow execution engine.

    Orchestrates the execution of workflow nodes using depth-first search,
    manages error handling, retry mechanisms, and provides various execution
    strategies for different node types.
    """

    engine_ctx: WorkflowEngineCtx = None  # type: ignore

from ._sequential_routed_agent import SequentialRoutedAgent


class BaseGroupChatManager(SequentialRoutedAgent, ABC):
    """Base class for a group chat manager that manages a group chat with multiple participants.

    It is the responsibility of the caller to ensure:
    - All participants must subscribe to the group chat topic and each of their own topics.
    - The group chat manager must subscribe to the group chat topic.
    - The agent types of the participants must be unique.
    - For each participant, the agent type must be the same as the topic type.

    Without the above conditions, the group chat will not function correctly.

 * This system can scale to hundreds of sub-agents, each with dozens or even hundreds of tools.
 * It is designed to handle complex inquiries by allowing agents to hand off requests to other agents.
 */
export class MultiAgentOrchestrator {
  private _agents: SubAgent[]
  private _currentAgentName: string

  private _previousHandoffs: Array<{
    from: string
    to: string
    reason: string
  }> = []

DEFAULT_RELAY_URL = "wss://oo.openonion.ai"


def host(
    create_agent: Callable,
    port: int = None,
    trust: Union[str, "Agent"] = None,
    result_ttl: int = None,
    workers: int = None,
    reload: bool = None,
    *,
    relay_url: str = DEFAULT_RELAY_URL,
    blacklist: list | None = None,

This keeps bus payloads small, preserves durability, and allows audit tooling
to dereference pointers when needed (`GET /api/v1/memory?ptr=...`).

## 5) Control Plane Flow (High Level)

1. Gateway writes context JSON to Redis, sets `context_ptr`, publishes
   `BusPacket{JobRequest}` to `sys.job.submit`.
2. Scheduler receives `JobRequest`, records state, calls Safety Kernel, and
   applies `PolicyConstraints`.
3. Scheduler selects a pool and/or direct worker subject using heartbeats
   and publishes to `job.*` or `worker.<id>.jobs`.
4. Worker fetches `context_ptr`, executes, writes result to Redis, publishes
   `BusPacket{JobResult}` with `result_ptr` and `artifact_ptrs`.

from enum import Enum


class Process(str, Enum):
    """
    Class representing the different processes that can be used to tackle tasks
    """

    sequential = "sequential"
    hierarchical = "hierarchical"
    # TODO: consensual = 'consensual'

// The actual tool name is dynamic: `run_<agent_name>`.
export const RUN_AGENT_PLACEHOLDER_TOOL_NAME = "run_agent" as const;

export const RUN_AGENT_CONFIGURABLE_PROPERTIES = {
  executionMode: z
    .object({
      options: z
        .union([
          z
            .object({
              value: z.literal("run-agent"),
              label: z.literal("Agent runs in the background"),
            })

}

/// Parent PostToolUse: bootstrap, messages, vanilla binding.
fn handle_posttooluse(
    db: &HcomDb,
    ctx: &HcomContext,
    payload: &HookPayload,
    instance_name: &str,
    instance_data: &InstanceRow,
    updates: &serde_json::Map<String, Value>,
) -> (i32, String) {
    let tool_name = payload.tool_name.as_str();
    let mut outputs: Vec<Value> = Vec::new();

#!/usr/bin/env python3
"""
Delegate Tool -- Subagent Architecture

Spawns child AIAgent instances with isolated context, restricted toolsets,
and their own terminal sessions. Supports single-task and batch (parallel)
modes. The parent blocks until all children complete.

Each child gets:
  - A fresh conversation (no parent history)

    branch_timeout_seconds: float = 300.0


class Orchestrator:
    """
    Executes agent graphs.

    Example:
        executor = GraphExecutor(
            runtime=runtime,
            llm=llm,
            tools=tools,
            tool_executor=my_tool_executor,

    pass


class Sequential(Agent):
    """Sequential is an agent container that forwards messages to each agent
    in the order they are added."""

    def __init__(self, *agents: Union[Agent, Iterable], **kwargs):
        super().__init__(**kwargs)
        self._agents = OrderedDict()
        if not agents:
            raise ValueError('At least one agent should be provided')
        if isinstance(agents[0], Iterable) and not isinstance(agents[0], Agent):

"""Surface nested named agents as typed `run.subagents` handles.

Detects subagents via the `lc_agent_name` transition that langgraph's base
`_TasksLifecycleBase` now computes. `create_agent(name=...)` binds
`lc_agent_name` into the run config; the base transformer records, per
namespace, the `lc_agent_name` seen on each task start (first-write-wins).

A subagent boundary is a nested run whose `lc_agent_name` is set *and* differs
from its parent namespace's `lc_agent_name`. Plain subgraphs inherit the
parent's name (so they compare equal and are excluded); unnamed agents have

    """Metaclass for AgentWorkflow that inherits from WorkflowMeta."""


class AgentWorkflow(Workflow, PromptMixin, metaclass=AgentWorkflowMeta):
    """A workflow for managing multiple agents with handoffs."""

    def __init__(
        self,
        agents: List[BaseWorkflowAgent],
        initial_state: Optional[Dict] = None,
        root_agent: Optional[str] = None,
        handoff_prompt: Optional[Union[str, BasePromptTemplate]] = None,
        handoff_output_prompt: Optional[Union[str, BasePromptTemplate]] = None,

                exc_info=True,
            )

    async def _try_start_new_cycle(self) -> bool:
        """
        Atomically checks conditions and starts a new evolution cycle if possible.
        """
        async with self._iteration_lock:
            if (
                len(self._running_tasks) < self.max_workers
                and self._current_iteration < self.max_iterations
                and not self._stop_event.is_set()
            ):

# Claude 3 Haiku    claude-3-haiku-20240307
# Claude 3.5 Sonnet claude-3-5-sonnet-20240620

ORCHESTRATOR_MODEL = "claude-3-5-sonnet-20240620"
SUB_AGENT_MODEL = "claude-3-5-sonnet-20240620"
REFINER_MODEL = "claude-3-5-sonnet-20240620"

def calculate_subagent_cost(model, input_tokens, output_tokens):
    # Pricing information per model
    pricing = {
        "claude-3-opus-20240229": {"input_cost_per_mtok": 15.00, "output_cost_per_mtok": 75.00},
        "claude-3-haiku-20240307": {"input_cost_per_mtok": 0.25, "output_cost_per_mtok": 1.25},
        "claude-3-sonnet-20240229": {"input_cost_per_mtok": 3.00, "output_cost_per_mtok": 15.00},

   * }
   * ```
   */
  async network(
    messages: MessageListInput,
    options?: MultiPrimitiveExecutionOptions<undefined>,
  ): Promise<MastraAgentNetworkStream<undefined>>;
  async network<OUTPUT extends {}>(
    messages: MessageListInput,
    options?: MultiPrimitiveExecutionOptions<OUTPUT>,
  ): Promise<MastraAgentNetworkStream<OUTPUT>>;
  async network<OUTPUT = undefined>(messages: MessageListInput, options?: MultiPrimitiveExecutionOptions<OUTPUT>) {
    const requestContextToUse = options?.requestContext || new RequestContext();

)


class Team(BaseModel):
    """
    Team: Possesses one or more roles (agents), SOP (Standard Operating Procedures), and a env for instant messaging,
    dedicated to env any multi-agent activity, such as collaboratively writing executable code.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    env: Optional[Environment] = None
    investment: float = Field(default=10.0)

// The message is sent synchronously and the function will block until a response is received or the timeout is reached.
// The timeout can be set using the WithTimeout option. If there is no timeout set, it will wait indefinitely until a
// response is received, or until calling function is cancelled.
func SendMessage(agentId, msgName string, options ...messageOption) (*string, error) {
	m := &message{
		name:    msgName,
		timeout: math.MaxInt64, // default to no timeout (wait indefinitely)
	}

	for _, opt := range options {
		opt(m)
	}

            self._status.error = str(context.error)


class SubagentManager:
    """Manages background subagent execution."""

    def __init__(
        self,
        provider: LLMProvider,
        workspace: Path,
        bus: MessageBus,
        max_tool_result_chars: int,
        model: str | None = None,

   * @param team - A team created via {@link createTeam} (or `new Team(...)`).
   * @param goal - High-level natural-language goal for the team.
   */
  async runTeam(
    team: Team,
    goal: string,
    options?: RunTeamOptions,
  ): Promise<TeamRunResult> {
    const agentConfigs = team.getAgents()
    const coordinatorOverrides = options?.coordinator

    // ------------------------------------------------------------------
    // Short-circuit: skip coordinator for simple, single-action goals.

/** Implements ACP subagent/session spawning, binding, limits, and parent-stream setup. */
import crypto from "node:crypto";
import fs from "node:fs/promises";
import {
  resolveAcpSessionCwd,
  resolveAcpThreadSessionDetailLines,
} from "@openclaw/acp-core/runtime/session-identifiers";
import type { AcpRuntimeSessionMode } from "@openclaw/acp-core/runtime/types";
import {
  normalizeOptionalLowercaseString,

<tr>
<td width="50%" valign="top">

**Addressing**
- 48-bit virtual addresses (`N:NNNN.HHHH.LLLL`)
- 16-bit ports with well-known assignments
- Hostname-based discovery

**Transport**
- Reliable streams (TCP-equivalent)
- Sliding window, SACK, congestion control (AIMD)
- Flow control (advertised receive window)
- Nagle coalescing, auto segmentation, zero-window probing

from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup


class ParallelPipeline(BasePipeline):
    """Pipeline that processes frames through multiple sub-pipelines concurrently.

    Creates multiple parallel processing branches from the provided processor lists,
    coordinating frame flow and ensuring proper synchronization of lifecycle events
    like EndFrames. Each branch runs independently while system frames are handled
    specially to maintain pipeline coordination.
    """

    def __init__(self, *args):

            max_iter=self.max_iter
        )
        
        if self.process == "workflow":
            for task_id in process.workflow():
                self.run_task(task_id)
        elif self.process == "sequential":
            for task_id in process.sequential():
                self.run_task(task_id)
        elif self.process == "hierarchical":
            for task_id in process.hierarchical():
                if isinstance(task_id, Task):
                    task_id = self.add_task(task_id)

/// </summary>
/// <typeparam name="TInput">The type of the input to the orchestration.</typeparam>
/// <typeparam name="TOutput">The type of the result output by the orchestration.</typeparam>
public abstract partial class AgentOrchestration<TInput, TOutput>
{
    /// <summary>
    /// Initializes a new instance of the <see cref="AgentOrchestration{TInput, TOutput}"/> class.
    /// </summary>
    /// <param name="members">Specifies the member agents or orchestrations participating in this orchestration.</param>
    protected AgentOrchestration(params Agent[] members)
    {
        // Capture orchestration root name without generic parameters for use in
        // agent type and topic formatting as well as logging.

            logger.warning("Graph without execution limits may run indefinitely if cycles exist")


class Graph(MultiAgentBase):
    """Directed Graph multi-agent orchestration with configurable revisit behavior."""

    def __init__(
        self,
        nodes: dict[str, GraphNode],
        edges: set[GraphEdge],
        entry_points: set[GraphNode],
        max_node_executions: int | None = None,
        execution_timeout: float | None = None,

            case Result() as result:
                return result

            case Agent() as agent:
                return Result(
                    value=json.dumps({"assistant": agent.name}),
                    agent=agent,
                )
            case _:
                try:
                    return Result(value=str(result))
                except Exception as e:
                    error_message = f"Failed to cast response to string: {result}. Make sure agent functions return a string or Result object. Error: {str(e)}"

// Core: Spawn a Native Subagent
// ---------------------------------------------------------------------------

export function spawnSubagent(
  input: SpawnSubagentInput,
  context: SubagentContext,
): Promise<SubagentHandle> {
  return spawnSubagentImpl(input, context)
}

async function spawnSubagentImpl(
  input: SpawnSubagentInput,
  context: SubagentContext,

    pass


class SwarmRouter:
    """
    A class that dynamically routes tasks to different swarm types based on user selection or automatic matching.

    The SwarmRouter enables flexible task execution by either using a specified swarm type or automatically determining
    the most suitable swarm type for a given task. It handles task execution while managing logging, type validation,
    and metadata capture.

    Args:
        name (str, optional): Name identifier for the SwarmRouter instance. Defaults to "swarm-router".

from .result_combiner import ResultCombiner


class Team:
    """A callable class for multi-agent operations using the Upsonic client.
    
    Supports both Agent and Team instances as entities, enabling
    hierarchical multi-agent workflows with nested teams.
    Streaming (stream/astream) is text-only; no events are yielded.
    """
    
    def __init__(self,
                 entities: Optional[List[Union[Agent, "Team"]]] = None,

/**
 * SubAgentManager - Manages sub-agents and delegation functionality for an Agent
 */
export class SubAgentManager {
  /**
   * The name of the agent that owns this sub-agent manager
   */
  private agentName: string;

  /**
   * Sub-agent configurations that the parent agent can delegate tasks to
   */
  private subAgentConfigs: SubAgentConfig[] = [];