Spaces:
Sleeping
Sleeping
File size: 4,623 Bytes
0279c66 8d9359c 0279c66 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | import logging
from typing import List, Dict, Any
from param_mem.memory.parametric import ParametricMemory
from param_mem.memory.retrieval import CrossSampleMemory
logger = logging.getLogger(__name__)
class ParamAgent:
"""
ParamMem Agent Loop
Unifies episodic memory, cross-sample trajectory banks (ParamAgent-plus),
and parametric memory for iterative task solving.
"""
def __init__(self, memory_module: ParametricMemory, retrieval_module: CrossSampleMemory = None):
self.memory_module = memory_module
self.retrieval_module = retrieval_module
def solve_task(self, problem: str, max_iterations: int = 3) -> Dict[str, Any]:
"""
Executes the iterative reasoning loop.
"""
episodic_memory = []
current_solution = None
# 1. Retrieve cross-sample patterns if ParamAgent-plus
cross_sample_context = ""
if self.retrieval_module:
patterns = self.retrieval_module.retrieve_similar_patterns(problem)
if patterns:
cross_sample_context = "Helpful reasoning patterns from past experiences:\n"
for p in patterns:
cross_sample_context += f"- {p.get('reflection', '')}\n"
for iteration in range(max_iterations):
logger.info(f"Starting iteration {iteration + 1}/{max_iterations}")
# Step 1: Propose Solution
prompt = self._build_prompt(problem, cross_sample_context, episodic_memory)
# Use abstracted generation method
current_solution = self.memory_module.generate_solution(
prompt,
max_new_tokens=1024,
temperature=0.2
)
# Step 2: Evaluate (mock evaluation for this deployable scaffold)
is_correct, error_signal = self._mock_evaluate(current_solution)
if is_correct:
logger.info("Solution is correct!")
if self.retrieval_module:
self.retrieval_module.add_trajectory(
trajectory_id=f"traj_{hash(problem)}",
problem_description=problem,
final_reflection=f"Successfully solved by: {current_solution}"
)
return {
"success": True,
"solution": current_solution,
"iterations": iteration + 1,
"history": episodic_memory
}
# Step 3: Reflect (Using Parametric Memory)
temp = min(0.7 + (iteration * 0.1), 1.0)
reflection = self.memory_module.generate_reflection(
context=current_solution,
error_signal=error_signal,
temperature=temp
)
episodic_memory.append({
"attempt": current_solution,
"error": error_signal,
"reflection": reflection
})
logger.info(f"Reflection generated: {reflection}")
return {
"success": False,
"solution": current_solution,
"iterations": max_iterations,
"history": episodic_memory
}
def _build_prompt(self, problem: str, cross_sample_context: str, episodic_memory: List[Dict]) -> str:
prompt = f"Problem: {problem}\n"
if cross_sample_context:
prompt += f"{cross_sample_context}\n"
if episodic_memory:
prompt += "Previous Attempts and Reflections:\n"
for i, mem in enumerate(episodic_memory):
prompt += f"Attempt {i+1}: {mem['attempt']}\n"
prompt += f"Error: {mem['error']}\n"
prompt += f"Reflection: {mem['reflection']}\n"
prompt += "\nProvide the best solution to the problem based on the above context:\n"
return prompt
def _mock_evaluate(self, solution: str):
# Always fails once to demonstrate reflection loop, then succeeds if 'mock_success' is mentioned
# Or just simulate a success on iteration 2
if "correctly" in solution or "fixed" in solution:
return True, ""
# Default mock behavior
if "def " in solution and "return" in solution:
return False, "AssertionError: Expected output 5, got 3. Check your logic."
return False, "SyntaxError: invalid syntax"
|