推理循环

        return False, None

    def generate_reply(
        self,
        messages: list[dict[str, Any]] | None = None,
        sender: Optional["Agent"] = None,
        exclude: Container[Any] = (),
    ) -> str | dict[str, Any] | None:
        """Reply based on the conversation history and the sender.

        Either messages or sender must be provided.
        Register a reply_func with `None` as one trigger for it to be activated when `messages` is non-empty and `sender` is `None`.

                logger.warning(f"Entering async context for server {server.name}")
                await mcp_stack.enter_async_context(server)  # type: ignore[arg-type]

        result = await Runner.run(
            starting_agent=agent,
            input=history_for_runner,
            context=master_context_for_run,
            hooks=hooks_override,
            run_config=with_codex_model_input_role_rewrite(run_config_override or RunConfig()),
            max_turns=kwargs.get("max_turns", 1000000),
        )
    return result

        yield {"type": "answer", "content": final_answer, "complete": True}

    def extract_commands_from_response(self, response):
        # Extract all <execute>...</execute> blocks
        command_blocks = re.findall(r"(<execute>.*?</execute>)", response, re.DOTALL)
        extracted_commands = []
        for command_block in command_blocks:
            # Extract the content inside <execute>...</execute>
            command_content = re.search(
                r"<execute>(.*?)</execute>", command_block, re.DOTALL
            ).group(1)
            # Extract the command name
            name_match = re.search(r"<name>(.*?)</name>", command_content, re.DOTALL)

   * @returns A Promise resolving to the `StreamTextResult` object from the Vercel AI SDK,
   *          containing the live stream and usage promise for the caller to handle.
   */
  async handleMessage(
    options: AgentNodeHandleMessageOptions
  ): Promise<AgentDockStreamResult<Record<string, CoreTool>, any>> {
    const {
      messages,
      sessionId,
      orchestrationManager,
      systemOverride,
      useFallback = false,
      config: runtimeOverrides,

    )


async def execute_tool_call_loop(
    agent: "Agent",
    messages: List[Dict[str, Any]],
    tools: List[Dict[str, Any]],
    config: ToolCallConfig,
    needs_lazy_hydration: bool,
    litellm_params: Dict[str, Any],
    make_completion: Callable,
) -> tuple[Any, ToolCallTrace]:
    """Execute the LLM tool-call loop.

# ---------------------------------------------------------------------------


