Confucius Code Agent

The WorkingMemory is a dataclass that maintains hierarchical context for the agent.

from dataclasses import dataclass, field

@dataclass
class WorkingMemory:
    """Hierarchical context for clean AX (Agent Experience)."""
    
    goal: str                           # What we're trying to achieve
    plan: list[str] = field(default_factory=list)  # High-level steps
    key_facts: list[str] = field(default_factory=list)  # Important discoveries
    recent: list[str] = field(default_factory=list)  # Last N observations
    compressed_history: str = ""        # Summarized older context
    
    def maybe_compress(self, turns: list, max_recent: int = 5):
        """Compress older entries to prevent context overflow."""
        if len(self.recent) > max_recent:
            old = self.recent[:-max_recent]
            self.compressed_history += "\n".join(old)
            self.recent = self.recent[-max_recent:]

Tools are modular functions the agent can invoke to interact with the codebase.

class Tools:
    """Modular tool implementations (DX principle)."""
    
    def __init__(self, repo_root: Path):
        self.repo_root = repo_root
    
    def search_files(self, pattern: str) -> str:
        """Search for pattern in all Python files."""
        matches = []
        for path in self.repo_root.rglob("*.py"):
            content = path.read_text()
            if pattern in content:
                matches.append(str(path.relative_to(self.repo_root)))
        return "\n".join(matches) if matches else "No matches found"
    
    def read_file(self, path: str) -> str:
        """Read file contents."""
        return (self.repo_root / path).read_text()
    
    def write_file(self, path: str, content: str) -> str:
        """Write content to file."""
        (self.repo_root / path).write_text(content)
        return f"Wrote {len(content)} chars to {path}"
    
    def run_tests(self) -> str:
        """Execute tests and return results."""
        result = subprocess.run(
            ["python", "-m", "pytest", "-q"],
            cwd=self.repo_root,
            capture_output=True, text=True
        )
        return result.stdout + result.stderr

The Orchestrator runs the Think → Act → Observe loop and manages working memory.

class Orchestrator:
    """Core loop: Think → Act → Observe."""
    
    def __init__(self, llm, tools: Tools, wm: WorkingMemory):
        self.llm = llm
        self.tools = tools
        self.wm = wm
        self.turns = []
    
    def run(self, task: str, max_steps: int = 8) -> bool:
        """Execute agent loop until success or max steps."""
        last_obs = ""
        
        for step in range(max_steps):
            # Build prompt from working memory
            prompt = self._build_prompt(task) + f"\nLAST_OBS: {last_obs}"
            
            # Think: LLM generates thought + action
            response = self.llm.generate(prompt)
            thought, action = self._parse_action(response)
            
            # Act: Execute tool
            if action:
                tool_name, args = action
                result = self._invoke_tool(tool_name, args)
                last_obs = result.output
                
                # Update memory
                self.wm.recent.append(f"[{step}] {tool_name} → {last_obs[:200]}")
                self.wm.maybe_compress(self.turns)
                
                # Check success
                if "passed" in last_obs.lower():
                    return True
        
        return False

After each run, generate Markdown notes with trajectory and hindsight for user transparency.

def write_notes(self, title: str) -> Path:
    """Generate Markdown notes for UX transparency."""
    notes_dir = self.run_dir / "notes"
    notes_dir.mkdir(parents=True, exist_ok=True)
    
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    path = notes_dir / f"{timestamp}-{title}.md"
    
    content = f"""# {title}
Date: {datetime.now().isoformat()}

## Goal
{self.wm.goal}

## Plan
{chr(10).join('- ' + p for p in self.wm.plan)}

## Key Events (Trajectory)
{chr(10).join('- **' + t.action.tool + '** → ' + ('OK' if t.action.ok else 'ERR') for t in self.turns)}

## Hindsight (Failures / Lessons)
{chr(10).join('- ' + f for f in self.wm.key_facts if 'FAILED' in f.upper())}
"""
    path.write_text(content)
    return path