kodeagent.orchestrator#

Planner and Observer for Agent orchestration.

Observer

Monitors an agent's behavior to detect issues like loops or stalled plans.

Planner

Given a task, generate and maintain a step-by-step plan to solve it.

Planner and Observer for Agent orchestration.

class kodeagent.orchestrator.Observer(model_name: str, tool_names: set[str], litellm_params: dict | None = None, threshold: int | None = 3, max_retries: int = 3, usage_tracker: UsageTracker | None = None, tracer_manager: AbstractTracerManager | None = None)[source]#

Bases: object

Monitors an agent’s behavior to detect issues like loops or stalled plans.

Create an Observer for an agent.

Parameters:

  • model_name – The LLM to use.

  • tool_names – The set of tools available to the agent.

  • litellm_params – LiteLLM parameters.

  • threshold – Observation threshold, i.e., how frequently the observer will analyze the chat history.

  • max_retries – Maximum number of retries for LLM calls.

  • usage_tracker – Optional UsageTracker instance to record usage.

  • tracer_manager – Optional AbstractTracerManager for hierarchical tracing.

async observe(iteration: int, task: Task, history: str, plan_before: str | AgentPlan | None, plan_after: str | AgentPlan | None, parent_trace: AbstractObservation | None = None) → str | None[source]#

Observe the agent’s state and return a corrective message if a problem is detected.

Parameters:

  • iteration – The current iteration of the agent.

  • task – The task being solved by the agent.

  • history – Task progress history (LLM chat history).

  • plan_before – The agent’s plan before this iteration.

  • plan_after – The updated plan.

  • parent_trace – Optional parent observation for hierarchical tracing.

Returns:

Optional correction message for the agent (LLM), e.g., what to do or avoid.

reset()[source]#

Reset the observer state.

class kodeagent.orchestrator.Planner(model_name: str, litellm_params: dict | None = None, max_retries: int = 3, usage_tracker: UsageTracker | None = None, tracer_manager: AbstractTracerManager | None = None)[source]#

Bases: object

Given a task, generate and maintain a step-by-step plan to solve it.

Create a planner using the given model.

Parameters:

  • model_name – The name of the LLM to use.

  • litellm_params – LiteLLM parameters.

  • max_retries – Maximum number of retries for LLM calls.

  • usage_tracker – Optional UsageTracker instance to record usage.

  • tracer_manager – Optional AbstractTracerManager for hierarchical tracing.

async create_plan(task: Task, agent_type: str, parent_trace: AbstractObservation | None = None) → AgentPlan[source]#

Create a plan to solve the given task and store it.

Parameters:

  • task – The task to solve.

  • agent_type – Type of the agent that would solve the task.

  • parent_trace – Optional parent observation for hierarchical tracing.

Returns:

A plan to solve the task.

get_formatted_plan(scope: Literal['all', 'done', 'pending'] = 'all') → str[source]#

Convert the agent’s plan into a Markdown checklist.

get_steps_done() → list[PlanStep][source]#

Returns the completed steps from the current plan.

Returns:

A list of completed PlanStep objects.

get_steps_pending() → list[PlanStep][source]#

Returns the pending steps from the current plan.

Returns:

A list of pending PlanStep objects.

reset()[source]#

Reset the planner state.

async update_plan(thought: str, observation: str, task_id: str, parent_trace: AbstractObservation | None = None)[source]#

Update the plan based on the last thought and observation.

Parameters:

  • thought – The ReAct/CodeAct agent’s thought.

  • observation – The agent’s observation.

  • task_id – ID of the task for which the plan is to be updated.

  • parent_trace – Optional parent observation for hierarchical tracing.