wehub-skill-sync
863 lines
30 KiB
Python
863 lines
30 KiB
Python
"""
|
|
Context Compression Evaluation
|
|
|
|
Public API for evaluating context compression quality using probe-based
|
|
assessment. This module provides three composable components:
|
|
|
|
- **ProbeGenerator**: Extracts factual claims, file operations, and decisions
|
|
from conversation history, then generates typed probes for evaluation.
|
|
Use when: building a compression evaluation pipeline and needing to
|
|
automatically derive test questions from raw conversation history.
|
|
|
|
- **CompressionEvaluator**: Scores probe responses against a multi-dimensional
|
|
rubric (accuracy, context awareness, artifact trail, completeness,
|
|
continuity, instruction following). Use when: comparing compression methods
|
|
or validating that a compression strategy preserves critical information.
|
|
|
|
- **StructuredSummarizer**: Implements anchored iterative summarization with
|
|
explicit sections for session intent, file tracking, decisions, and next
|
|
steps. Use when: compressing long-running coding sessions where file
|
|
tracking and decision rationale must survive compression.
|
|
|
|
Top-level convenience function:
|
|
- **evaluate_compression_quality**: End-to-end pipeline that generates probes,
|
|
collects model responses, evaluates them, and returns a scored summary with
|
|
recommendations. Use when: running a one-shot compression quality check
|
|
without wiring up individual components.
|
|
|
|
PRODUCTION NOTES:
|
|
- The LLM judge calls are stubbed for demonstration. Production systems
|
|
should implement actual API calls to a frontier model.
|
|
- Token estimation uses simplified heuristics. Production systems should
|
|
use model-specific tokenizers.
|
|
- Ground truth extraction uses pattern matching. Production systems may
|
|
benefit from more sophisticated fact extraction.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Dict, Optional, Callable
|
|
from enum import Enum
|
|
import json
|
|
import re
|
|
|
|
__all__ = [
|
|
"ProbeType",
|
|
"Probe",
|
|
"CriterionResult",
|
|
"EvaluationResult",
|
|
"RUBRIC_CRITERIA",
|
|
"ProbeGenerator",
|
|
"CompressionEvaluator",
|
|
"StructuredSummarizer",
|
|
"evaluate_compression_quality",
|
|
]
|
|
|
|
|
|
class ProbeType(Enum):
|
|
"""Types of evaluation probes for compression quality assessment."""
|
|
RECALL = "recall"
|
|
ARTIFACT = "artifact"
|
|
CONTINUATION = "continuation"
|
|
DECISION = "decision"
|
|
|
|
|
|
@dataclass
|
|
class Probe:
|
|
"""A probe question for evaluating compression quality.
|
|
|
|
Use when: constructing evaluation inputs for CompressionEvaluator.
|
|
Each probe targets a specific information category that compression
|
|
may have lost.
|
|
"""
|
|
probe_type: ProbeType
|
|
question: str
|
|
ground_truth: Optional[str] = None
|
|
context_reference: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class CriterionResult:
|
|
"""Result for a single evaluation criterion."""
|
|
criterion_id: str
|
|
score: float
|
|
reasoning: str
|
|
|
|
|
|
@dataclass
|
|
class EvaluationResult:
|
|
"""Complete evaluation result for a probe response.
|
|
|
|
Contains per-criterion scores, per-dimension aggregates, and an
|
|
overall aggregate score.
|
|
"""
|
|
probe: Probe
|
|
response: str
|
|
criterion_results: List[CriterionResult]
|
|
aggregate_score: float
|
|
dimension_scores: Dict[str, float] = field(default_factory=dict)
|
|
|
|
|
|
# Evaluation Rubrics
|
|
|
|
RUBRIC_CRITERIA: Dict[str, List[Dict]] = {
|
|
"accuracy": [
|
|
{
|
|
"id": "accuracy_factual",
|
|
"question": "Are facts, file paths, and technical details correct?",
|
|
"weight": 0.6
|
|
},
|
|
{
|
|
"id": "accuracy_technical",
|
|
"question": "Are code references and technical concepts correct?",
|
|
"weight": 0.4
|
|
}
|
|
],
|
|
"context_awareness": [
|
|
{
|
|
"id": "context_conversation_state",
|
|
"question": "Does the response reflect current conversation state?",
|
|
"weight": 0.5
|
|
},
|
|
{
|
|
"id": "context_artifact_state",
|
|
"question": "Does the response reflect which files/artifacts were accessed?",
|
|
"weight": 0.5
|
|
}
|
|
],
|
|
"artifact_trail": [
|
|
{
|
|
"id": "artifact_files_created",
|
|
"question": "Does the agent know which files were created?",
|
|
"weight": 0.3
|
|
},
|
|
{
|
|
"id": "artifact_files_modified",
|
|
"question": "Does the agent know which files were modified?",
|
|
"weight": 0.4
|
|
},
|
|
{
|
|
"id": "artifact_key_details",
|
|
"question": "Does the agent remember function names, variable names, error messages?",
|
|
"weight": 0.3
|
|
}
|
|
],
|
|
"completeness": [
|
|
{
|
|
"id": "completeness_coverage",
|
|
"question": "Does the response address all parts of the question?",
|
|
"weight": 0.6
|
|
},
|
|
{
|
|
"id": "completeness_depth",
|
|
"question": "Is sufficient detail provided?",
|
|
"weight": 0.4
|
|
}
|
|
],
|
|
"continuity": [
|
|
{
|
|
"id": "continuity_work_state",
|
|
"question": "Can the agent continue without re-fetching information?",
|
|
"weight": 0.4
|
|
},
|
|
{
|
|
"id": "continuity_todo_state",
|
|
"question": "Does the agent maintain awareness of pending tasks?",
|
|
"weight": 0.3
|
|
},
|
|
{
|
|
"id": "continuity_reasoning",
|
|
"question": "Does the agent retain rationale behind previous decisions?",
|
|
"weight": 0.3
|
|
}
|
|
],
|
|
"instruction_following": [
|
|
{
|
|
"id": "instruction_format",
|
|
"question": "Does the response follow the requested format?",
|
|
"weight": 0.5
|
|
},
|
|
{
|
|
"id": "instruction_constraints",
|
|
"question": "Does the response respect stated constraints?",
|
|
"weight": 0.5
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
class ProbeGenerator:
|
|
"""Generate typed probes from conversation history.
|
|
|
|
Use when: automatically deriving evaluation questions from raw
|
|
conversation history at compression points. Extracts facts, file
|
|
operations, and decisions via pattern matching, then produces
|
|
one probe per category.
|
|
|
|
For production systems, replace the regex-based extraction with
|
|
an LLM-based extractor for higher recall.
|
|
"""
|
|
|
|
def __init__(self, conversation_history: str) -> None:
|
|
self.history = conversation_history
|
|
self.extracted_facts = self._extract_facts()
|
|
self.extracted_files = self._extract_files()
|
|
self.extracted_decisions = self._extract_decisions()
|
|
|
|
def generate_probes(self) -> List[Probe]:
|
|
"""Generate all probe types for evaluation.
|
|
|
|
Use when: preparing evaluation inputs at a compression point.
|
|
Returns one probe per category (recall, artifact, continuation,
|
|
decision) based on extractable content from the history.
|
|
"""
|
|
probes: List[Probe] = []
|
|
|
|
# Recall probes
|
|
if self.extracted_facts:
|
|
probes.append(Probe(
|
|
probe_type=ProbeType.RECALL,
|
|
question="What was the original error or issue that started this session?",
|
|
ground_truth=self.extracted_facts.get("original_error"),
|
|
context_reference="session_start"
|
|
))
|
|
|
|
# Artifact probes
|
|
if self.extracted_files:
|
|
probes.append(Probe(
|
|
probe_type=ProbeType.ARTIFACT,
|
|
question="Which files have we modified? Describe what changed in each.",
|
|
ground_truth=json.dumps(self.extracted_files),
|
|
context_reference="file_operations"
|
|
))
|
|
|
|
# Continuation probes
|
|
probes.append(Probe(
|
|
probe_type=ProbeType.CONTINUATION,
|
|
question="What should we do next?",
|
|
ground_truth=self.extracted_facts.get("next_steps"),
|
|
context_reference="task_state"
|
|
))
|
|
|
|
# Decision probes
|
|
if self.extracted_decisions:
|
|
probes.append(Probe(
|
|
probe_type=ProbeType.DECISION,
|
|
question="What key decisions did we make and why?",
|
|
ground_truth=json.dumps(self.extracted_decisions),
|
|
context_reference="decision_points"
|
|
))
|
|
|
|
return probes
|
|
|
|
def _extract_facts(self) -> Dict[str, str]:
|
|
"""Extract factual claims from history."""
|
|
facts: Dict[str, str] = {}
|
|
|
|
# Extract error patterns
|
|
error_patterns = [
|
|
r"error[:\s]+(.+?)(?:\n|$)",
|
|
r"(\d{3})\s+(Unauthorized|Not Found|Internal Server Error)",
|
|
r"exception[:\s]+(.+?)(?:\n|$)"
|
|
]
|
|
|
|
for pattern in error_patterns:
|
|
match = re.search(pattern, self.history, re.IGNORECASE)
|
|
if match:
|
|
facts["original_error"] = match.group(0).strip()
|
|
break
|
|
|
|
# Extract next steps
|
|
next_step_patterns = [
|
|
r"next[:\s]+(.+?)(?:\n|$)",
|
|
r"TODO[:\s]+(.+?)(?:\n|$)",
|
|
r"remaining[:\s]+(.+?)(?:\n|$)"
|
|
]
|
|
|
|
for pattern in next_step_patterns:
|
|
match = re.search(pattern, self.history, re.IGNORECASE)
|
|
if match:
|
|
facts["next_steps"] = match.group(0).strip()
|
|
break
|
|
|
|
return facts
|
|
|
|
def _extract_files(self) -> List[Dict[str, str]]:
|
|
"""Extract file operations from history."""
|
|
files: List[Dict[str, str]] = []
|
|
|
|
# Common file patterns
|
|
file_patterns = [
|
|
r"(?:modified|changed|updated|edited)\s+([^\s]+\.[a-z]+)",
|
|
r"(?:created|added)\s+([^\s]+\.[a-z]+)",
|
|
r"(?:read|examined|opened)\s+([^\s]+\.[a-z]+)"
|
|
]
|
|
|
|
for pattern in file_patterns:
|
|
matches = re.findall(pattern, self.history, re.IGNORECASE)
|
|
for match in matches:
|
|
if match not in [f["path"] for f in files]:
|
|
files.append({
|
|
"path": match,
|
|
"operation": "modified" if "modif" in pattern else "created" if "creat" in pattern else "read"
|
|
})
|
|
|
|
return files
|
|
|
|
def _extract_decisions(self) -> List[Dict[str, str]]:
|
|
"""Extract decision points from history."""
|
|
decisions: List[Dict[str, str]] = []
|
|
|
|
decision_patterns = [
|
|
r"decided to\s+(.+?)(?:\n|$)",
|
|
r"chose\s+(.+?)(?:\n|$)",
|
|
r"going with\s+(.+?)(?:\n|$)",
|
|
r"will use\s+(.+?)(?:\n|$)"
|
|
]
|
|
|
|
for pattern in decision_patterns:
|
|
matches = re.findall(pattern, self.history, re.IGNORECASE)
|
|
for match in matches:
|
|
decisions.append({
|
|
"decision": match.strip(),
|
|
"context": pattern.split("\\s+")[0]
|
|
})
|
|
|
|
return decisions[:5] # Limit to 5 decisions
|
|
|
|
|
|
class CompressionEvaluator:
|
|
"""Evaluate compression quality using probes and LLM judge.
|
|
|
|
Use when: comparing compression methods or validating that a specific
|
|
compression pass preserved critical information. Scores responses
|
|
across six dimensions (accuracy, context awareness, artifact trail,
|
|
completeness, continuity, instruction following) and produces an
|
|
aggregate quality score.
|
|
|
|
The evaluate() method is the primary entry point. Call it once per
|
|
probe, then call get_summary() to retrieve aggregated results.
|
|
"""
|
|
|
|
def __init__(self, model: str = "gpt-5.2") -> None:
|
|
self.model = model
|
|
self.results: List[EvaluationResult] = []
|
|
|
|
def evaluate(self,
|
|
probe: Probe,
|
|
response: str,
|
|
compressed_context: str) -> EvaluationResult:
|
|
"""Evaluate a single probe response against the rubric.
|
|
|
|
Use when: scoring how well a model's response (given compressed
|
|
context) answers a probe question. Returns per-criterion scores,
|
|
per-dimension aggregates, and an overall score.
|
|
|
|
Args:
|
|
probe: The probe question with expected ground truth.
|
|
response: The model's response to evaluate.
|
|
compressed_context: The compressed context that was provided
|
|
to the model when generating the response.
|
|
|
|
Returns:
|
|
EvaluationResult with scores and reasoning across all
|
|
applicable dimensions.
|
|
"""
|
|
# Get relevant criteria based on probe type
|
|
criteria = self._get_criteria_for_probe(probe.probe_type)
|
|
|
|
# Evaluate each criterion
|
|
criterion_results: List[CriterionResult] = []
|
|
for criterion in criteria:
|
|
result = self._evaluate_criterion(
|
|
criterion,
|
|
probe,
|
|
response,
|
|
compressed_context
|
|
)
|
|
criterion_results.append(result)
|
|
|
|
# Calculate dimension scores
|
|
dimension_scores = self._calculate_dimension_scores(criterion_results)
|
|
|
|
# Calculate aggregate score
|
|
aggregate_score = sum(dimension_scores.values()) / len(dimension_scores) if dimension_scores else 0.0
|
|
|
|
result = EvaluationResult(
|
|
probe=probe,
|
|
response=response,
|
|
criterion_results=criterion_results,
|
|
aggregate_score=aggregate_score,
|
|
dimension_scores=dimension_scores
|
|
)
|
|
|
|
self.results.append(result)
|
|
return result
|
|
|
|
def get_summary(self) -> Dict:
|
|
"""Get summary of all evaluation results.
|
|
|
|
Use when: all probes have been evaluated and an aggregate
|
|
report is needed to compare methods or make a go/no-go
|
|
decision on a compression strategy.
|
|
|
|
Returns:
|
|
Dictionary with total evaluations, average score,
|
|
per-dimension averages, and weakest/strongest dimensions.
|
|
"""
|
|
if not self.results:
|
|
return {"error": "No evaluations performed"}
|
|
|
|
avg_score = sum(r.aggregate_score for r in self.results) / len(self.results)
|
|
|
|
# Average dimension scores
|
|
dimension_totals: Dict[str, float] = {}
|
|
dimension_counts: Dict[str, int] = {}
|
|
|
|
for result in self.results:
|
|
for dim, score in result.dimension_scores.items():
|
|
dimension_totals[dim] = dimension_totals.get(dim, 0) + score
|
|
dimension_counts[dim] = dimension_counts.get(dim, 0) + 1
|
|
|
|
avg_dimensions = {
|
|
dim: dimension_totals[dim] / dimension_counts[dim]
|
|
for dim in dimension_totals
|
|
}
|
|
|
|
return {
|
|
"total_evaluations": len(self.results),
|
|
"average_score": avg_score,
|
|
"dimension_averages": avg_dimensions,
|
|
"weakest_dimension": min(avg_dimensions, key=avg_dimensions.get) if avg_dimensions else None,
|
|
"strongest_dimension": max(avg_dimensions, key=avg_dimensions.get) if avg_dimensions else None,
|
|
}
|
|
|
|
def _get_criteria_for_probe(self, probe_type: ProbeType) -> List[Dict]:
|
|
"""Get relevant criteria for probe type."""
|
|
criteria: List[Dict] = []
|
|
|
|
# All probes get accuracy and completeness
|
|
criteria.extend(RUBRIC_CRITERIA["accuracy"])
|
|
criteria.extend(RUBRIC_CRITERIA["completeness"])
|
|
|
|
# Add type-specific criteria
|
|
if probe_type == ProbeType.ARTIFACT:
|
|
criteria.extend(RUBRIC_CRITERIA["artifact_trail"])
|
|
elif probe_type == ProbeType.CONTINUATION:
|
|
criteria.extend(RUBRIC_CRITERIA["continuity"])
|
|
elif probe_type == ProbeType.RECALL:
|
|
criteria.extend(RUBRIC_CRITERIA["context_awareness"])
|
|
elif probe_type == ProbeType.DECISION:
|
|
criteria.extend(RUBRIC_CRITERIA["context_awareness"])
|
|
criteria.extend(RUBRIC_CRITERIA["continuity"])
|
|
|
|
criteria.extend(RUBRIC_CRITERIA["instruction_following"])
|
|
|
|
return criteria
|
|
|
|
def _evaluate_criterion(self,
|
|
criterion: Dict,
|
|
probe: Probe,
|
|
response: str,
|
|
context: str) -> CriterionResult:
|
|
"""
|
|
Evaluate a single criterion using LLM judge.
|
|
|
|
PRODUCTION NOTE: This is a stub implementation.
|
|
Production systems should call the actual LLM API:
|
|
|
|
```python
|
|
result = openai.chat.completions.create(
|
|
model="gpt-5.2",
|
|
messages=[
|
|
{"role": "system", "content": JUDGE_SYSTEM_PROMPT},
|
|
{"role": "user", "content": self._format_judge_input(criterion, probe, response, context)}
|
|
]
|
|
)
|
|
return self._parse_judge_output(result)
|
|
```
|
|
"""
|
|
# Stub implementation - in production, call LLM judge
|
|
score = self._heuristic_score(criterion, response, probe.ground_truth)
|
|
reasoning = f"Evaluated {criterion['id']} based on response content."
|
|
|
|
return CriterionResult(
|
|
criterion_id=criterion["id"],
|
|
score=score,
|
|
reasoning=reasoning
|
|
)
|
|
|
|
def _heuristic_score(self,
|
|
criterion: Dict,
|
|
response: str,
|
|
ground_truth: Optional[str]) -> float:
|
|
"""
|
|
Heuristic scoring for demonstration.
|
|
|
|
Production systems should use LLM judge instead.
|
|
"""
|
|
score = 3.0 # Base score
|
|
|
|
# Adjust based on response length and content
|
|
if len(response) < 50:
|
|
score -= 1.0 # Too short
|
|
elif len(response) > 500:
|
|
score += 0.5 # Detailed
|
|
|
|
# Check for technical content
|
|
if any(ext in response for ext in [".ts", ".py", ".js", ".md"]):
|
|
score += 0.5 # Contains file references
|
|
|
|
overlap_ratio = self._ground_truth_overlap_ratio(response, ground_truth)
|
|
if overlap_ratio >= 0.75:
|
|
score += 1.0
|
|
elif overlap_ratio >= 0.4:
|
|
score += 0.5
|
|
elif ground_truth:
|
|
score -= 0.5
|
|
|
|
return min(5.0, max(0.0, score))
|
|
|
|
def _ground_truth_overlap_ratio(self,
|
|
response: str,
|
|
ground_truth: Optional[str]) -> float:
|
|
if not ground_truth:
|
|
return 0.0
|
|
|
|
terms = self._extract_ground_truth_terms(ground_truth)
|
|
if not terms:
|
|
return 1.0 if ground_truth.lower() in response.lower() else 0.0
|
|
|
|
response_lower = response.lower()
|
|
matches = sum(1 for term in terms if term in response_lower)
|
|
return matches / len(terms)
|
|
|
|
def _extract_ground_truth_terms(self, ground_truth: str) -> List[str]:
|
|
try:
|
|
parsed = json.loads(ground_truth)
|
|
except json.JSONDecodeError:
|
|
return [ground_truth.lower()] if ground_truth.strip() else []
|
|
|
|
terms: List[str] = []
|
|
|
|
def collect(value) -> None:
|
|
if isinstance(value, str):
|
|
normalized = value.strip().lower()
|
|
if normalized:
|
|
terms.append(normalized)
|
|
elif isinstance(value, dict):
|
|
for nested in value.values():
|
|
collect(nested)
|
|
elif isinstance(value, list):
|
|
for nested in value:
|
|
collect(nested)
|
|
|
|
collect(parsed)
|
|
return list(dict.fromkeys(terms))
|
|
|
|
def _calculate_dimension_scores(self,
|
|
criterion_results: List[CriterionResult]) -> Dict[str, float]:
|
|
"""Calculate dimension scores from criterion results."""
|
|
dimension_scores: Dict[str, float] = {}
|
|
|
|
for dimension, criteria in RUBRIC_CRITERIA.items():
|
|
criterion_ids = [c["id"] for c in criteria]
|
|
relevant_results = [
|
|
r for r in criterion_results
|
|
if r.criterion_id in criterion_ids
|
|
]
|
|
|
|
if relevant_results:
|
|
# Weighted average
|
|
total_weight = sum(
|
|
c["weight"] for c in criteria
|
|
if c["id"] in [r.criterion_id for r in relevant_results]
|
|
)
|
|
weighted_sum = sum(
|
|
r.score * next(c["weight"] for c in criteria if c["id"] == r.criterion_id)
|
|
for r in relevant_results
|
|
)
|
|
dimension_scores[dimension] = weighted_sum / total_weight if total_weight > 0 else 0.0
|
|
|
|
return dimension_scores
|
|
|
|
|
|
class StructuredSummarizer:
|
|
"""Generate structured summaries with explicit sections.
|
|
|
|
Use when: implementing anchored iterative summarization for
|
|
long-running coding sessions. Maintains a persistent summary
|
|
with dedicated sections for session intent, file modifications,
|
|
decisions, current state, and next steps.
|
|
|
|
Call update_from_span() each time a new content span is truncated.
|
|
The summarizer merges new information into existing sections rather
|
|
than regenerating, preventing cumulative detail loss.
|
|
"""
|
|
|
|
TEMPLATE = """## Session Intent
|
|
{intent}
|
|
|
|
## Files Modified
|
|
{files_modified}
|
|
|
|
## Files Read (Not Modified)
|
|
{files_read}
|
|
|
|
## Decisions Made
|
|
{decisions}
|
|
|
|
## Current State
|
|
{current_state}
|
|
|
|
## Next Steps
|
|
{next_steps}
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.sections: Dict = {
|
|
"intent": "",
|
|
"files_modified": [],
|
|
"files_read": [],
|
|
"decisions": [],
|
|
"current_state": "",
|
|
"next_steps": []
|
|
}
|
|
|
|
def update_from_span(self, new_content: str) -> str:
|
|
"""Update summary from newly truncated content span.
|
|
|
|
Use when: a compression trigger fires and a portion of
|
|
conversation history is about to be discarded. Pass the
|
|
content that will be truncated; the summarizer extracts
|
|
structured information and merges it with prior state.
|
|
|
|
Args:
|
|
new_content: The conversation span being truncated.
|
|
|
|
Returns:
|
|
Formatted summary string with all sections populated.
|
|
"""
|
|
# Extract information from new content
|
|
new_info = self._extract_from_content(new_content)
|
|
|
|
# Merge with existing sections
|
|
self._merge_sections(new_info)
|
|
|
|
# Generate formatted summary
|
|
return self._format_summary()
|
|
|
|
def _extract_from_content(self, content: str) -> Dict:
|
|
"""Extract structured information from content."""
|
|
extracted: Dict = {
|
|
"intent": "",
|
|
"files_modified": [],
|
|
"files_read": [],
|
|
"decisions": [],
|
|
"current_state": "",
|
|
"next_steps": []
|
|
}
|
|
|
|
# Extract file modifications
|
|
mod_pattern = r"(?:modified|changed|updated|fixed)\s+([^\s]+\.[a-z]+)[:\s]*(.+?)(?:\n|$)"
|
|
for match in re.finditer(mod_pattern, content, re.IGNORECASE):
|
|
extracted["files_modified"].append({
|
|
"path": match.group(1),
|
|
"change": match.group(2).strip()[:100]
|
|
})
|
|
|
|
# Extract file reads
|
|
read_pattern = r"(?:read|examined|opened|checked)\s+([^\s]+\.[a-z]+)"
|
|
for match in re.finditer(read_pattern, content, re.IGNORECASE):
|
|
file_path = match.group(1)
|
|
if file_path not in [f["path"] for f in extracted["files_modified"]]:
|
|
extracted["files_read"].append(file_path)
|
|
|
|
# Extract decisions
|
|
decision_pattern = r"(?:decided|chose|going with|will use)\s+(.+?)(?:\n|$)"
|
|
for match in re.finditer(decision_pattern, content, re.IGNORECASE):
|
|
extracted["decisions"].append(match.group(1).strip()[:150])
|
|
|
|
return extracted
|
|
|
|
def _merge_sections(self, new_info: Dict) -> None:
|
|
"""Merge new information with existing sections."""
|
|
# Update intent if empty
|
|
if new_info["intent"] and not self.sections["intent"]:
|
|
self.sections["intent"] = new_info["intent"]
|
|
|
|
# Merge file lists (deduplicate by path)
|
|
existing_mod_paths = [f["path"] for f in self.sections["files_modified"]]
|
|
for file_info in new_info["files_modified"]:
|
|
if file_info["path"] not in existing_mod_paths:
|
|
self.sections["files_modified"].append(file_info)
|
|
|
|
# Merge read files
|
|
for file_path in new_info["files_read"]:
|
|
if file_path not in self.sections["files_read"]:
|
|
self.sections["files_read"].append(file_path)
|
|
|
|
# Append decisions
|
|
self.sections["decisions"].extend(new_info["decisions"])
|
|
|
|
# Update current state (latest wins)
|
|
if new_info["current_state"]:
|
|
self.sections["current_state"] = new_info["current_state"]
|
|
|
|
# Merge next steps
|
|
self.sections["next_steps"].extend(new_info["next_steps"])
|
|
|
|
def _format_summary(self) -> str:
|
|
"""Format sections into summary string."""
|
|
files_modified_str = "\n".join(
|
|
f"- {f['path']}: {f['change']}"
|
|
for f in self.sections["files_modified"]
|
|
) or "None"
|
|
|
|
files_read_str = "\n".join(
|
|
f"- {f}" for f in self.sections["files_read"]
|
|
) or "None"
|
|
|
|
decisions_str = "\n".join(
|
|
f"- {d}" for d in self.sections["decisions"][-5:] # Keep last 5
|
|
) or "None"
|
|
|
|
next_steps_str = "\n".join(
|
|
f"{i+1}. {s}" for i, s in enumerate(self.sections["next_steps"][-5:])
|
|
) or "None"
|
|
|
|
return self.TEMPLATE.format(
|
|
intent=self.sections["intent"] or "Not specified",
|
|
files_modified=files_modified_str,
|
|
files_read=files_read_str,
|
|
decisions=decisions_str,
|
|
current_state=self.sections["current_state"] or "In progress",
|
|
next_steps=next_steps_str
|
|
)
|
|
|
|
|
|
def evaluate_compression_quality(
|
|
original_history: str,
|
|
compressed_context: str,
|
|
model_response_fn: Callable[[str, str], str],
|
|
) -> Dict:
|
|
"""Evaluate compression quality for a conversation end-to-end.
|
|
|
|
Use when: running a one-shot quality check on a compression pass.
|
|
Generates probes from original history, collects model responses
|
|
using the compressed context, evaluates each response, and returns
|
|
a scored summary with actionable recommendations.
|
|
|
|
Args:
|
|
original_history: The full conversation before compression.
|
|
compressed_context: The compressed version to evaluate.
|
|
model_response_fn: Callable that takes (compressed_context, question)
|
|
and returns the model's response string.
|
|
|
|
Returns:
|
|
Dictionary with total evaluations, average score, per-dimension
|
|
averages, weakest/strongest dimensions, and recommendations list.
|
|
"""
|
|
# Generate probes
|
|
generator = ProbeGenerator(original_history)
|
|
probes = generator.generate_probes()
|
|
|
|
# Evaluate each probe
|
|
evaluator = CompressionEvaluator()
|
|
|
|
for probe in probes:
|
|
# Get model response using compressed context
|
|
response = model_response_fn(compressed_context, probe.question)
|
|
|
|
# Evaluate response
|
|
evaluator.evaluate(probe, response, compressed_context)
|
|
|
|
# Get summary
|
|
summary = evaluator.get_summary()
|
|
|
|
# Add recommendations
|
|
summary["recommendations"] = []
|
|
|
|
if summary.get("weakest_dimension") == "artifact_trail":
|
|
summary["recommendations"].append(
|
|
"Consider implementing separate artifact tracking outside compression"
|
|
)
|
|
|
|
if summary.get("average_score", 0) < 3.5:
|
|
summary["recommendations"].append(
|
|
"Compression quality is below threshold - consider less aggressive compression"
|
|
)
|
|
|
|
return summary
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Demo: generate probes and evaluate a sample compression
|
|
|
|
sample_history = """
|
|
User reported error: 401 Unauthorized on /api/auth/login endpoint.
|
|
Examined auth.controller.ts - JWT generation looks correct.
|
|
Examined middleware/cors.ts - no issues found.
|
|
Modified config/redis.ts: Fixed connection pooling configuration.
|
|
Modified services/session.service.ts: Added retry logic for transient failures.
|
|
Decided to use Redis connection pool instead of per-request connections.
|
|
Modified tests/auth.test.ts: Updated mock setup for new config.
|
|
14 tests passing, 2 failing (mock setup issues).
|
|
Next: Fix remaining test failures in session service mocks.
|
|
"""
|
|
|
|
sample_compressed = """
|
|
## Session Intent
|
|
Debug 401 Unauthorized on /api/auth/login.
|
|
|
|
## Root Cause
|
|
Stale Redis connection in session store.
|
|
|
|
## Files Modified
|
|
- config/redis.ts: Fixed connection pooling
|
|
- services/session.service.ts: Added retry logic
|
|
- tests/auth.test.ts: Updated mock setup
|
|
|
|
## Test Status
|
|
14 passing, 2 failing
|
|
|
|
## Next Steps
|
|
1. Fix remaining test failures
|
|
"""
|
|
|
|
# Stub model response function
|
|
def mock_model_response(context: str, question: str) -> str:
|
|
if "error" in question.lower():
|
|
return "The original error was a 401 Unauthorized on /api/auth/login."
|
|
if "files" in question.lower():
|
|
return "Modified config/redis.ts, services/session.service.ts, tests/auth.test.ts."
|
|
if "next" in question.lower():
|
|
return "Fix remaining test failures in session service mocks."
|
|
if "decision" in question.lower():
|
|
return "Decided to use Redis connection pool instead of per-request connections."
|
|
return "No specific information available."
|
|
|
|
# Run evaluation
|
|
result = evaluate_compression_quality(
|
|
original_history=sample_history,
|
|
compressed_context=sample_compressed,
|
|
model_response_fn=mock_model_response,
|
|
)
|
|
|
|
print("=== Compression Quality Evaluation ===")
|
|
print(f"Total evaluations: {result['total_evaluations']}")
|
|
print(f"Average score: {result['average_score']:.2f}")
|
|
print()
|
|
print("Dimension averages:")
|
|
for dim, score in result.get("dimension_averages", {}).items():
|
|
print(f" {dim}: {score:.2f}")
|
|
print()
|
|
print(f"Weakest dimension: {result.get('weakest_dimension')}")
|
|
print(f"Strongest dimension: {result.get('strongest_dimension')}")
|
|
print()
|
|
if result.get("recommendations"):
|
|
print("Recommendations:")
|
|
for rec in result["recommendations"]:
|
|
print(f" - {rec}")
|
|
else:
|
|
print("No recommendations - compression quality looks acceptable.")
|