File size: 1,979 Bytes
2862b00 c3b49d6 2862b00 c3b49d6 2862b00 c3b49d6 2862b00 c3b49d6 2862b00 c3b49d6 2862b00 c3b49d6 2862b00 c3b49d6 2862b00 c3b49d6 2862b00 c3b49d6 2862b00 c3b49d6 2862b00 c3b49d6 2862b00 c3b49d6 | 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 | """
Plan Manager for CodeAct Agent.
Handles plan creation, updates, and progress tracking.
"""
import re
class PlanManager:
"""Manages plan creation, updates, and progress tracking."""
@staticmethod
def extract_plan_from_content(content: str) -> str | None:
"""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 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
|