class Pipeline:
    """Ordered sequence of steps.  Satisfies StepProtocol — can be nested.

    Build via the fluent API::

        pipe = (
            Pipeline()
            .then(AgentStep())
            .then(EvaluateStep())
            .then(ReflectStep())   # ReflectStep.async_boundary = True

        # Step 3: Enter the reasoning-acting loop until reaching max_iters or
        #  no more tool calls to execute
        # ===================================================================
        while self.state.cur_iter < self.react_config.max_iters:
            # ===============================================================
            # Step 3.1:
            # ===============================================================
            action, data = self._check_next_action()
            if action == "exit" and isinstance(data, Msg):
                yield data
                return

            # ===============================================================

  const lastMessage = messages[messages.length - 1]!.content as string;

  for (let i = 0; i < maxEvals; i++) {
    const { queries: newQueries, totalTokens: queriesTokens } =
      await generateQueries(model, messages, queries);

    newQueries.forEach((q) => {
      if (queries.some((q2) => q2.query === q.query)) return;
      queries.push(q);
    });

    totalTokens += queriesTokens;

        )
        super().__init__(rule=rule, **kwargs)

    async def step(self) -> List[Message]:
        """Run one step of the environment"""

        # Get the next agent index
        agent_ids = self.rule.get_next_agent_idx(self)

        # Generate current environment description
        env_descriptions = self.rule.get_env_description(self)

        # Generate the next message

                knowledge_config.unwrap_or_default()
            ).await?;
            let tool_descs = Self::get_tool_descs(&tools);
            loop {
                let mut assistant_msg_delta = MessageDelta::new().with_role(Role::Assistant);
                {
                    let mut model = self.lm.clone();
                    let mut strm = model.infer_delta(
                        messages.clone(),
                        tool_descs.clone(),
                        docs.clone(),
                        inference_config.clone().unwrap_or_default()
                    );

            cot_step.action_output = plugin_response.result
            yield AgentResponse(typ="cot_step", content=cot_step, model=self.model.name)

    async def run(
        self, span: Span, node_trace_log: NodeTraceLog
    ) -> AsyncIterator[AgentResponse]:
        """cot run"""

        with span.start("RunCotAgent") as sp:

            system_prompt = await self.create_system_prompt()
            user_prompt_template = await self.create_user_prompt()

            yield model_result

    @classmethod
    async def _process_model_result(
        cls,
        model_result: CreateResult,
        inner_messages: List[BaseAgentEvent | BaseChatMessage],
        cancellation_token: CancellationToken,
        agent_name: str,
        system_messages: List[SystemMessage],
        model_context: ChatCompletionContext,
        workbench: Sequence[Workbench],
        handoff_tools: List[BaseTool[Any, Any]],

  })

  try {
    while (true) {
      if (ctx.iterations.length >= ctx.loop) {
        return new ErrorExecutionResult(ctx, new LoopExceededError())
      }

      const iteration = await ctx.nextIteration()

      try {
        await executeOnIterationStartHook({
          iteration,

            {"role": "user", "content": prompt}
        ]

    def _run_iteration_loop(self, max_iterations: int) -> str:
        """Run the main LLM/tool iteration loop until complete or max iterations."""
        while self.current_session['iteration'] < max_iterations:
            self.current_session['iteration'] += 1

            # Fire before_iteration (poll IO, check mode changes)
            self._invoke_events('before_iteration')

            # Get LLM response
            response = self._get_llm_decision()

                msg["files"] = files
                break

    def _invoke_loop(self) -> AgentFinish:
        """Execute agent loop until completion.

        Checks if the LLM supports native function calling and uses that
        approach if available, otherwise falls back to the ReAct text pattern.

        Returns:
            Final answer from the agent.
        """
        use_native_tools = (

  }
}

export async function agentLoopWorkflow({
  authType,
  initialStartTime,
  agentLoopArgs,
  startStep,
}: {
  authType: AuthenticatorType;
  initialStartTime: number;
  agentLoopArgs: AgentLoopArgs;
  startStep: number;

            return component_output

    def run(  # noqa: PLR0915, PLR0912, C901
        self,
        data: dict[str, Any],
        include_outputs_from: set[str] | None = None,
        *,
        break_point: Breakpoint | AgentBreakpoint | None = None,
        pipeline_snapshot: PipelineSnapshot | None = None,
        snapshot_callback: SnapshotCallback | None = None,
    ) -> dict[str, Any]:
        """

            should_review_memory=_should_review_memory,
        )

    while (api_call_count < agent.max_iterations and agent.iteration_budget.remaining > 0) or agent._budget_grace_call:
        # Reset per-turn checkpoint dedup so each iteration can take one snapshot
        agent._checkpoint_mgr.new_turn()

        # Check for interrupt request (e.g., user sent new message)
        if agent._interrupt_requested:
            interrupted = True
            _turn_exit_reason = "interrupted_by_user"
            if not agent.quiet_mode:
                agent._safe_print("\n⚡ Breaking out of tool loop due to interrupt...")

"""AgentLoop: Multi-turn LLM streaming loop with tool execution and judge evaluation.

Implements AgentProtocol and runs a streaming event loop:
1. Calls LLMProvider.stream() to get streaming events
2. Processes text deltas, tool calls, and finish events
3. Executes tools and feeds results back to the conversation
4. Uses judge evaluation (or implicit stop-reason) to decide loop termination
5. Publishes lifecycle events to EventBus
6. Persists conversation and outputs via write-through to ConversationStore
"""

        )
        super().__init__(**kwargs)

    def forward(self, message: AgentMessage, session_id=0, **kwargs) -> AgentMessage:
        for _ in range(self.max_turn):
            message = self.select_agent(message, session_id=session_id, **kwargs)
            if self.finish_condition(message):
                return message
            message = self.actions(message, session_id=session_id)
        return message


