规划与任务分解

#       steps:
#         - parallel: [skill-a, skill-b]
#         - skill: skill-c, consume: [skill-a, skill-b]
chains:
  # daily-routine:
  #   schedule: "0 7 * * *"
  #   on_error: fail-fast
  #   steps:
  #     - parallel: [token-movers, paper-pick, github-issues, hacker-news-digest]
  #     - skill: daily-routine, consume: [token-movers, paper-pick, github-issues, hacker-news-digest]

# Default model for all skills. Override per-run via workflow_dispatch.
# Options: claude-opus-4-8, claude-sonnet-4-6, claude-haiku-4-5-20251001

        MAX_BATCH_TOKENS = 100000

        # Calculate base prompt template tokens (without the batch content)
        base_selection_prompt = f"""You are an expert assistant that selects relevant commands/tools for a user request. You manage IT infrastructure, remote machines, tickets, desktop support, code execution, web browsing, media generation, and more. Your job is to select exactly the right set of commands the agent will need.

## User's Request
{user_input}

## Context
{context}

{{BATCH_COMMANDS}}

   * @param topic The research topic
   * @returns The research answer
   */
  async runResearch(topic: string) {
    console.log(`\x1b[36m🔍 Researching topic: ${topic}\x1b[0m`);

    // Step 1: Generate initial queries
    const initialQueries = await this.generateInitialQueries({ topic });

    // Step 2: Perform initial search
    const initialResults = await this.performSearch({
      queries: initialQueries,
    });

    from agentverse.message import Message


@decision_maker_registry.register("dynamic")
class DynamicDecisionMaker(BaseDecisionMaker):
    """
    Discuss in a horizontal manner.
    """

    name: str = "dynamic"

    ## To Do: implement dynamic
    # def step(

        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

            await self._output_message_queue.put(msg)

        # Outer Loop for first time
        # Create the initial task ledger
        #################################
        # Combine all message contents for task
        self._task = " ".join([msg.to_model_text() for msg in message.messages])
        planning_conversation: List[LLMMessage] = []

        # 1. GATHER FACTS
        # create a closed book task and generate a response and update the chat history
        planning_conversation.append(
            UserMessage(content=self._get_task_ledger_facts_prompt(self._task), source=self._name)

1. After user input: Acknowledge request (show we understood)
2. After tool execution: Reflect on results

Planning is left to the main agent - this plugin just adds intent recognition and reflection.

Trace kinds for frontend rendering:
- kind='intent' → Show as "Understanding..." card
- kind='reflect' → Show as "Reflecting..." card

Usage:
    from connectonion import Agent
    from connectonion.useful_plugins import re_act

            )
        return task_outputs

    def _handle_crew_planning(self) -> None:
        """Handles the Crew planning."""
        self._logger.log("info", "Planning the crew execution")
        result = CrewPlanner(
            tasks=self.tasks, planning_agent_llm=self.planning_llm
        )._handle_crew_planning()

        plan_map: dict[int, str] = {}
        for step_plan in result.list_of_plans_per_task:
            if step_plan.task_number in plan_map:

_queen_tools_independent = """
# Tools

## Planning — use FIRST for multi-step work
- task_create_batch — When a request has 2+ atomic steps, your FIRST \
tool call is `task_create_batch` with one entry per step (atomic, \
one round-trip).
- task_create — One-off mid-run additions when you discover \
unplanned work AFTER the initial plan is laid out.
- task_update / task_list / task_get — Mark progress, inspect, or \
re-read state.

See "Independent execution" for the per-step flow and granularity rule.

    """The current status of the todo item."""


class PlanningState(AgentState[ResponseT]):
    """State schema for the todo middleware.

    Type Parameters:
        ResponseT: The type of the structured response. Defaults to `Any`.
    """

    todos: Annotated[NotRequired[list[Todo]], OmitFromInput]
    """List of todo items for tracking task progress."""

from llama_index.core.tools import BaseTool, FunctionTool
from llama_index.core.workflow import Context

