335 lines
12 KiB
Python
335 lines
12 KiB
Python
"""Agentic loop — multi-step task execution with progress tracking.
|
|
|
|
Provides `run_agentic_loop()`, the engine behind the `plan_and_execute` tool.
|
|
When the model recognises a task needs 3+ sequential steps, it calls
|
|
`plan_and_execute(task)` which spawns this loop in the background.
|
|
|
|
Flow:
|
|
1. Planning — ask the model to break the task into numbered steps
|
|
2. Execution — run each step sequentially, feeding results forward
|
|
3. Adaptation — on failure, ask the model to adapt the plan
|
|
4. Summary — ask the model to summarise what was accomplished
|
|
|
|
Progress is broadcast via WebSocket so the dashboard can show live updates.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import re
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data structures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class AgenticStep:
|
|
"""Result of a single step in the agentic loop."""
|
|
|
|
step_num: int
|
|
description: str
|
|
result: str
|
|
status: str # "completed" | "failed" | "adapted"
|
|
duration_ms: int
|
|
|
|
|
|
@dataclass
|
|
class AgenticResult:
|
|
"""Final result of the entire agentic loop."""
|
|
|
|
task_id: str
|
|
task: str
|
|
summary: str
|
|
steps: list[AgenticStep] = field(default_factory=list)
|
|
status: str = "completed" # "completed" | "partial" | "failed"
|
|
total_duration_ms: int = 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Agent factory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_loop_agent = None
|
|
_loop_agent_lock = threading.Lock()
|
|
|
|
|
|
def _get_loop_agent():
|
|
"""Create a fresh agent for the agentic loop.
|
|
|
|
Returns the same type of agent as `create_timmy()` but with a
|
|
dedicated session so it doesn't pollute the main chat history.
|
|
"""
|
|
global _loop_agent
|
|
if _loop_agent is None:
|
|
with _loop_agent_lock:
|
|
if _loop_agent is None:
|
|
from timmy.agent import create_timmy
|
|
|
|
_loop_agent = create_timmy()
|
|
return _loop_agent
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plan parser
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_STEP_RE = re.compile(r"^\s*(\d+)[.)]\s*(.+)$", re.MULTILINE)
|
|
|
|
|
|
def _parse_steps(plan_text: str) -> list[str]:
|
|
"""Extract numbered steps from the model's planning output."""
|
|
matches = _STEP_RE.findall(plan_text)
|
|
if matches:
|
|
return [desc.strip() for _, desc in matches]
|
|
# Fallback: split on newlines, ignore blanks
|
|
return [line.strip() for line in plan_text.strip().splitlines() if line.strip()]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Core loop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def run_agentic_loop(
|
|
task: str,
|
|
*,
|
|
session_id: str = "agentic",
|
|
max_steps: int = 0,
|
|
on_progress: Callable | None = None,
|
|
) -> AgenticResult:
|
|
"""Execute a multi-step task with planning, execution, and adaptation.
|
|
|
|
Args:
|
|
task: Full description of the task to execute.
|
|
session_id: Agno session_id for conversation continuity.
|
|
max_steps: Max steps to execute (0 = use config default).
|
|
on_progress: Optional async callback(description, step_num, total_steps).
|
|
|
|
Returns:
|
|
AgenticResult with steps, summary, and status.
|
|
"""
|
|
from config import settings
|
|
|
|
if max_steps <= 0:
|
|
max_steps = getattr(settings, "max_agent_steps", 10)
|
|
|
|
task_id = str(uuid.uuid4())[:8]
|
|
start_time = time.monotonic()
|
|
|
|
agent = _get_loop_agent()
|
|
result = AgenticResult(task_id=task_id, task=task, summary="")
|
|
|
|
# ── Phase 1: Planning ──────────────────────────────────────────────────
|
|
plan_prompt = (
|
|
f"Break this task into numbered steps (max {max_steps}). "
|
|
f"Return ONLY a numbered list, nothing else.\n\n"
|
|
f"Task: {task}"
|
|
)
|
|
try:
|
|
plan_run = await asyncio.to_thread(
|
|
agent.run, plan_prompt, stream=False, session_id=f"{session_id}_plan"
|
|
)
|
|
plan_text = plan_run.content if hasattr(plan_run, "content") else str(plan_run)
|
|
except Exception as exc: # broad catch intentional: agent.run can raise any error
|
|
logger.error("Agentic loop: planning failed: %s", exc)
|
|
result.status = "failed"
|
|
result.summary = f"Planning failed: {exc}"
|
|
result.total_duration_ms = int((time.monotonic() - start_time) * 1000)
|
|
return result
|
|
|
|
steps = _parse_steps(plan_text)
|
|
if not steps:
|
|
result.status = "failed"
|
|
result.summary = "Planning produced no steps."
|
|
result.total_duration_ms = int((time.monotonic() - start_time) * 1000)
|
|
return result
|
|
|
|
# Enforce max_steps — track if we truncated
|
|
planned_steps = len(steps)
|
|
steps = steps[:max_steps]
|
|
total_steps = len(steps)
|
|
was_truncated = planned_steps > total_steps
|
|
|
|
# Broadcast plan
|
|
await _broadcast_progress(
|
|
"agentic.plan_ready",
|
|
{
|
|
"task_id": task_id,
|
|
"task": task,
|
|
"steps": steps,
|
|
"total": total_steps,
|
|
},
|
|
)
|
|
|
|
# ── Phase 2: Execution ─────────────────────────────────────────────────
|
|
completed_results: list[str] = []
|
|
|
|
for i, step_desc in enumerate(steps, 1):
|
|
step_start = time.monotonic()
|
|
|
|
recent = completed_results[-2:] if completed_results else []
|
|
context = (
|
|
f"Task: {task}\n"
|
|
f"Step {i}/{total_steps}: {step_desc}\n"
|
|
f"Recent progress: {recent}\n\n"
|
|
f"Execute this step and report what you did."
|
|
)
|
|
|
|
try:
|
|
step_run = await asyncio.to_thread(
|
|
agent.run, context, stream=False, session_id=f"{session_id}_step{i}"
|
|
)
|
|
step_result = step_run.content if hasattr(step_run, "content") else str(step_run)
|
|
|
|
# Clean the response
|
|
from timmy.session import _clean_response
|
|
|
|
step_result = _clean_response(step_result)
|
|
|
|
step = AgenticStep(
|
|
step_num=i,
|
|
description=step_desc,
|
|
result=step_result,
|
|
status="completed",
|
|
duration_ms=int((time.monotonic() - step_start) * 1000),
|
|
)
|
|
result.steps.append(step)
|
|
completed_results.append(f"Step {i}: {step_result[:200]}")
|
|
|
|
# Broadcast progress
|
|
await _broadcast_progress(
|
|
"agentic.step_complete",
|
|
{
|
|
"task_id": task_id,
|
|
"step": i,
|
|
"total": total_steps,
|
|
"description": step_desc,
|
|
"result": step_result[:200],
|
|
},
|
|
)
|
|
|
|
if on_progress:
|
|
await on_progress(step_desc, i, total_steps)
|
|
|
|
except Exception as exc: # broad catch intentional: agent.run can raise any error
|
|
logger.warning("Agentic loop step %d failed: %s", i, exc)
|
|
|
|
# ── Adaptation: ask model to adapt ─────────────────────────────
|
|
adapt_prompt = (
|
|
f"Step {i} failed with error: {exc}\n"
|
|
f"Original step was: {step_desc}\n"
|
|
f"Adapt the plan and try an alternative approach for this step."
|
|
)
|
|
try:
|
|
adapt_run = await asyncio.to_thread(
|
|
agent.run,
|
|
adapt_prompt,
|
|
stream=False,
|
|
session_id=f"{session_id}_adapt{i}",
|
|
)
|
|
adapt_result = (
|
|
adapt_run.content if hasattr(adapt_run, "content") else str(adapt_run)
|
|
)
|
|
from timmy.session import _clean_response
|
|
|
|
adapt_result = _clean_response(adapt_result)
|
|
|
|
step = AgenticStep(
|
|
step_num=i,
|
|
description=f"[Adapted] {step_desc}",
|
|
result=adapt_result,
|
|
status="adapted",
|
|
duration_ms=int((time.monotonic() - step_start) * 1000),
|
|
)
|
|
result.steps.append(step)
|
|
completed_results.append(f"Step {i} (adapted): {adapt_result[:200]}")
|
|
|
|
await _broadcast_progress(
|
|
"agentic.step_adapted",
|
|
{
|
|
"task_id": task_id,
|
|
"step": i,
|
|
"total": total_steps,
|
|
"description": step_desc,
|
|
"error": str(exc),
|
|
"adaptation": adapt_result[:200],
|
|
},
|
|
)
|
|
|
|
if on_progress:
|
|
await on_progress(f"[Adapted] {step_desc}", i, total_steps)
|
|
|
|
except Exception as adapt_exc: # broad catch intentional: agent.run can raise any error
|
|
logger.error("Agentic loop adaptation also failed: %s", adapt_exc)
|
|
step = AgenticStep(
|
|
step_num=i,
|
|
description=step_desc,
|
|
result=f"Failed: {exc}; Adaptation also failed: {adapt_exc}",
|
|
status="failed",
|
|
duration_ms=int((time.monotonic() - step_start) * 1000),
|
|
)
|
|
result.steps.append(step)
|
|
completed_results.append(f"Step {i}: FAILED")
|
|
|
|
# ── Phase 3: Summary ───────────────────────────────────────────────────
|
|
completed_count = sum(1 for s in result.steps if s.status == "completed")
|
|
adapted_count = sum(1 for s in result.steps if s.status == "adapted")
|
|
failed_count = sum(1 for s in result.steps if s.status == "failed")
|
|
parts = [f"Completed {completed_count}/{total_steps} steps"]
|
|
if adapted_count:
|
|
parts.append(f"{adapted_count} adapted")
|
|
if failed_count:
|
|
parts.append(f"{failed_count} failed")
|
|
result.summary = f"{task}: {', '.join(parts)}."
|
|
|
|
# Determine final status
|
|
if was_truncated:
|
|
result.status = "partial"
|
|
elif len(result.steps) < total_steps:
|
|
result.status = "partial"
|
|
elif any(s.status == "failed" for s in result.steps):
|
|
result.status = "partial"
|
|
else:
|
|
result.status = "completed"
|
|
|
|
result.total_duration_ms = int((time.monotonic() - start_time) * 1000)
|
|
|
|
await _broadcast_progress(
|
|
"agentic.task_complete",
|
|
{
|
|
"task_id": task_id,
|
|
"status": result.status,
|
|
"steps_completed": len(result.steps),
|
|
"summary": result.summary[:300],
|
|
"duration_ms": result.total_duration_ms,
|
|
},
|
|
)
|
|
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WebSocket broadcast helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _broadcast_progress(event: str, data: dict) -> None:
|
|
"""Broadcast agentic loop progress via WebSocket (best-effort)."""
|
|
try:
|
|
from infrastructure.ws_manager.handler import ws_manager
|
|
|
|
await ws_manager.broadcast(event, data)
|
|
except (ImportError, AttributeError, ConnectionError, RuntimeError) as exc:
|
|
logger.warning("Agentic loop broadcast failed: %s", exc)
|
|
logger.debug("Agentic loop: WS broadcast failed for %s", event)
|