File size: 2,937 Bytes
0e3d4b8
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
79
80
81
82
83
"""Coder Agent — writes code, creates files, debugs.

Handles steps that involve writing code, creating files, or fixing bugs.
Can use the write_file and shell_exec tools.
"""

from __future__ import annotations

import logging
import re
from typing import Any

from .agent_base import BaseAgent
from ..memory.goal_memory import Goal

logger = logging.getLogger(__name__)


class CoderAgent(BaseAgent):
    """Writes code and creates files for goal steps."""

    def __init__(self, goal_memory, persistent_memory=None, generate_fn=None):
        super().__init__(
            name="coder",
            role="Code Writer",
            description="Writes code, creates files, and debugs issues",
            goal_memory=goal_memory,
            persistent_memory=persistent_memory,
            generate_fn=generate_fn,
            poll_interval_s=2.0,
        )

    def _can_handle(self, goal: Goal) -> bool:
        """Coder handles goals related to code."""
        keywords = ["code", "function", "file", "implement", "write", "create", "build",
                    "debug", "fix", "bug", "program", "script", "module", "class"]
        text = (goal.title + " " + goal.description).lower()
        return any(kw in text for kw in keywords)

    def process_goal(self, goal: Goal) -> dict[str, Any]:
        """Execute a coding step."""
        if goal.current_step >= len(goal.steps):
            return {"success": True, "output": "No more steps"}

        step = goal.steps[goal.current_step]
        prompt = (
            f"You are a code writer agent. Execute this step:\n"
            f"Goal: {goal.title}\n"
            f"Step: {step['title']}\n"
            f"Description: {step['description']}\n"
            f"Write the code or create the file needed. Be concise.\n"
        )

        response = self._generate(prompt)

        # Extract file content from response
        files_created = self._extract_and_save_files(response)

        if files_created:
            return {"success": True, "output": f"Created files: {', '.join(files_created)}"}
        return {"success": True, "output": response[:200]}

    def _extract_and_save_files(self, text: str) -> list[str]:
        """Extract file blocks from LLM output and save them."""
        files = []
        # Pattern: ```file: path\n content ```
        pattern = r"```file:\s*(.+?)\n(.*?)```"
        matches = re.findall(pattern, text, re.DOTALL)

        for path, content in matches:
            path = path.strip()
            try:
                import os
                os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
                with open(path, "w", encoding="utf-8") as f:
                    f.write(content.strip())
                files.append(path)
                logger.info("Coder created file: %s", path)
            except Exception as e:
                logger.error("Failed to create file %s: %s", path, e)

        return files