Source code for curate_ipsum.synthesis.llm_client
"""
Abstract LLM client for code synthesis.
Defines the LLMClient ABC and MockLLMClient for testing.
Cloud and local backends are in separate modules (cloud_llm.py, local_llm.py).
Design decision: mirrors D-001's dual extractor pattern — abstract base class
with multiple concrete backends selectable at runtime.
"""
from __future__ import annotations
import abc
import logging
from curate_ipsum.synthesis.models import Counterexample, Specification
LOG = logging.getLogger("synthesis.llm_client")
[docs]
class LLMClient(abc.ABC):
"""Abstract base class for LLM code generation backends."""
[docs]
@abc.abstractmethod
async def generate_candidates(
self,
prompt: str,
n: int = 5,
temperature: float = 0.8,
) -> list[str]:
"""
Generate n code candidate strings from the LLM.
Returns raw code strings (not parsed). Caller is responsible for
syntactic validation.
"""
...
[docs]
async def close(self) -> None:
"""Clean up resources (e.g., HTTP clients). Override if needed."""
pass
[docs]
class MockLLMClient(LLMClient):
"""
Mock LLM client for testing.
Returns canned responses or generates simple variants of a template.
"""
def __init__(self, responses: list[str] | None = None) -> None:
self._responses = responses or []
self._call_count = 0
[docs]
async def generate_candidates(
self,
prompt: str,
n: int = 5,
temperature: float = 0.8,
) -> list[str]:
self._call_count += 1
if self._responses:
# Return up to n responses, cycling if needed
result = []
for i in range(n):
result.append(self._responses[i % len(self._responses)])
return result
# Default: return simple placeholder functions
return [f"def patched_func(x):\n return x + {i}\n" for i in range(n)]
@property
def call_count(self) -> int:
return self._call_count
[docs]
def build_synthesis_prompt(
spec: Specification,
counterexamples: list[Counterexample] | None = None,
context_code: str = "",
) -> str:
"""
Build an LLM prompt for code synthesis.
Includes:
- The original code being replaced
- Test requirements
- Surviving mutant information
- Counterexample history (what previous attempts failed on)
- Preconditions and postconditions from M3 assertions
"""
parts: list[str] = []
parts.append("Generate a Python function that satisfies the following requirements.\n")
if spec.original_code:
parts.append(f"## Original Code\n```python\n{spec.original_code}\n```\n")
if context_code:
parts.append(f"## Context\n```python\n{context_code}\n```\n")
if spec.preconditions:
parts.append("## Preconditions")
for pre in spec.preconditions:
parts.append(f"- {pre}")
parts.append("")
if spec.postconditions:
parts.append("## Postconditions")
for post in spec.postconditions:
parts.append(f"- {post}")
parts.append("")
if spec.surviving_mutant_ids:
parts.append("## Target: Kill surviving mutants")
parts.append(f"Mutant IDs: {', '.join(spec.surviving_mutant_ids)}")
parts.append("The patch must cause these mutants to be detected (killed) by the test suite.\n")
if spec.test_commands:
parts.append("## Tests that must pass")
for cmd in spec.test_commands:
parts.append(f"- `{cmd}`")
parts.append("")
if counterexamples:
parts.append("## Previous attempts failed on these counterexamples")
for ce in counterexamples[-5:]: # Show last 5 CEs to avoid prompt bloat
parts.append(f"- Input: {ce.input_values}, Expected: {ce.expected_output}, Got: {ce.actual_output}")
if ce.error_message:
parts.append(f" Error: {ce.error_message}")
parts.append("")
parts.append("Return ONLY the Python code, no explanations or markdown fences.")
return "\n".join(parts)