Continuous Learning GRPO - Training-Free AI Model Improvement¶

{
  "experiences": {
    "G0": "When solving equations, verify by substitution...",
    "G1": "For optimization, check boundary conditions...",
    ...
  },
  "next_id": 42
}

def summarize_trajectory(trajectory, use_groundtruth=True):
    """
    Input: Full trajectory, correctness label, ground truth
    Output: Step-by-step natural language summary

    Example:
      1. Applied quadratic formula with a=1, b=2, c=5
      2. Calculated discriminant: b²-4ac = 4-20 = -16
      3. Concluded no real solutions (correct)
    """

def extract_group_advantage(trajectories, experiences):
    """
    Input: G trajectories (both correct/incorrect), current experiences
    Output: JSON operations [{"option": "add", "experience": "..."}, ...]

    Process:
      - Compare successful vs failed trajectories
      - Identify what made successful ones succeed
      - Propose max 3 operations: add/modify/delete
    """

def consolidate_batch(all_group_operations, experiences):
    """
    Input: All group operations from batch, current experiences
    Output: Final consolidated operations

    Process:
      - Merge similar suggestions
      - Ensure ≤32 words per experience
      - Eliminate redundancy
    """

from gym.train.grpo.api_model_adapter import DeepSeekAdapter

adapter = DeepSeekAdapter(api_key="sk-xxx", model="deepseek-chat")
response = adapter.generate_with_experiences(
    query="Solve: x² + 2x + 5 = 0",
    experiences=experience_library.format_for_prompt()
)

from gym.train.grpo.experience_manager import ExperienceManager
from gym.train.grpo.semantic_extractor import SemanticExtractor, LLMClient
from gym.train.grpo.api_model_adapter import DeepSeekAdapter

# 1. Initialize components
experience_manager = ExperienceManager(checkpoint_path="./experiences.json")
llm_client = LLMClient(api_key="sk-xxx", model="deepseek-chat")
extractor = SemanticExtractor(llm_client, max_operations=3)
model_adapter = DeepSeekAdapter(api_key="sk-xxx")

# 2. Training loop
queries = [
    "Solve: x² + 2x + 5 = 0",
    "Find derivative of x³ + 2x",
    ...
]
groundtruths = ["No real solutions", "3x² + 2", ...]

for epoch in range(3):
    print(f"\n=== Epoch {epoch+1}/3 ===")

    for query, gt in zip(queries, groundtruths):
        # Generate G rollouts
        experiences = experience_manager.format_for_prompt()
        trajectories = []

        for _ in range(5):  # G=5
            output = model_adapter.generate_with_experiences(query, experiences)
            reward = evaluate_answer(output, gt)  # Your reward function
            trajectories.append(Trajectory(query, output, reward, gt))

        # Stage 1: Summarize
        for traj in trajectories:
            traj.summary = extractor.summarize_trajectory(traj)

        # Stage 2: Extract group advantage
        operations = extractor.extract_group_advantage(
            trajectories, experiences, use_groundtruth=True
        )

        # Apply operations
        experience_manager.apply_operations(operations)

    # Save checkpoint
    experience_manager.save(f"./experiences_epoch{epoch+1}.json")
    print(f"Experience library size: {len(experience_manager)}")

# 3. Final evaluation
print("\n=== Final Experience Library ===")
print(experience_manager.format_for_prompt())

def custom_reward_function(query: str, output: str, groundtruth: str) -> float:
    """
    Custom reward that considers partial correctness.

    Returns:
        reward: Float in [0, 1], higher is better
    """
    # Exact match
    if output.strip().lower() == groundtruth.strip().lower():
        return 1.0

    # Partial credit for methodology
    if "discriminant" in output and "quadratic" in query:
        return 0.5

    # Wrong answer
    return 0.0

# Use in training
reward = custom_reward_function(query, output, gt)

class DomainAwareExperienceManager(ExperienceManager):
    def __init__(self, checkpoint_path=None):
        super().__init__(checkpoint_path)
        self.domains = {}  # exp_id -> domain

    def add_with_domain(self, experience: str, domain: str) -> str:
        exp_id = self.add(experience)
        self.domains[exp_id] = domain
        return exp_id

    def format_for_domain(self, domain: str) -> str:
        """Return only experiences for specific domain."""
        filtered = {
            exp_id: text
            for exp_id, text in self.experiences.items()
            if self.domains.get(exp_id) == domain
        }
        return "\n".join([f"[{k}]. {v}" for k, v in filtered.items()])

# Usage
manager = DomainAwareExperienceManager()
manager.add_with_domain("For calculus problems...", domain="math.calculus")
manager.add_with_domain("When parsing JSON...", domain="coding.json")

# Retrieve domain-specific experiences
math_exp = manager.format_for_domain("math.calculus")

from sentence_transformers import SentenceTransformer

class SemanticExperienceManager(ExperienceManager):
    def __init__(self, checkpoint_path=None):
        super().__init__(checkpoint_path)
        self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
        self.embeddings = {}  # exp_id -> embedding

    def add(self, experience: str) -> str:
        exp_id = super().add(experience)
        self.embeddings[exp_id] = self.embedder.encode(experience)
        return exp_id

    def retrieve_relevant(self, query: str, top_k: int = 5) -> str:
        """Retrieve top-k most relevant experiences."""
        query_emb = self.embedder.encode(query)

        # Compute similarities
        similarities = {
            exp_id: cosine_similarity(query_emb, emb)
            for exp_id, emb in self.embeddings.items()
        }

        # Get top-k
        top_ids = sorted(similarities, key=similarities.get, reverse=True)[:top_k]

        # Format
        relevant = {exp_id: self.experiences[exp_id] for exp_id in top_ids}
        return "\n".join([f"[{k}]. {v}" for k, v in relevant.items()])