DEFAULT_CODE_ACT_PROMPT = """You are a helpful AI assistant that can write and execute Python code to solve problems.

You will be given a task to perform. You should output:
- Python code wrapped in <execute>...</execute> tags that provides the solution to the task, or a step towards the solution. Any output you want to extract from the code should be printed to the console.
- Text to be shown directly to the user, if you want to ask for more information or provide the final answer.
- If the previous code execution can be used to respond to the user, then respond directly (typically you want to avoid mentioning anything related to the code execution in your response).

## Response Format:
Example of proper code format:
<execute>

from loongflow.framework.pes.register import get_worker


class Planner:
    """Planner Class"""

    def __init__(self, planner_name: str, config: Any, db: Any):
        self.planner = get_worker(
            name=planner_name, phase="planner", config=config, db=db
        )

    async def run(self, context: Any, message: Message | None) -> Message:
        """

# Initialize the Rich Console
console = Console()

def opus_orchestrator(objective, file_content=None, previous_results=None, use_search=False):
    console.print(f"\n[bold]Calling Orchestrator for your objective[/bold]")
    previous_results_text = "\n".join(previous_results) if previous_results else "None"
    if file_content:
        console.print(Panel(f"File content:\n{file_content}", title="[bold blue]File Content[/bold blue]", title_align="left", border_style="blue"))
    
    messages = [
        {
            "role": "user",
            "content": [

   * The step's inputSchema must be satisfied by the previous step's output (or workflow input for first step).
   * This means: TPrevSchema must be assignable to TStepInput
   */
  then<TStepId extends string, TStepState, TStepInput, TSchemaOut>(
    step: Step<
      TStepId,
      // Allow steps with any/unknown state, or steps whose state is a subset of workflow state
      unknown extends TStepState ? TStepState : SubsetOf<TStepState, TState>,
      // Check: previous output (TPrevSchema) must satisfy step's input requirements (TStepInput)
      // If TPrevSchema can be assigned to TStepInput, allow it. Otherwise show expected type.
      TPrevSchema extends TStepInput ? TStepInput : TPrevSchema,
      TSchemaOut,
      any,

    def current_task_id(self):
        return self.plan.current_task_id

    async def update_plan(self, goal: str = "", max_tasks: int = 3, max_retries: int = 3):
        if goal:
            self.plan = Plan(goal=goal)

        plan_confirmed = False
        while not plan_confirmed:
            context = self.get_useful_memories()
            rsp = await WritePlan().run(context, max_tasks=max_tasks)
            self.working_memory.add(Message(content=rsp, role="assistant", cause_by=WritePlan))

    return _session_goal_raw(metadata)


def sustained_goal_active(metadata: Mapping[str, Any] | None) -> bool:
    """True when this session has an active sustained objective (``long_task`` bookkeeping)."""
    goal = parse_goal_state(goal_state_raw(metadata))
    return isinstance(goal, dict) and goal.get("status") == "active"


def sustained_goal_turn(
    metadata: Mapping[str, Any] | None,
    *,
    message_metadata: Mapping[str, Any] | None = None,

    // ------------------------------------------------------------------
    // Step 2: Parse tasks from coordinator output
    // ------------------------------------------------------------------
    const taskSpecs = parseTaskSpecs(decompositionResult.output)

    const queue = new TaskQueue()
    const scheduler = new Scheduler('dependency-first')
    const taskMetrics = new Map<string, TaskExecutionMetrics>()

    if (taskSpecs && taskSpecs.length > 0) {
      // Map title-based dependsOn references to real task IDs so we can
      // build the dependency graph before adding tasks to the queue.
      this.loadSpecsIntoQueue(taskSpecs, agentConfigs, queue)

      const nextTurnSnapshot = await config.prepareNextTurn?.(nextTurnContext);
      if (nextTurnSnapshot) {
        currentContext = nextTurnSnapshot.context ?? currentContext;
        config = Object.assign({}, config, {
          model: nextTurnSnapshot.model ?? config.model,
          reasoning:
            nextTurnSnapshot.thinkingLevel === undefined
              ? config.reasoning
              : nextTurnSnapshot.thinkingLevel === "off"
                ? undefined
                : nextTurnSnapshot.thinkingLevel,
        });
      }

                await self._downstream_push_frame(frame, direction)


class Pipeline(BasePipeline):
    """Main pipeline implementation that connects frame processors in sequence.

    Creates a linear chain of frame processors with automatic source and sink
    processors for external frame handling. Manages processor lifecycle and
    provides metrics collection from contained processors.
    """

    def __init__(
        self,

from .plan import Plan, PlanStep

# Define READ_ONLY_TOOLS locally to avoid circular import
READ_ONLY_TOOLS = [
    "read_file",
    "list_directory",
    "search_codebase",
    "search_files",
    "grep_search",
    "find_files",
    "web_search",
    "get_file_content",
    "list_files",

    /// Builds the process.
    /// </summary>
    /// <returns>An instance of <see cref="KernelProcess"/></returns>
    public KernelProcess Build(KernelProcessStateMetadata? stateMetadata = null)
    {
        // Build the edges first
        var builtEdges = this.Edges.ToDictionary(kvp => kvp.Key, kvp => kvp.Value.Select(e => e.Build()).ToList());

        // Build the steps and injecting initial state if any is provided
        var builtSteps = this.BuildWithStateMetadata(stateMetadata);

        // Create the process
        KernelProcessState state = new(this.Name, version: this.Version, id: this.Id);

            raise ValueError("Session persistence is not supported for Graph agents yet.")


class GraphBuilder:
    """Builder pattern for constructing graphs."""

    def __init__(self) -> None:
        """Initialize GraphBuilder with empty collections."""
        self.nodes: dict[str, GraphNode] = {}
        self.edges: set[GraphEdge] = set()
        self.entry_points: set[GraphNode] = set()

        # Configuration options

type MainLoopObjectiveSource = 'goal' | 'working_state' | 'run_context' | 'legacy_tag'

export interface MainLoopState {
  goal: string | null
  goalContract: GoalContract | null
  objectiveSource: MainLoopObjectiveSource | null
  summary: string | null
  nextAction: string | null
  planSteps: string[]
  completedPlanSteps: string[]
  currentPlanStep: string | null
  reviewNote: string | null

            role=self.agent_name, content=message
        )

    def plan(self, task: str, *args, **kwargs) -> None:
        """
        Create a strategic plan for executing the given task.

        This method generates a step-by-step plan by combining the conversation
        history, planning prompt, and current task. The plan is then added to
        the agent's short-term memory for reference during execution.

        Args:
            task (str): The task to create a plan for

            if is_async_storage_backend(self.storage):
                await self.storage.aupsert_session(session, deserialize=True)
            else:
                self.storage.upsert_session(session, deserialize=True)

            if self.debug:
                status_str = "completed" if is_completed else "incomplete"
                info_log(
                    f"Session saved for run {output.run_id} ({status_str})",
                    "AgentSessionMemory",
                )
        except Exception as e:
            if self.debug:

)


class SubQuestionQueryWorkflow(Workflow):
    """
    Workflow for sub-question query engine.
    """

    # Workflow inputs/outputs
    class InputsModel(BaseModel):
        """
        Inputs for the workflow.
        """

const DEFAULT_GENERAL_PURPOSE_DESCRIPTION =
  "General-purpose agent for researching complex questions, searching for files and content, and executing multi-step tasks. When you are searching for a keyword or file and are not confident that you will find the right match in the first few tries use this agent to perform the search for you. This agent has access to all tools as the main agent.";

const PLANNING_SYSTEM_MESSAGE = [
  "<planagent_planning>",
  "Use write_todos to plan any multi-step task or any tool usage.",
  "If the user asks for a plan, you must call write_todos.",
  "Before using any tool other than write_todos, call write_todos with a short plan.",
  "Keep exactly one todo in_progress at a time and update the list as you go.",
  "Only mark todos done after completing the work.",
  "</planagent_planning>",
].join("\n");