class AsyncReAct(AsyncAgent):

            structured_response=structured_response,
        )

    def model_node(state: AgentState[Any], runtime: Runtime[ContextT]) -> list[Command[Any]]:
        """Sync model request handler with sequential middleware processing."""
        request = ModelRequest(
            model=model,
            tools=default_tools,
            system_message=system_message,
            response_format=initial_response_format,
            messages=state["messages"],
            tool_choice=None,
            state=state,

The sequence diagram with more details of the communication between components is below:

```mermaid


sequenceDiagram
    participant U as User
    participant S as Shields
    participant E as Executor
    participant L as LLM

        return tool_output

    @step
    async def init_run(self, ctx: Context, ev: AgentWorkflowStartEvent) -> AgentInput:
        """Sets up the workflow and validates inputs."""
        await self._init_context(ctx, ev)

        user_msg: Optional[Union[str, ChatMessage]] = ev.get("user_msg")
        chat_history: Optional[List[ChatMessage]] = ev.get("chat_history", [])

        # Convert string user_msg to ChatMessage
        if isinstance(user_msg, str):

    def tool_by_names(self) -> Dict[str, ToolInterface]:
        return {tool.name: tool for tool in self.tools}

    def run(self, question: str):
        previous_responses = []
        num_loops = 0
        prompt = self.prompt_template.format(
                today = datetime.date.today(),
                tool_description=self.tool_description,
                tool_names=self.tool_names,
                question=question,
                previous_responses='{previous_responses}'
        )

        self.logger.info(f"Registering Summary worker: '{name}'")
        register_worker(name, SUMMARY, worker_class)

    async def _evolution_cycle(self, iteration_id: int) -> None:
        """
        Represents a single, complete evolution cycle for a given iteration ID.
        """
        trace_id = str(uuid.uuid4().hex[:12])
        self.logger.info(
            f"Trace ID: {trace_id}, Starting evolution cycle for iteration {iteration_id}."
        )

        previous_prompt_tokens = self.total_prompt_tokens

task_exchanges = []
haiku_tasks = []

while True:
    # Call Orchestrator to break down the objective into the next sub-task or provide the final output
    previous_results = [result for _, result in task_exchanges]
    if not task_exchanges:
        # Pass the file content only in the first iteration if available
        opus_result, file_content_for_haiku, search_query = opus_orchestrator(objective, file_content, previous_results, use_search)
    else:
        opus_result, _, search_query = opus_orchestrator(objective, previous_results=previous_results, use_search=use_search)

    if "The task is complete:" in opus_result:

import type { LoopOptions, LoopRun, StreamInternal } from './types';
import { workflowLoopStream } from './workflows/stream';

export function loop<Tools extends ToolSet = ToolSet, OUTPUT = undefined>({
  resumeContext,
  models,
  logger,
  runId,
  idGenerator,
  messageList,
  includeRawChunks,
  modelSettings,
  tools,

            return
        self.rc.msg_buffer.push(message)

    async def _react(self) -> Message:
        """Think first, then act, until the Role _think it is time to stop and requires no more todo.
        This is the standard think-act loop in the ReAct paper, which alternates thinking and acting in task solving, i.e. _think -> _act -> _think -> _act -> ...
        Use llm to select actions in _think dynamically
        """
        actions_taken = 0
        rsp = AIMessage(content="No actions taken yet", cause_by=Action)  # will be overwritten after Role _act
        while actions_taken < self.rc.max_react_loop:
            # think
            has_todo = await self._think()

	// OnReceiveMessage is called when the agent receives a message.
	// Custom agents may implement this method to handle incoming messages.
	OnReceiveMessage(msgName string, data *string) (*string, error)
}

// The AgentBase struct provides default implementations for the Agent interface methods.
// Custom agents can embed this struct to avoid implementing all methods.
type AgentBase struct {
}

func (a *AgentBase) Id() string {
	return *activeAgentId

UNIFIED_SESSION_KEY = "unified:default"

class TurnState(Enum):
    RESTORE = auto()
    COMPACT = auto()
    COMMAND = auto()
    BUILD = auto()
    RUN = auto()
    SAVE = auto()
    RESPOND = auto()
    DONE = auto()

      // -----------------------------------------------------------------
      // Main agentic loop — `while (true)` until end_turn or maxTurns
      // -----------------------------------------------------------------
      while (true) {
        // Respect abort before each LLM call.
        if (effectiveAbortSignal?.aborted) {
          break
        }

        // Guard against unbounded loops.
        if (turns >= this.maxTurns) {
          break
        }

/**
 * Main loop logic shared by agentLoop and agentLoopContinue.
 */
async function runLoop(
  initialContext: AgentContext,
  newMessages: AgentMessage[],
  initialConfig: AgentLoopConfig,
  signal: AbortSignal | undefined,
  emit: AgentEventSink,
  streamFn?: StreamFn,
  runtime?: AgentCoreStreamRuntimeDeps,
): Promise<void> {
  let currentContext = initialContext;

        if self.__input_event:
            self.__input_event.set()

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        """Process a frame.

        Args:
            frame: The frame to process.
            direction: The direction of frame flow.
        """
        if self._observer:
            timestamp = self._clock.get_time() if self._clock else 0
            data = FrameProcessed(

        # Self-reflection (legacy OpenAI path)
        reflection_count = 0
        while True:
            reflection_prompt = f"""
Reflect on your previous response: '{response_text}'.
{self.reflect_prompt if self.reflect_prompt else "Identify any flaws, improvements, or actions."}
Provide a "satisfactory" status ('yes' or 'no').
Output MUST be JSON with 'reflection' and 'satisfactory'.
            """
            reflection_messages = messages + [
                {"role": "assistant", "content": response_text},
                {"role": "user", "content": reflection_prompt},

/// 4. Invoke each function and add the function result to the <see cref="ChatHistory"/>.
/// </summary>
[ExcludeFromCodeCoverage]
internal sealed class FunctionCallsProcessor
{
    /// <summary>
    /// The maximum number of auto-invokes that can be in-flight at any given time as part of the current
    /// asynchronous chain of execution.
    /// </summary>
    /// <remarks>
    /// This is a fail-safe mechanism. If someone accidentally manages to set up execution settings in such a way that
    /// auto-invocation is invoked recursively, and in particular where a prompt function is able to auto-invoke itself,
    /// we could end up in an infinite loop. This const is a backstop against that happening. We should never come close

    )


async def event_loop_cycle(
    agent: "Agent",
    invocation_state: dict[str, Any],
    structured_output_context: StructuredOutputContext | None = None,
    limits: Limits | None = None,
) -> AsyncGenerator[TypedEvent, None]:
    """Execute a single cycle of the event loop.

    This core function processes a single conversation turn, handling model inference, tool execution, and error
    recovery. It manages the entire lifecycle of a conversation turn, including:

        history = copy.deepcopy(messages)
        init_len = len(messages)

        while len(history) - init_len < max_turns and active_agent:

            # get completion with current history, agent
            completion = self.get_chat_completion(
                agent=active_agent,
                history=history,
                context_variables=context_variables,
                model_override=model_override,
                stream=stream,
                debug=debug,

  endToolBuildPerf({ toolCount: tools.length })

  const checkpointer = new MemorySaver()
  let agent = createReactAgent({
    llm,
    tools,
    prompt,
    checkpointer,
  })
  let pendingGraphMessages = [...langchainMessages]

  // -------------------------------------------------------------------------
  // Init turn state and limits

                self.save()
                self._autosave_config_step(loop_count=0)

            while (
                self.max_loops == "auto"
                or loop_count < self.max_loops
            ):
                loop_count += 1

                # Compress short-term memory if an auto-loop run has
                # crossed the configured fraction of the context window.
                if self._context_compressor is not None:
                    self._context_compressor.maybe_compress(self)

                self._finalize_step_result(step_result, context)


class ModelExecutionStep(Step):
    """Execute the model request."""
    
    @property
    def name(self) -> str:
        return "model_execution"
    
    @property
    def description(self) -> str:
        return "Execute model request"

                current_agent = self._get_current_agent()

                # Deal with workflow-based agent types (Function Calling and ReAct)
                if self._get_current_agent_type() in [
                    AgentType.FUNCTION_CALLING,
                    AgentType.REACT,
                ]:
                    from llama_index.core.workflow import Context

                    # Create context and pass memory to the workflow agent
                    # According to LlamaIndex docs, we should let the workflow manage memory internally
                    ctx = Context(current_agent)

                      // Default values
                      temperature: this.temperature,
                      maxOutputTokens: this.maxOutputTokens,
                      stopWhen: options?.stopWhen ?? this.stopWhen ?? stepCountIs(maxSteps),
                      // User overrides from AI SDK options
                      ...aiSDKOptions,
                      maxRetries: 0,
                      // Structured output if provided
                      output,
                      // Provider-specific options
                      providerOptions,
                      // VoltAgent controlled (these should not be overridden)
                      abortSignal: oc.abortController.signal,