Spaces:
Running
Running
| """ | |
| 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.""" | |
| 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 | |
| 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) | |
| 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) | |
| 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 |