Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
| """Claude Code CLI client — invokes `claude -p` as a subprocess.""" | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| import shutil | |
| import subprocess | |
| from src.llm.client import LLMClient, load_config | |
| logger = logging.getLogger(__name__) | |
| class ClaudeCLIClient(LLMClient): | |
| """LLM client that shells out to the Claude Code CLI. | |
| Uses `claude -p` (print mode) with JSON output. Expert knowledge is | |
| injected via --system-prompt so it benefits from prompt caching across | |
| calls that share the same knowledge file. | |
| """ | |
| def __init__(self, config: dict | None = None): | |
| self.config = config or load_config() | |
| llm_cfg = self.config["llm"] | |
| self.model = llm_cfg.get("model", "sonnet") | |
| self.max_tokens = llm_cfg.get("max_tokens", 4096) | |
| self.claude_bin = llm_cfg.get("claude_bin", "claude") | |
| resolved = shutil.which(self.claude_bin) | |
| if resolved is None: | |
| raise FileNotFoundError( | |
| f"Claude Code CLI not found at '{self.claude_bin}'. " | |
| "Install it or set llm.claude_bin in config.yaml." | |
| ) | |
| self.claude_bin = resolved | |
| def call( | |
| self, | |
| prompt_template: str, | |
| variables: dict, | |
| expert_knowledge: str = "", | |
| seed: int | None = None, | |
| ) -> str: | |
| if seed is not None: | |
| raise NotImplementedError( | |
| "Anthropic API does not yet expose a deterministic seed parameter. " | |
| "Issue A reserves the seed kwarg for the local backend only." | |
| ) | |
| variables_with_knowledge = {**variables, "expert_knowledge": expert_knowledge} | |
| prompt = prompt_template.format(**variables_with_knowledge) | |
| cmd = [ | |
| self.claude_bin, | |
| "-p", | |
| "--output-format", "json", | |
| "--model", self.model, | |
| "--bare", | |
| "--no-session-persistence", | |
| "--max-tokens", str(self.max_tokens), | |
| ] | |
| if expert_knowledge: | |
| cmd.extend(["--system-prompt", expert_knowledge]) | |
| logger.debug(f"Claude CLI call: model={self.model}, prompt_len={len(prompt)}") | |
| result = subprocess.run( | |
| cmd, | |
| input=prompt, | |
| capture_output=True, | |
| text=True, | |
| timeout=300, | |
| ) | |
| if result.returncode != 0: | |
| logger.error(f"Claude CLI failed (rc={result.returncode}): {result.stderr[:500]}") | |
| raise RuntimeError(f"Claude CLI exited with code {result.returncode}: {result.stderr[:300]}") | |
| return self._parse_output(result.stdout) | |
| def _parse_output(self, stdout: str) -> str: | |
| """Extract the response text from Claude CLI JSON output.""" | |
| try: | |
| data = json.loads(stdout) | |
| except json.JSONDecodeError: | |
| # If not JSON, treat the raw stdout as the response (text mode fallback) | |
| return stdout.strip() | |
| if data.get("is_error"): | |
| error_msg = data.get("result", "Unknown CLI error") | |
| raise RuntimeError(f"Claude CLI error: {error_msg}") | |
| return data.get("result", "") | |