File size: 2,756 Bytes
f580ad3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 |
"""
Plan Manager for CodeAct Agent.
Handles plan creation, updates, and progress tracking.
"""
import re
from typing import Optional, Dict
class PlanManager:
"""Manages plan creation, updates, and progress tracking."""
@staticmethod
def extract_plan_from_content(content: str) -> Optional[str]:
"""Extract plan from agent content."""
plan_pattern = r'\d+\.\s*\[[^\]]*\]\s*[^\n]+(?:\n\d+\.\s*\[[^\]]*\]\s*[^\n]+)*'
matches = re.findall(plan_pattern, content)
# Return the last (most recent) plan if multiple plans exist
return matches[-1] if matches else None
@staticmethod
def update_plan_for_solution(plan_text: str) -> str:
"""Update plan to mark all remaining steps as completed when providing final solution."""
if not plan_text:
return plan_text
lines = plan_text.split('\n')
updated_lines = []
for line in lines:
# Mark any unchecked or failed steps as completed since we're providing final solution
if '[ ]' in line or '[✗]' in line:
updated_line = re.sub(r'\[\s*[^\]]*\]', '[✓]', line)
updated_lines.append(updated_line)
else:
updated_lines.append(line)
return '\n'.join(updated_lines)
@staticmethod
def mark_step_completed(plan_text: str, step_description: str) -> str:
"""Mark a specific step as completed based on its description."""
if not plan_text or not step_description:
return plan_text
lines = plan_text.split('\n')
updated_lines = []
for line in lines:
# Check if this line contains the step description and is unchecked
if step_description.lower() in line.lower() and ('[ ]' in line or '[✗]' in line):
updated_line = re.sub(r'\[\s*[^\]]*\]', '[✓]', line)
updated_lines.append(updated_line)
else:
updated_lines.append(line)
return '\n'.join(updated_lines)
@staticmethod
def get_plan_progress(plan_text: str) -> Dict[str, int]:
"""Get plan progress statistics."""
if not plan_text:
return {"total": 0, "completed": 0, "pending": 0, "failed": 0}
lines = plan_text.split('\n')
stats = {"total": 0, "completed": 0, "pending": 0, "failed": 0}
for line in lines:
if re.search(r'\d+\.\s*\[', line):
stats["total"] += 1
if '[✓]' in line:
stats["completed"] += 1
elif '[ ]' in line:
stats["pending"] += 1
elif '[✗]' in line:
stats["failed"] += 1
return stats |