# Usage
manager = SemanticExperienceManager()
# ... add experiences ...
relevant_exp = manager.retrieve_relevant("Solve quadratic equation", top_k=3)

class GovernedExperienceManager(ExperienceManager):
    def __init__(self, checkpoint_path=None):
        super().__init__(checkpoint_path)
        self.votes = {}  # exp_id -> {"upvotes": int, "downvotes": int}
        self.vote_threshold = 0.66  # 2/3 majority

    def propose_experience(self, experience: str) -> str:
        """Propose experience for community vote."""
        exp_id = f"P{self._next_id}"  # Pending
        self.votes[exp_id] = {"upvotes": 0, "downvotes": 0, "text": experience}
        self._next_id += 1
        return exp_id

    def vote(self, exp_id: str, upvote: bool):
        """Cast vote on proposed experience."""
        if upvote:
            self.votes[exp_id]["upvotes"] += 1
        else:
            self.votes[exp_id]["downvotes"] += 1

    def finalize_votes(self):
        """Accept experiences with 2/3 majority."""
        for exp_id, vote_data in list(self.votes.items()):
            total = vote_data["upvotes"] + vote_data["downvotes"]
            if total == 0:
                continue

            approval_rate = vote_data["upvotes"] / total

            if approval_rate >= self.vote_threshold:
                # Accept experience
                self.add(vote_data["text"])
                print(f"✅ Accepted: {exp_id} ({approval_rate:.1%})")
            else:
                print(f"❌ Rejected: {exp_id} ({approval_rate:.1%})")

            del self.votes[exp_id]

# Phase 1: Prototype (10 examples, 1 epoch)
# Goal: Validate setup, check experience quality

# Phase 2: Pilot (50 examples, 2 epochs)
# Goal: Measure performance improvement

# Phase 3: Production (100-500 examples, 3 epochs)
# Goal: Deploy and monitor

def audit_experience_library(manager: ExperienceManager):
    """Check for common quality issues."""
    issues = []

    for exp_id, exp_text in manager.experiences.items():
        # Check length
        word_count = len(exp_text.split())
        if word_count > 32:
            issues.append(f"{exp_id}: Too long ({word_count} words)")

        # Check if too specific (mentions numbers)
        if any(char.isdigit() for char in exp_text):
            issues.append(f"{exp_id}: Contains numbers (too specific)")

        # Check if actionable (has verb)
        if not any(word in exp_text.lower() for word in ["use", "check", "verify", "apply", "calculate"]):
            issues.append(f"{exp_id}: No action verb (not actionable)")

    return issues

# Run after each epoch
issues = audit_experience_library(experience_manager)
if issues:
    print("⚠️  Quality Issues:")
    for issue in issues:
        print(f"  - {issue}")

# Save after every epoch
for epoch in range(num_epochs):
    # ... training ...

    # Checkpoint
    checkpoint_path = f"{output_dir}/experiences_epoch{epoch+1}.json"
    experience_manager.save(checkpoint_path)

    # Also save metadata
    metadata = {
        "epoch": epoch + 1,
        "num_experiences": len(experience_manager),
        "timestamp": datetime.now().isoformat(),
        "performance": evaluate_on_validation_set()
    }
    with open(f"{output_dir}/metadata_epoch{epoch+1}.json", 'w') as f:
        json.dump(metadata, f, indent=2)

# Hybrid approach:
# 1. Fine-tune base model on large general dataset (one-time)
# 2. Use Continuous Learning for task-specific adaptation (ongoing)

# Step 1: Fine-tune (traditional)
gym train \
  --model_name_or_path Qwen/Qwen3-4B \
  --finetuning_type lora \
  --dataset large_math_corpus \
  --output_dir ./output/math_base

# Step 2: Continuous Learning (on top of fine-tuned model)
gym train \
  --model_name_or_path ./output/math_base \
  --adapter_name_or_path ./output/math_base/lora_adapter \
  --config configs/continuous_learning.yaml \
  --dataset specific_geometry_problems \
  --output_dir ./output/geometry_adapted

# Test generalization
def cross_domain_evaluation(experience_manager):
    """Test if experiences help on unseen domains."""
    domains = ["math", "coding", "logic", "creative"]
    results = {}

    for domain in domains:
        test_set = load_test_set(domain)
        accuracy = evaluate_with_experiences(
            test_set,
            experience_manager.format_for_prompt()
        )
        results[domain] = accuracy

    return results

# Run after training
results = cross_domain_evaluation(experience_manager)
print("Cross-Domain Performance:")
for domain, acc in results.items():
    print(f"  {domain}: {acc:.1%}")

# 1. Increase group size
grpo_group_size: 5  # or higher

# 2. Check reward distribution
rewards = [t.reward for t in trajectories]
print(f"Reward std: {np.std(rewards)}")  # Should be > 0

# 3. Debug JSON parsing
response = extractor.extract_group_advantage(...)
if not response:
    print("No operations extracted - check LLM output")

# 1. Use stronger base model
model_name_or_path: "Qwen/Qwen3-7B-Instruct"  # or DeepSeek-V3 via API

# 2. Audit experience quality (see Best Practices)

# 3. Increase epochs
num_train_epochs: 5  # instead of 3

import time

# Add retry logic
def generate_with_retry(adapter, prompt, max_retries=3):
    for i in range(max_retries):
        try:
            return adapter.generate(prompt)
        except RateLimitError:
            if i < max_retries - 1:
                time.sleep(2 ** i)  # Exponential backoff
            else:
                raise

# Or use batch processing with delays
for batch in batches:
    process_batch(batch)
    time.sleep(1)  # 1 second between batches