awesome-long-horizon-agents-survey

# Awesome Long-Horizon Agents Survey > Skill by [ara.so](https://ara.so) — AI Agent Skills collection. This skill helps you navigate and utilize the **RUC-NLPIR/Awesome-Long-Horizon-Agents** repository, a comprehensive survey and curated reading list covering the landscape of long-horizon AI agents. The resource organizes research into two main pillars: **externalized harness engineering** (loops, memory, tools, orchestration) and **internalized model optimization** (training, RL, self-evolution). ## Overview The repository accompanies the paper ["Towards Long-Horizon Agents: A Survey"](https://openreview.net/pdf?id=HyhfhlbWGh) and provides: - **Three-level formalization** (H1→H2→H3) of long-horizon tasks - **Evolution timeline** from prompt engineering (2020-2023) to runtime harnesses (2025+) - **Systematic taxonomy** of agent capabilities and implementation patterns - **Curated paper lists** organized by technique and application domain - **Benchmarks and evaluation frameworks** for long-horizon tasks ### Key Concept: Long-Horizon Defined The survey defines long-horizon agents across three nested levels: | Level | Horizon | Capability Required | |-------|---------|---------------------| | **H1** | Intra-context (~minutes) | Interactive reasoning within one context window | | **H2** | Cross-context (~hours-days) | State persistence and memory across sessions | | **H3** | Cross-task (open-ended) | Experience accumulation and self-evolution | **Agent Formula**: `Agent = π_θ ⊕ H` (base policy + harness) ## Installation & Access ### Clone the Repository ```bash git clone https://github.com/RUC-NLPIR/Awesome-Long-Horizon-Agents.git cd Awesome-Long-Horizon-Agents ``` ### Access the Resources The repository is primarily a curated markdown document with links to papers, not executable code. Use it as: 1. **Reference guide** for researching agent techniques 2. **Reading list** organized by capability area 3. **Taxonomy** for categorizing your own agent implementations ### Quick Links - **Paper**: [OpenReview](https://openreview.net/pdf?id=HyhfhlbWGh) | [Preprint](https://www.preprints.org/manuscript/202607.1328) - **Website**: [Long-Horizon-Agents.github.io](https://Long-Horizon-Agents.github.io) - **Discussion**: [X Thread](https://x.com/kakakbibibi/status/2078076130037514640) ## Key Research Areas & Paper Collections ### 1. Foundations: Formalizing Agents **Core Papers**: - **ReAct** (ICLR 2023): Synergizing reasoning and acting - [arxiv:2210.03629](https://arxiv.org/abs/2210.03629) - **Chain-of-Thought** (NeurIPS 2022): Multi-step reasoning - [arxiv:2201.11903](https://arxiv.org/abs/2201.11903) - **Tree of Thoughts** (NeurIPS 2023): Deliberate problem solving - [arxiv:2305.10601](https://arxiv.org/abs/2305.10601) **Use Case**: Understanding foundational agent reasoning patterns ```python # Example: Implementing ReAct pattern class ReActAgent: def __init__(self, llm, tools): self.llm = llm self.tools = tools def run(self, task): thought = self.llm.generate(f"Thought: {task}") action = self.parse_action(thought) observation = self.tools[action['name']].execute(action['args']) # Iterate: Thought → Action → Observation return self.llm.generate(f"Based on {observation}, the answer is...") def parse_action(self, thought): # Extract action from "Action: tool_name[arg1, arg2]" return {'name': 'search', 'args': ['query']} ``` ### 2. Harnesses: External Capabilities (Pillar I) #### Loops and Workflows **Key Papers**: - **Reflexion** - Self-reflection for iterative improvement - **Voyager** - Curriculum learning in open worlds - **AutoGPT** - Autonomous task execution loops **Pattern**: Agent control flow architecture ```python # Example: Multi-agent orchestration harness class AgentHarness: def __init__(self, agents: dict, memory: Memory): self.agents = agents self.memory = memory async def execute_workflow(self, task): """H2-level: Cross-context workflow""" # 1. Planning phase plan = await self.agents['planner'].decompose(task) self.memory.save('plan', plan) # 2. Execution loop for step in plan.steps: context = self.memory.retrieve_relevant(step) result = await self.agents['executor'].run(step, context) self.memory.save(f'step_{step.id}', result) # Verification hook if not self.verify(result): result = await self.agents['refiner'].fix(result, context) # 3. Synthesis return self.agents['synthesizer'].combine(self.memory.get_all()) def verify(self, result): return result.confidence > 0.8 ``` #### Context and Memory **Key Papers**: - **RAG** (NeurIPS 2020): Retrieval-Augmented Generation - **RAPTOR** (ICLR 2024): Tree-organized hierarchical retrieval - **MemGPT**: Operating system for LLM memory management **Pattern**: Long-term memory systems ```python # Example: Hierarchical memory with RAPTOR-style retrieval from sentence_transformers import SentenceTransformer class HierarchicalMemory: def __init__(self, embedding_model='all-MiniLM-L6-v2'): self.encoder = SentenceTransformer(embedding_model) self.short_term = [] # Recent context self.long_term = {} # Episodic memory self.semantic = {} # Indexed by topic clusters def add(self, text, metadata=None): """Add to short-term, periodically consolidate""" embedding = self.encoder.encode(text) self.short_term.append({ 'text': text, 'embedding': embedding, 'metadata': metadata, 'timestamp': time.time() }) if len(self.short_term) > 100: self.consolidate() def retrieve_relevant(self, query, k=5): """H2 capability: Cross-context retrieval""" query_emb = self.encoder.encode(query) # Search across both short and long-term all_memories = self.short_term + list(self.long_term.values()) scores = [cosine_similarity(query_emb, m['embedding']) for m in all_memories] top_k = sorted(zip(scores, all_memories), reverse=True)[:k] return [m['text'] for _, m in top_k] def consolidate(self): """Move short-term to long-term with summarization""" summary = self.summarize_cluster(self.short_term) self.long_term[summary['id']] = summary self.short_term = [] ``` #### Tools, MCP, and Skills **Model Context Protocol (MCP)**: Standard for agent-tool integration ```python # Example: MCP-compliant tool integration from typing import Protocol class MCPTool(Protocol): """Standard interface for agent tools""" name: str description: str def get_schema(self) -> dict: """Return JSON schema for tool parameters""" ... async def execute(self, **kwargs) -> dict: """Execute tool and return structured result""" ... class WebSearchTool: name = "web_search" description = "Search the web for current information" def get_schema(self): return { "type": "object", "properties": { "query": {"type": "string"}, "num_results": {"type": "integer", "default": 5} }, "required": ["query"] } async def execute(self, query: str, num_results: int = 5): # Integration with search API import os api_key = os.getenv('SEARCH_API_KEY') results = await search_api.query(query, limit=num_results) return {"results": results, "count": len(results)} # Agent uses tools via MCP class MCPAgent: def __init__(self, llm, tools: list[MCPTool]): self.llm = llm self.tools = {t.name: t for t in tools} async def run(self, task): # LLM selects tool based on schemas tool_schemas = {n: t.get_schema() for n, t in self.tools.items()} decision = self.llm.select_tool(task, tool_schemas) # Execute via MCP interface result = await self.tools[decision['tool']].execute(**decision['args']) return self.llm.synthesize(task, result) ``` #### Verification **Key Papers**: - **Self-Consistency**: Multiple sampling for verification - **Critic models**: Learned verification functions - **Process supervision**: Step-by-step correctness checking ```python # Example: Multi-stage verification class VerificationHarness: def __init__(self, critic_model, test_suite): self.critic = critic_model self.tests = test_suite async def verify_with_feedback(self, agent_output, task): """Multi-level verification with repair loop""" # 1. Syntax/format check if not self.validate_format(agent_output): return {'valid': False, 'feedback': 'Format error'} # 2. Semantic correctness (critic model) critique = await self.critic.evaluate(agent_output, task) if critique.score < 0.7: return {'valid': False, 'feedback': critique.reasoning} # 3. Execution tests (for code generation tasks) if self.tests: test_results = self.tests.run(agent_output) if not test_results.all_passed(): return { 'valid': False, 'feedback': f'Failed tests: {test_results.failures}' } return {'valid': True} def validate_format(self, output): # Check structure, required fields, etc. return True ``` ### 3. Optimization: Internal Capabilities (Pillar II) #### Agentic Reinforcement Learning **Key Papers**: - **AgentQ**: Online RL for web agents - **Agent-FLAN**: Multi-task instruction tuning - **Reflexion**: Reinforcement via verbal feedback **Pattern**: Training agents with trajectory feedback ```python # Conceptual example: RL training loop for agents class AgentRLTrainer: def __init__(self, base_model, environment, reward_model): self.policy = base_model self.env = environment self.reward_model = reward_model def train_episode(self, task): """Single training episode with trajectory collection""" trajectory = [] state = self.env.reset(task) for step in range(max_steps): # Agent takes action action = self.policy.generate_action(state) next_state, env_reward = self.env.step(action) # Compute learned reward (outcome + process) reward = self.reward_model.score( state=state, action=action, outcome=next_state, success=env_reward ) trajectory.append({ 'state': state, 'action': action, 'reward': reward }) if self.env.is_done(): break state = next_state # Update policy using PPO/DPO self.update_policy(trajectory) return sum(t['reward'] for t in trajectory) def update_policy(self, trajectory): """Update using preference optimization""" # Implement DPO, PPO, or GRPO update pass ``` #### Self-Evolution **Key Papers**: - **Voyager**: Skill library via self-play - **AutoGPT**: Autonomous capability expansion - **Self-Instruct**: Bootstrap via self-generated data **Pattern**: H3-level cross-task learning ```python # Example: Skill library with self-evolution class EvolvingAgent: def __init__(self, base_model): self.model = base_model self.skill_library = {} self.experience_buffer = [] def execute_and_learn(self, task): """H3: Learn from task execution""" # Try to use existing skills relevant_skills = self.match_skills(task) result = self.model.run( task, context=relevant_skills, exploration=True ) # Store successful patterns if result.success: self.experience_buffer.append({ 'task': task, 'solution': result.trajectory, 'performance': result.metrics }) # Periodically distill into reusable skills if len(self.experience_buffer) > 100: self.evolve_skills() return result def evolve_skills(self):