菜单
Enable agent learning through episodic memory and neural case selection without fine-tuning the underlying LLM, achieving efficient continual adaptation via policy updates in memory space.
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/ADu2021/skillXiv --skill agentfly-memory-augmented-agents命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
正在显示 SKILL.md
| name | agentfly-memory-augmented-agents |
| title | AgentFly: Memory-Augmented Learning Without LLM Fine-tuning |
| version | 0.0.2 |
| engine | skillxiv-v0.0.2-claude-opus-4.6 |
| license | MIT |
| url | https://arxiv.org/abs/2508.16153 |
| keywords | ["memory-augmented-learning","agent-adaptation","episodic-memory","reinforcement-learning","in-context-learning"] |
| description | Enable agent learning through episodic memory and neural case selection without fine-tuning the underlying LLM, achieving efficient continual adaptation via policy updates in memory space. |
AgentFly (Memento) enables language model agents to learn and adapt without fine-tuning the base LLM. Instead, it uses memory-augmented online reinforcement learning with episodic memory storage and a learned case-selection policy. Past experiences are stored in differentiable or non-parametric memory, and a neural policy selects relevant experiences to guide decision-making. This approach achieves rapid adaptation on research-oriented tasks while maintaining the benefits of frozen, pre-trained models.
Create abstraction for experience storage:
from dataclasses import dataclass
from typing import Dict, List, Any, Optional
import numpy as np
from collections import deque
@dataclass
class Experience:
"""Single experience stored in memory."""
state: str # Task context or observation
action: str # Agent action taken
observation: str # Result/feedback
reward: float # Task reward signal
success: bool # Was action successful?
timestamp: float
metadata: [, ] =
:
experiences: [Experience]
task_id:
success_rate:
last_accessed:
:
():
.experiences: deque = deque(maxlen=max_size)
.max_size = max_size
.similarity_metric = similarity_metric
.task_groups: [, [Experience]] = {}
():
.experiences.append(experience)
task_id = experience.metadata.get(, )
task_id .task_groups:
.task_groups[task_id] = []
.task_groups[task_id].append(experience)
() -> [[Experience, ]]:
candidates = .task_groups.get(filter_task) filter_task (.experiences)
candidates:
[]
similarities = []
exp candidates:
similarity = ._compute_similarity(query_state, exp.state)
similarities.append((exp, similarity))
similarities.sort(key= x: x[], reverse=)
similarities[:k]
() -> :
.similarity_metric == :
emb1 = ._encode(state1)
emb2 = ._encode(state2)
np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2) + )
.similarity_metric == :
set1 = (state1.split())
set2 = (state2.split())
intersection = (set1 & set2)
union = (set1 | set2)
intersection / (union + )
() -> np.ndarray:
() -> [, ]:
{
: (.experiences),
: (.task_groups),
: (.experiences) / ((.task_groups), ),
: (.experiences) / .max_size
}
Learn which memories to retrieve for decisions:
import torch
import torch.nn as nn
class CaseSelectionPolicy(nn.Module):
"""Policy for selecting relevant cases from memory."""
def __init__(
self,
embedding_dim: int = 768,
hidden_dim: int = 512,
max_cases: int = 5
):
super().__init__()
self.embedding_dim = embedding_dim
self.max_cases = max_cases
# Encoder for current state
self.state_encoder = nn.Sequential(
nn.Linear(embedding_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, embedding_dim)
)
# Relevance scorer for each candidate case
self.relevance_scorer = nn.Sequential(
nn.Linear(embedding_dim * 2, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, 1)
)
# Selection gate for whether to use memory at all
self.use_memory_gate = nn.Sequential(
nn.Linear(embedding_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, 1),
nn.Sigmoid()
)
def forward(
self,
current_state: torch.Tensor, # (embedding_dim,)
candidate_cases: List[torch.Tensor], # [(embedding_dim,), ...]
return_scores: bool = False
) -> torch.Tensor:
"""
Select top cases and return their indices/scores.
"""
# Encode current state
state_encoded = self.state_encoder(current_state)
# Compute relevance for each candidate
relevance_scores = []
for case in candidate_cases:
combined = torch.cat([state_encoded, case])
score = self.relevance_scorer(combined)
relevance_scores.append(score)
relevance_scores = torch.stack(relevance_scores).squeeze(-1)
# Decide whether to use memory
memory_use_prob = self.use_memory_gate(state_encoded)
# Select top-k
topk_scores, topk_indices = torch.topk(
relevance_scores,
k=min(self.max_cases, len(candidate_cases))
)
if return_scores:
return topk_indices, topk_scores
return topk_indices
def train_on_trajectory(
self,
states: List[torch.Tensor],
selected_cases: List[int],
outcomes: List[float],
gamma: float = 0.99
):
"""Train policy on trajectory data."""
# Compute returns
returns = []
G = 0
for outcome in reversed(outcomes):
G = outcome + gamma * G
returns.insert(0, G)
returns = torch.tensor(returns, dtype=torch.float32)
returns = (returns - returns.mean()) / (returns.std() + 1e-8)
# Policy gradient loss
total_loss = 0.0
for state, case_idx, ret in zip(states, selected_cases, returns):
# Get probability of selected case
with torch.no_grad():
logits = self.relevance_scorer(state)
# Compute policy gradient
log_prob = torch.log_softmax(logits, dim=-1)[case_idx]
loss = -log_prob * ret
total_loss += loss
return total_loss / len(states)
Integrate memory into decision process:
class MemoryAugmentedMDP:
"""Markov decision process with episodic memory."""
def __init__(
self,
base_model: "LLM",
memory: EpisodicMemory,
selection_policy: CaseSelectionPolicy,
max_memory_context_length: int = 2000
):
self.base_model = base_model
self.memory = memory
self.selection_policy = selection_policy
self.max_context_length = max_memory_context_length
def get_action(
self,
task_description: str,
current_observation: str,
use_memory: bool = True
) -> Tuple[str, Dict[str, Any]]:
"""
Select action considering memory context.
"""
# Retrieve relevant experiences from memory
relevant_experiences = []
if use_memory and len(self.memory.experiences) > 0:
relevant_experiences = self.memory.retrieve_similar(
current_observation,
k=5,
filter_task=task_description
)
# Build context for LLM
context = self._build_context(
task_description,
current_observation,
relevant_experiences
)
# Get action from LLM
prompt = f"{context}\n\nNext action:"
action = self.base_model.generate(prompt, max_tokens=100)
return action, {"context_length": len(context), "memories_used": len(relevant_experiences)}
def _build_context(
self,
task: str,
observation: str,
experiences: List[Tuple[Experience, float]]
) -> str:
"""Build LLM context including relevant memories."""
context = f"Task: {task}\n\nCurrent Observation: {observation}\n"
if experiences:
context += "\nRelevant Past Experiences:\n"
for i, (exp, score) in enumerate(experiences[:3]): # Top 3
context += f"{i+1}. Previous: {exp.state}\n"
context += f" Action: {exp.action}\n"
context += f" Result: {exp.observation}\n"
context += f" Success: {exp.success}\n"
return context
def collect_rollout(
self,
task_description: str,
max_steps: int = 20
) -> List[Experience]:
"""Collect trajectory using memory-augmented decisions."""
trajectory = []
observation = "Initial state"
for step in range(max_steps):
# Get action from MDP
action, metadata = self.get_action(task_description, observation)
# Execute action (simulated or real)
new_observation, reward, done = self._execute_action(
task_description,
action
)
# Store experience
experience = Experience(
state=observation,
action=action,
observation=new_observation,
reward=reward,
success=(reward > 0.5),
timestamp=0,
metadata={"task_id": task_description, **metadata}
)
trajectory.append(experience)
if done:
break
observation = new_observation
return trajectory
def _execute_action(
self,
task: str,
action: str
) -> Tuple[str, float, bool]:
"""Execute action and get feedback."""
# In practice: execute in real environment or simulator
pass
Update memory and policy without LLM fine-tuning:
class ContinualAdapter:
"""Handles continual learning without LLM updates."""
def __init__(
self,
mdp: MemoryAugmentedMDP,
selection_policy: CaseSelectionPolicy,
learning_rate: float = 1e-4
):
self.mdp = mdp
self.selection_policy = selection_policy
self.optimizer = torch.optim.Adam(
selection_policy.parameters(),
lr=learning_rate
)
def adapt_to_trajectory(self, trajectory: List[Experience]):
"""Update memory and policy based on new trajectory."""
# Store all experiences in memory
for exp in trajectory:
self.mdp.memory.store_experience(exp)
# Train selection policy on successful outcomes
successful_trajectory = [e for e in trajectory if e.success]
if successful_trajectory:
self._train_selection_policy(successful_trajectory)
# Clean up memory if needed
if len(self.mdp.memory.experiences) > self.mdp.memory.max_size * 0.9:
self._prune_memory()
def _train_selection_policy(self, trajectory: List[Experience]):
"""Train policy on successful trajectory."""
# Encode states
states = [self._encode_state(e.state) for e in trajectory]
outcomes = [e.reward for e in trajectory]
# Get selected cases (which ones would the policy choose?)
selected_cases = []
for state in states:
candidates = self._get_candidate_embeddings(state)
indices = self.selection_policy(
state, candidates, return_scores=False
)
selected_cases.append(indices[0].item())
# Compute policy gradient loss
loss = self.selection_policy.train_on_trajectory(
states, selected_cases, outcomes
)
# Update policy
self.optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.selection_policy.parameters(), 1.0)
self.optimizer.step()
return loss.item()
def _prune_memory(self):
"""Remove least useful experiences."""
# Keep only high-success experiences or recently accessed
pass
def _encode_state(self, state: str) -> torch.Tensor:
"""Encode state to embedding."""
pass
def _get_candidate_embeddings(self, state_embedding) -> List[torch.Tensor]:
"""Get embeddings of candidate memory cases."""
pass
Measure adaptation efficiency without LLM fine-tuning:
def evaluate_agentfly_learning(
mdp: MemoryAugmentedMDP,
test_tasks: List[Dict],
num_rollouts_per_task: int = 3
) -> Dict[str, float]:
"""
Evaluate memory-augmented agent performance.
"""
task_successes = []
memory_hit_rates = []
trajectory_lengths = []
for task in test_tasks:
task_desc = task["description"]
expected_solution = task["expected"]
for _ in range(num_rollouts_per_task):
# Collect trajectory
trajectory = mdp.collect_rollout(task_desc, max_steps=50)
# Evaluate success
final_obs = trajectory[-1].observation if trajectory else ""
success = evaluate_task_success(final_obs, expected_solution)
task_successes.append(success)
# Track memory efficiency
memory_hits = sum(
1 for exp in trajectory
if exp.metadata.get("memories_used", 0) > 0
)
memory_hit_rates.append(memory_hits / len(trajectory))
trajectory_lengths.append(len(trajectory))
# Adapt to this trajectory
mdp_adapter = ContinualAdapter(mdp, mdp.selection_policy)
mdp_adapter.adapt_to_trajectory(trajectory)
return {
"avg_success_rate": sum(task_successes) / len(task_successes),
"avg_memory_hit_rate": sum(memory_hit_rates) / len(memory_hit_rates),
"avg_trajectory_length": sum(trajectory_lengths) / len(trajectory_lengths),
"memory_size": len(mdp.memory.experiences)
}
Researchers. (2024). AgentFly: Fine-tuning LLM Agents without Fine-tuning LLMs. arXiv preprint arXiv:2508.16153.
基于 SOC 职业分类
