gaia-experience¶
第8章 · Agent 的自我进化 · 配套项目
chapter8/gaia-experience
项目说明¶
GAIA Experience Learning System¶
对应《深入理解 AI Agent》第 8 章 · 实验 8-1 ★★「从成功经验中学习：策略摘要」。 本目录是实验的顶层封装脚本；
AWorld/为上游框架副本，请勿修改。
A modified version of AWorld that adds learning from experience capabilities for the GAIA benchmark. This system can capture successful task trajectories, summarize them into reusable experiences, and apply learned knowledge to improve performance on new tasks.
实验要点（why this matters）：GAIA 是需要多步推理、综合使用浏览器/文件/代码解释器的高难度基准。
本实验演示一个「学习-应用」闭环——Agent 每成功解一题就把轨迹提炼为经验入库，遇到新题时检索相似经验注入提示词。
命题是：复用积累的经验能提升 GAIA 成绩。使用 --compare 可以在同一批题上做 A/B 对照来直观检验这一命题（见下文
A/B 对照实验）。
🌟 Features¶
1. Learning from Experience¶
- Captures the actual execution trajectory produced by AWorld (read back from
TaskResponse.trajectory) when a task completes successfully - Uses LLM to summarize trajectories into natural language experiences
- Stores learned experiences (approach, key insights, tools, step count) for future reference
2. Knowledge Base with Semantic Search¶
- Indexes the
gaia-validation.jsonlfile for preloaded experiences - Uses sentence embeddings for semantic similarity search
- Supports both semantic and keyword-based retrieval
3. Experience Application¶
- Retrieves relevant past experiences for new queries
- Enhances system prompts with relevant experiences
- Improves task performance by leveraging past successes
📁 Project Structure¶
gaia-experience/
├── AWorld/ # Original AWorld repository
├── experience_agent.py # Extended agent with experience learning
├── knowledge_base.py # Knowledge base for indexing and retrieval
├── trajectory_summarizer.py # Summarizes trajectories into experiences
├── run_with_experience.py # Main execution script with learning features
├── demo.py # Demo script showcasing all features
├── config.yaml # Configuration file
├── requirements.txt # Python dependencies
├── gaia-validation.jsonl # GAIA validation dataset
└── README.md # This file
🚀 Installation¶
Prerequisites¶
- Python 3.8+
- Conda (recommended for environment management)
- Node.js 22 LTS (for AWorld web UI)
Setup Steps¶
-
Clone and setup the environment:
-
Setup AWorld (if not already done):
-
Configure environment variables: Create a
.envfile:# LLM Configuration LLM_PROVIDER=openai LLM_MODEL_NAME=gpt-5.6-luna LLM_API_KEY=your_api_key_here LLM_BASE_URL=https://api.openai.com/v1 # Optional # 兜底：若 LLM_API_KEY/OPENAI_API_KEY 都缺失但设了 OPENROUTER_API_KEY，自动走 OpenRouter（映射到 openai/gpt-5.6-luna 等） # OPENROUTER_API_KEY=sk-or-... # Dataset paths GAIA_DATASET_PATH=./AWorld/examples/gaia/GAIA AWORLD_WORKSPACE=./workspace
💡 Usage¶
Basic Usage¶
1. Run with Learning Mode¶
Capture and learn from successful trajectories:
2. Run with Experience Application¶
Apply learned experiences to new tasks:
python run_with_experience.py \
--apply-experience \
--preload-kb \
--start 5 --end 10 \
--split validation
3. Combined Mode¶
Learn and apply experiences simultaneously:
python run_with_experience.py \
--learning-mode \
--apply-experience \
--preload-kb \
--start 0 --end 10
4. A/B Comparison Mode¶
Evaluate the same tasks twice — once without experience, once with — and report the delta:
python run_with_experience.py --compare --start 10 --end 20 \
--experience-db ./learned_experiences.json
The full Chinese
--help(with runnable examples) is always available without installing the heavy stack:python run_with_experience.py --help.
Command Line Arguments¶
| Argument | Description | Default |
|---|---|---|
--learning-mode |
Enable learning from successful trajectories | False |
--apply-experience |
Apply learned experiences to new tasks | False |
--compare |
A/B mode: run the slice with and without experience, report accuracy delta | False |
--preload-kb |
Preload knowledge base from gaia-validation.jsonl (can leak answers, see below) | False |
--kb-path |
Path to store knowledge base index | ./kb_index |
--experience-db |
Path to store learned experiences | ./learned_experiences.json |
--validation-file |
Path to gaia-validation.jsonl | gaia-validation.jsonl |
--embedding-model |
Sentence transformer model | all-MiniLM-L6-v2 |
--summary-model |
Model for trajectory summarization | gpt-5.6-luna |
--model |
Main agent model (overrides LLM_MODEL_NAME) |
env / gpt-5.6-luna |
--output |
Results JSON output path | $AWORLD_WORKSPACE/experience_results.json |
--start |
Start index of dataset | 0 |
--end |
End index of dataset | 20 |
--q |
Specific task ID to run | None |
--skip |
Skip tasks already present in the results file | False |
--split |
Dataset split (validation/test) | validation |
--blacklist_file_path |
Optional file of task_ids to skip | None |
Demo Script¶
Run the interactive demo to explore all features:
# Run complete workflow demo
python demo.py
# Interactive mode
python demo.py --interactive
# Specific demos
python demo.py --kb # Knowledge base demo
python demo.py --summarize # Summarization demo
python demo.py --agent # Agent demo
🔧 Configuration¶
The config.yaml file provides detailed configuration options:
Key Configuration Sections:¶
- Learning Settings
- Summarizer model and temperature
- Experience storage settings
-
Maximum trajectory steps
-
Knowledge Base Settings
- Embedding model configuration
- Search parameters (top-k, similarity threshold)
-
Index storage path
-
Application Settings
- Experience incorporation strategy
- Filtering criteria (by level, tools, recency)
- Maximum experiences in prompt
📊 How It Works¶
Learning Process¶
- Trajectory Capture: After a task runs, the actual step-by-step trajectory (tools, actions, params) is read back from AWorld's
TaskResponse.trajectoryand normalized for the summarizer - Success Detection: When a task completes successfully, the trajectory is marked for learning
- Summarization: The trajectory is analyzed by an LLM to extract:
- High-level approach
- Key insights and patterns
- Tools used effectively
- Generalizable strategies
- Storage: The summarized experience is stored with the original question for retrieval
Retrieval and Application¶
- Query Analysis: New questions are analyzed for similarity to past experiences
- Semantic Search: The knowledge base finds relevant experiences using embeddings
- Experience Selection: Top-k most relevant experiences are selected
- Prompt Enhancement: Selected experiences are formatted and added to the system prompt
- Execution: The agent solves the task with the benefit of past experiences
Knowledge Base Preloading¶
The system can preload the gaia-validation.jsonl file to bootstrap the knowledge base:
- Each entry is parsed for question, approach, and tools used
- Experiences are indexed using sentence embeddings
- Enables immediate access to a rich set of problem-solving patterns
📈 Performance Benefits¶
- Improved Success Rate: Agents learn from past successes to avoid repeating mistakes
- Faster Problem Solving: Relevant experiences guide the agent to efficient solutions
- Knowledge Transfer: Experiences from similar problems apply to new challenges
- Reduced Token Usage: Past insights can reduce exploration and trial-and-error
🧪 A/B 对照实验（reproduce the experiment's point）¶
本实验的核心命题是「复用积累的经验能提升 GAIA 成绩」。--compare 模式把这个命题变成一个
可运行、可复现的对照实验：对同一批题目跑两遍——一遍关闭经验复用（baseline），一遍打开
经验复用（with-experience）——并报告两者的准确率之差（delta）。
推荐工作流（避免数据泄漏）： 先在一批题上积累经验，再在另一批未见过的题上做对照。
直接对评测题 --preload-kb 会把这些题在 gaia-validation.jsonl 中的参考解法灌入知识库，
等于泄漏答案；脚本检测到该情况会打印告警。
# 1) 在第 0~10 题上积累经验（仅从成功轨迹中学习）
python run_with_experience.py --learning-mode --start 0 --end 10
# 2) 在第 10~20 题（未见过）上做 A/B 对照，复用第 1 步学到的经验
python run_with_experience.py --compare --start 10 --end 20 \
--experience-db ./learned_experiences.json
# 或： ./run.sh learn --start 0 --end 10 && ./run.sh compare --start 10 --end 20
输出：控制台打印如下汇总，同时把每题明细写入 comparison_results.json（或 --output 指定路径）：
============================================================
A/B COMPARISON: experience reuse vs. baseline
============================================================
Tasks evaluated : <N> (split=validation, range=[10, 20))
Reusable experiences : <M> learned, <K> preloaded
Baseline accuracy : <c1>/<N> = <p1>%
With-experience acc. : <c2>/<N> = <p2>%
Delta (with - base) : <p2 - p1>%
============================================================
预期结果：当经验库中确有与评测题相关的可迁移经验时，with-experience 的准确率应 ≥ baseline，
delta 为正——这正是「学习-应用闭环」带来的增益。所有数字均由 question_scorer 对真实运行结果计算得到，
脚本不写死任何成绩；若经验库为空或经验不相关，delta 可能为 0，属于正常现象（说明还没有可复用的相关经验）。
⚠️ 运行完整对照需要：可用的 LLAPI（
.env中的LLM_API_KEY等）、已安装的 AWorld（pip install -e ./AWorld）、sentence-transformers/faiss-cpu依赖，以及 GAIA 数据集（GAIA_DATASET_PATH）。 若只想确认命令行可用，无需上述环境即可运行python run_with_experience.py --help。
🔍 Example Experience¶
{
"question": "Find AI regulation paper from June 2022 on arXiv",
"answer": "egalitarian",
"summary": "Successfully located paper by using advanced search with date filters",
"approach": "Web search → Navigate to arXiv → Use advanced search → Filter by date",
"tools_used": ["web_search", "browser_navigate", "browser_click"],
"key_insights": [
"Advanced search provides better filtering options",
"Date range queries need specific format",
"Multiple searches refined the query"
]
}
🛠️ Extending the System¶
Adding Custom Summarizers¶
Create a new summarizer by extending TrajectorySummarizer:
class CustomSummarizer(TrajectorySummarizer):
def _create_summary_prompt(self, question, answer, trajectory):
# Custom prompt logic
pass
Custom Experience Filters¶
Add filtering logic in ExperienceAgent._get_relevant_experiences():
Alternative Embedding Models¶
Change the embedding model in configuration:
knowledge_base:
index:
embedding_model: "all-mpnet-base-v2" # Higher quality embeddings
embedding_dim: 768 # Adjust dimension accordingly
📝 Notes¶
- The system requires API access to an LLM for summarization
- Embedding models are downloaded on first use (may take time)
- Knowledge base indexing is incremental and persistent
- Experiences are saved after each successful task
🤝 Contributing¶
Feel free to extend and improve the system: - Add more sophisticated trajectory analysis - Implement experience merging and evolution - Add multi-agent experience sharing - Enhance retrieval with hybrid search methods
📄 License¶
This project extends AWorld and follows its licensing terms.
🙏 Acknowledgments¶
- Built on top of the AWorld framework by inclusionAI
- Uses the GAIA benchmark for evaluation
- Leverages sentence-transformers for semantic search
源代码¶
demo.py¶
"""
Demo script for Experience Learning System
This script demonstrates all features of the experience learning system.
"""
import asyncio
import json
import logging
import os
import sys
from pathlib import Path
from typing import Dict, Any, List
import yaml
from dotenv import load_dotenv
# Add parent directory to path
sys.path.append(str(Path(__file__).parent))
from experience_agent import ExperienceAgent
from knowledge_base import KnowledgeBase
from trajectory_summarizer import TrajectorySummarizer
from llm_env import resolve_llm, DEFAULT_MODEL
from AWorld.aworld.config.conf import AgentConfig, TaskConfig
from AWorld.aworld.core.task import Task
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class ExperienceLearningDemo:
"""Demo class for experience learning system."""
def __init__(self, config_path: str = "config.yaml"):
"""Initialize demo with configuration."""
self.config = self.load_config(config_path)
self.setup_environment()
self.knowledge_base = None
self.summarizer = None
self.agent = None
def load_config(self, config_path: str) -> Dict[str, Any]:
"""Load configuration from YAML file."""
if os.path.exists(config_path):
with open(config_path, 'r') as f:
return yaml.safe_load(f)
else:
logger.warning(f"Config file {config_path} not found, using defaults")
return {}
def setup_environment(self):
"""Setup environment and directories."""
load_dotenv()
# Create necessary directories
os.makedirs("./logs", exist_ok=True)
os.makedirs("./kb_index", exist_ok=True)
os.makedirs("./experiences", exist_ok=True)
async def demo_knowledge_base_indexing(self):
"""Demonstrate knowledge base indexing."""
print("\n" + "="*60)
print("DEMO 1: Knowledge Base Indexing")
print("="*60)
# Initialize knowledge base
kb_config = self.config.get('knowledge_base', {})
self.knowledge_base = KnowledgeBase(
index_path=kb_config.get('index', {}).get('path', './kb_index'),
model_name=kb_config.get('index', {}).get('embedding_model', 'all-MiniLM-L6-v2')
)
# Check if validation file exists
validation_file = self.config.get('dataset', {}).get('validation_file', 'gaia-validation.jsonl')
if os.path.exists(validation_file):
print(f"\n📚 Indexing GAIA validation data from {validation_file}...")
self.knowledge_base.index_gaia_validation(validation_file)
# Show statistics
stats = self.knowledge_base.get_statistics()
print(f"\n✅ Indexing complete!")
print(f" - Total documents: {stats['total_documents']}")
print(f" - Has embeddings: {stats['has_embeddings']}")
print(f" - Sources: {stats['sources']}")
# Demo search
test_query = "How to find information about scientific papers on arXiv?"
print(f"\n🔍 Testing search with query: '{test_query}'")
results = self.knowledge_base.search(test_query, top_k=3)
print(f"\n📋 Found {len(results)} relevant experiences:")
for i, result in enumerate(results, 1):
print(f"\n {i}. Question: {result.get('question', 'N/A')[:100]}...")
print(f" Approach: {result.get('approach', 'N/A')[:100]}...")
if result.get('tools_used'):
print(f" Tools: {', '.join(result['tools_used'][:3])}")
else:
print(f"\n⚠️ Validation file not found: {validation_file}")
print(" Creating synthetic experiences for demo...")
# Add synthetic experiences
synthetic_experiences = [
{
'question': "How to search for papers on arXiv?",
'approach': "Use web search to find arXiv, then use their search functionality",
'tools_used': ["web_browser", "search_engine"],
'answer': "Navigate to arxiv.org and use the search bar"
},
{
'question': "Calculate the distance between Earth and Moon",
'approach': "Search for astronomical data and perform calculations",
'tools_used': ["calculator", "web_search"],
'answer': "384,400 km average distance"
}
]
for exp in synthetic_experiences:
self.knowledge_base.add_experience(exp['question'], exp)
print(f"✅ Added {len(synthetic_experiences)} synthetic experiences")
async def demo_trajectory_summarization(self):
"""Demonstrate trajectory summarization."""
print("\n" + "="*60)
print("DEMO 2: Trajectory Summarization")
print("="*60)
# Initialize summarizer（OpenAI 直连，缺 Key 时 OpenRouter 兜底）
summarizer_model = self.config.get('learning', {}).get('summarizer', {}).get('model', DEFAULT_MODEL)
llm_kwargs = resolve_llm(model_override=summarizer_model)
agent_config = AgentConfig(**llm_kwargs)
self.summarizer = TrajectorySummarizer(
llm_config=agent_config,
model_name=llm_kwargs["llm_model_name"]
)
# Create sample trajectory
sample_trajectory = [
{
'action': {
'tool_name': 'web_search',
'params': {'query': 'arxiv.org AI regulation 2022'}
}
},
{
'action': {
'tool_name': 'browser_navigate',
'params': {'url': 'https://arxiv.org/search'}
}
},
{
'action': {
'tool_name': 'browser_click',
'params': {'element': 'advanced_search'}
}
},
{
'action': {
'tool_name': 'browser_fill',
'params': {'field': 'date_range', 'value': '2022-06-01 to 2022-07-01'}
}
}
]
# Create mock response
class MockResponse:
def __init__(self):
self.answer = "The paper shows a figure with three axes labeled: deontological, egalitarian, utilitarian"
print("\n📝 Summarizing sample trajectory...")
print(f" Trajectory has {len(sample_trajectory)} steps")
summary = await self.summarizer.summarize(
question="Find AI regulation paper from June 2022 on arXiv",
response=MockResponse(),
trajectory=sample_trajectory
)
print("\n✅ Summary generated:")
print(f" - Summary: {summary.get('summary', 'N/A')}")
print(f" - Approach: {summary.get('approach', 'N/A')}")
print(f" - Tools used: {', '.join(summary.get('tools_used', []))}")
if summary.get('key_insights'):
print(f" - Key insights:")
for insight in summary['key_insights']:
print(f" • {insight}")
async def demo_experience_agent(self):
"""Demonstrate the experience agent."""
print("\n" + "="*60)
print("DEMO 3: Experience Agent")
print("="*60)
# Initialize agent with all features（OpenAI 直连，缺 Key 时 OpenRouter 兜底）
agent_config = AgentConfig(
**resolve_llm(),
llm_temperature=0.0
)
# Basic system prompt
system_prompt = """You are an intelligent agent capable of learning from experience.
When solving problems, you can leverage past experiences to find better solutions.
Always provide your answer in the format: <answer>YOUR_ANSWER</answer>"""
self.agent = ExperienceAgent(
conf=agent_config,
name="demo_experience_agent",
system_prompt=system_prompt,
learning_mode=True,
apply_experience=True,
experience_db_path="./experiences/demo_experiences.json",
knowledge_base=self.knowledge_base,
summarizer=self.summarizer
)
print("\n🤖 Experience Agent initialized with:")
print(f" - Learning mode: {self.agent.learning_mode}")
print(f" - Apply experience: {self.agent.apply_experience}")
print(f" - Existing experiences: {len(self.agent.experiences)}")
# Test questions
test_questions = [
"What is the capital of France?",
"How many days are there in February during a leap year?"
]
for i, question in enumerate(test_questions, 1):
print(f"\n📌 Test Question {i}: {question}")
# Check for relevant experiences
relevant_exp = self.agent._get_relevant_experiences(question)
if relevant_exp:
print(f" Found {len(relevant_exp)} relevant experiences")
else:
print(" No relevant experiences found")
# Create task
task = Task(
input=question,
agent=self.agent,
conf=TaskConfig()
)
try:
# Execute task
print(" Executing task...")
response = await self.agent.execute_task(task)
if response and response.answer:
print(f" ✅ Answer: {response.answer[:100]}...")
else:
print(" ❌ No answer generated")
except Exception as e:
print(f" ❌ Error: {e}")
# Show learned experiences
if self.agent.experiences:
print(f"\n📚 Learned Experiences ({len(self.agent.experiences)} total):")
for exp_id, exp in list(self.agent.experiences.items())[:3]:
print(f"\n Experience ID: {exp_id[:8]}...")
print(f" Question: {exp.get('question', 'N/A')[:80]}...")
print(f" Summary: {exp.get('summary', 'N/A')[:80]}...")
async def demo_workflow(self):
"""Demonstrate complete workflow."""
print("\n" + "="*60)
print("COMPLETE EXPERIENCE LEARNING WORKFLOW DEMO")
print("="*60)
# Step 1: Index knowledge base
await self.demo_knowledge_base_indexing()
# Step 2: Setup summarizer
await self.demo_trajectory_summarization()
# Step 3: Run agent with experience learning
await self.demo_experience_agent()
print("\n" + "="*60)
print("DEMO COMPLETE!")
print("="*60)
# Summary statistics
if self.knowledge_base:
kb_stats = self.knowledge_base.get_statistics()
print(f"\n📊 Final Statistics:")
print(f" - Knowledge base documents: {kb_stats['total_documents']}")
if self.agent:
print(f" - Learned experiences: {len(self.agent.experiences)}")
print(f" - Experience DB: {self.agent.experience_db_path}")
def run_interactive_mode(self):
"""Run interactive mode for testing."""
print("\n" + "="*60)
print("INTERACTIVE MODE")
print("="*60)
print("\nCommands:")
print(" 1. Index knowledge base")
print(" 2. Test trajectory summarization")
print(" 3. Run experience agent")
print(" 4. Run complete workflow")
print(" 5. Exit")
while True:
try:
choice = input("\nEnter command (1-5): ").strip()
if choice == "1":
asyncio.run(self.demo_knowledge_base_indexing())
elif choice == "2":
asyncio.run(self.demo_trajectory_summarization())
elif choice == "3":
asyncio.run(self.demo_experience_agent())
elif choice == "4":
asyncio.run(self.demo_workflow())
elif choice == "5":
print("\nGoodbye!")
break
else:
print("Invalid choice. Please enter 1-5.")
except KeyboardInterrupt:
print("\n\nInterrupted by user.")
break
except Exception as e:
logger.error(f"Error in interactive mode: {e}")
print(f"\n❌ Error: {e}")
async def main():
"""Main entry point for demo."""
demo = ExperienceLearningDemo()
# Check for command line arguments
import sys
if len(sys.argv) > 1:
if sys.argv[1] == "--interactive":
demo.run_interactive_mode()
elif sys.argv[1] == "--kb":
await demo.demo_knowledge_base_indexing()
elif sys.argv[1] == "--summarize":
await demo.demo_trajectory_summarization()
elif sys.argv[1] == "--agent":
await demo.demo_experience_agent()
else:
await demo.demo_workflow()
else:
# Run complete workflow by default
await demo.demo_workflow()
if __name__ == "__main__":
print("\n🚀 Experience Learning System Demo")
print("=" * 60)
# Run the demo
asyncio.run(main())
experience_agent.py¶
"""
Experience Learning Agent for GAIA
This module extends the AWorld Agent with learning from experience capabilities.
"""
import json
import logging
import os
from typing import Dict, Any, List, Optional
from datetime import datetime
import hashlib
from AWorld.aworld.agents.llm_agent import Agent
from AWorld.aworld.config.conf import AgentConfig
from AWorld.aworld.core.task import Task, TaskResponse
from AWorld.aworld.runner import Runners
logger = logging.getLogger(__name__)
class ExperienceAgent(Agent):
"""
Extended Agent that can learn from successful trajectories and apply learned experiences.
"""
def __init__(
self,
conf: AgentConfig,
name: str = "experience_agent",
system_prompt: str = "",
learning_mode: bool = False,
apply_experience: bool = False,
experience_db_path: str = "./experience_db.json",
knowledge_base: Optional['KnowledgeBase'] = None,
summarizer: Optional['TrajectorySummarizer'] = None,
**kwargs
):
"""
Initialize the Experience Agent.
Args:
conf: Agent configuration
name: Agent name
system_prompt: Base system prompt
learning_mode: Whether to capture and learn from successful trajectories
apply_experience: Whether to apply learned experiences to new tasks
experience_db_path: Path to store learned experiences
knowledge_base: Knowledge base for retrieval
summarizer: Trajectory summarizer instance
**kwargs: Additional arguments for base Agent
"""
super().__init__(conf=conf, name=name, system_prompt=system_prompt, **kwargs)
self.learning_mode = learning_mode
self.apply_experience = apply_experience
self.experience_db_path = experience_db_path
self.knowledge_base = knowledge_base
self.summarizer = summarizer
self.base_system_prompt = system_prompt
# Load existing experiences if available
self.experiences = self._load_experiences()
# Track current task trajectory
self.current_trajectory = []
def _load_experiences(self) -> Dict[str, Any]:
"""Load existing experiences from file."""
if os.path.exists(self.experience_db_path):
try:
with open(self.experience_db_path, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
logger.error(f"Failed to load experiences: {e}")
return {}
def _save_experiences(self):
"""Save experiences to file."""
try:
with open(self.experience_db_path, 'w', encoding='utf-8') as f:
json.dump(self.experiences, f, indent=2, ensure_ascii=False)
except Exception as e:
logger.error(f"Failed to save experiences: {e}")
def _get_task_hash(self, question: str) -> str:
"""Generate a hash for a question to use as key."""
return hashlib.md5(question.encode()).hexdigest()
async def execute_task(self, task: Task) -> TaskResponse:
"""
Execute a task with experience learning/application.
Args:
task: The task to execute
Returns:
TaskResponse with result
"""
question = task.input
original_prompt = self.system_prompt
# Apply experience if enabled
if self.apply_experience:
relevant_experiences = self._get_relevant_experiences(question)
if relevant_experiences:
experience_text = self._format_experiences(relevant_experiences)
self.system_prompt = f"{self.base_system_prompt}\n\n# Relevant Past Experiences:\n{experience_text}"
logger.info(f"Applied {len(relevant_experiences)} relevant experiences to prompt")
# Reset the manual-capture buffer for the new task. The real trajectory
# is captured by AWorld and read back from the TaskResponse below.
self.current_trajectory = []
# Execute the task
result = await Runners.run_task(task)
task_response = result.get(task.id)
# Recover the actual execution trajectory produced by AWorld's replay
# buffer (TaskResponse.trajectory). Falls back to manually captured
# actions when the framework did not record a trajectory.
trajectory = self._extract_trajectory(task_response)
# Process result for learning if enabled
if self.learning_mode and task_response and self._is_successful(task_response, task):
await self._learn_from_success(question, task_response, trajectory)
# Restore original prompt
self.system_prompt = original_prompt
return task_response
def _extract_trajectory(self, task_response: Optional[TaskResponse]) -> List[Dict[str, Any]]:
"""
Normalize the trajectory recorded by AWorld into the step format the
TrajectorySummarizer expects: ``{'action': {'tool_name', 'action_name',
'params'}}``.
AWorld stores each step as a serialized replay-buffer ``DataRow`` whose
``exp_data.actions`` field holds the ``ActionModel`` objects taken at
that step. When no framework trajectory is available (e.g. the runner
did not populate it), we fall back to any actions that were captured
manually via :meth:`capture_action`.
Args:
task_response: The response returned by ``Runners.run_task``.
Returns:
A list of normalized trajectory steps.
"""
raw = getattr(task_response, "trajectory", None) if task_response else None
if not raw:
return list(self.current_trajectory)
steps: List[Dict[str, Any]] = []
for row in raw:
actions = []
if isinstance(row, dict):
exp_data = row.get("exp_data") or {}
if isinstance(exp_data, dict):
actions = exp_data.get("actions") or []
for action in actions:
if not isinstance(action, dict):
continue
steps.append({
"action": {
"tool_name": action.get("tool_name") or "unknown",
"action_name": action.get("action_name") or "",
"params": action.get("params") or {},
}
})
# If normalization yielded nothing usable, keep the raw rows so the
# summarizer at least reflects the correct step count.
return steps or list(self.current_trajectory) or list(raw)
def _get_relevant_experiences(self, question: str) -> List[Dict[str, Any]]:
"""
Retrieve relevant experiences for a given question.
Args:
question: The current question
Returns:
List of relevant experiences
"""
relevant = []
# First check if knowledge base has preloaded experiences
if self.knowledge_base:
kb_experiences = self.knowledge_base.search(question, top_k=3)
relevant.extend(kb_experiences)
# Then check learned experiences
if self.experiences:
# Simple similarity check - can be enhanced with embeddings
for exp_id, exp_data in self.experiences.items():
if self._is_similar(question, exp_data.get('question', '')):
relevant.append(exp_data)
if len(relevant) >= 5: # Limit to top 5 experiences
break
return relevant
def _is_similar(self, q1: str, q2: str) -> bool:
"""
Simple similarity check between questions.
Can be enhanced with semantic similarity using embeddings.
Args:
q1: First question
q2: Second question
Returns:
True if questions are similar
"""
# Simple keyword overlap for now
if q1 is None or q2 is None:
return False
q1_words = set(q1.lower().split())
q2_words = set(q2.lower().split())
overlap = len(q1_words & q2_words)
total = len(q1_words | q2_words)
if total == 0:
return False
similarity = overlap / total
return similarity > 0.3 # Threshold for similarity
def _format_experiences(self, experiences: List[Dict[str, Any]]) -> str:
"""
Format experiences for inclusion in system prompt.
Args:
experiences: List of experience dictionaries
Returns:
Formatted experience text
"""
formatted = []
for i, exp in enumerate(experiences, 1):
exp_text = f"## Experience {i}:\n"
if 'question' in exp:
exp_text += f"- Similar Question: {exp['question']}\n"
if 'summary' in exp:
exp_text += f"- Key Insights: {exp['summary']}\n"
if 'approach' in exp:
exp_text += f"- Approach: {exp['approach']}\n"
if 'tools_used' in exp:
exp_text += f"- Tools Used: {', '.join(exp['tools_used'])}\n"
formatted.append(exp_text)
return "\n".join(formatted)
def _is_successful(self, response: TaskResponse, task: Task) -> bool:
"""
Determine if a task execution was successful.
Args:
response: The task response
task: The original task
Returns:
True if the task was successful
"""
# Check if answer exists and is not empty
if not response or not response.answer:
return False
# Honor AWorld's own success signal when the framework provides one.
success_flag = getattr(response, "success", None)
status = getattr(response, "status", None)
if success_flag is False:
return False
if status and str(status).lower() in {"failed", "cancelled"}:
return False
# Additional success criteria can be added here
# For GAIA, we might check against known answers if available
return True
async def _learn_from_success(self, question: str, response: TaskResponse, trajectory: List[Dict[str, Any]]):
"""
Learn from a successful task execution.
Args:
question: The original question
response: The successful response
trajectory: The execution trajectory
"""
if not self.summarizer:
logger.warning("No summarizer configured, skipping learning")
return
try:
# Summarize the trajectory
summary = await self.summarizer.summarize(question, response, trajectory)
# Store the experience
exp_id = self._get_task_hash(question)
self.experiences[exp_id] = {
'question': question,
'answer': response.answer,
'summary': summary.get('summary', ''),
'approach': summary.get('approach', ''),
'tools_used': summary.get('tools_used', []),
'key_insights': summary.get('key_insights', []),
'general_strategy': summary.get('general_strategy', ''),
'num_steps': len(trajectory),
'timestamp': datetime.now().isoformat(),
'success': True
}
# Save to disk
self._save_experiences()
logger.info(f"Learned from successful execution: {exp_id[:8]}")
except Exception as e:
logger.error(f"Failed to learn from success: {e}")
def capture_action(self, action: Dict[str, Any]):
"""
Capture an action in the current trajectory.
Args:
action: The action to capture
"""
if self.learning_mode:
self.current_trajectory.append({
'timestamp': datetime.now().isoformat(),
'action': action
})
knowledge_base.py¶
"""
Knowledge Base for Experience Retrieval
This module provides indexing and retrieval capabilities for experiences.
"""
import json
import logging
import os
from typing import Dict, Any, List, Optional, Tuple
import numpy as np
from sentence_transformers import SentenceTransformer
import faiss
import pickle
logger = logging.getLogger(__name__)
class KnowledgeBase:
"""
Knowledge base for storing and retrieving experiences using semantic search.
"""
def __init__(
self,
index_path: str = "./kb_index",
model_name: str = "all-MiniLM-L6-v2",
embedding_dim: int = 384
):
"""
Initialize the knowledge base.
Args:
index_path: Path to store the index files
model_name: Name of the sentence transformer model
embedding_dim: Dimension of the embeddings
"""
self.index_path = index_path
self.model_name = model_name
self.embedding_dim = embedding_dim
# Initialize the sentence transformer
try:
self.encoder = SentenceTransformer(model_name)
except Exception as e:
logger.warning(f"Failed to load SentenceTransformer, falling back to simple search: {e}")
self.encoder = None
# Initialize FAISS index
self.index = None
self.documents = []
self.metadata = []
# Create index directory if it doesn't exist
os.makedirs(index_path, exist_ok=True)
# Load existing index if available
self._load_index()
def _load_index(self):
"""Load existing index from disk."""
index_file = os.path.join(self.index_path, "faiss.index")
docs_file = os.path.join(self.index_path, "documents.pkl")
meta_file = os.path.join(self.index_path, "metadata.pkl")
if os.path.exists(index_file) and os.path.exists(docs_file):
try:
if self.encoder:
self.index = faiss.read_index(index_file)
with open(docs_file, 'rb') as f:
self.documents = pickle.load(f)
if os.path.exists(meta_file):
with open(meta_file, 'rb') as f:
self.metadata = pickle.load(f)
else:
self.metadata = [{}] * len(self.documents)
logger.info(f"Loaded knowledge base with {len(self.documents)} documents")
except Exception as e:
logger.error(f"Failed to load index: {e}")
self._create_new_index()
else:
self._create_new_index()
def _create_new_index(self):
"""Create a new empty index."""
if self.encoder:
self.index = faiss.IndexFlatL2(self.embedding_dim)
self.documents = []
self.metadata = []
def _save_index(self):
"""Save index to disk."""
try:
if self.encoder and self.index:
index_file = os.path.join(self.index_path, "faiss.index")
faiss.write_index(self.index, index_file)
docs_file = os.path.join(self.index_path, "documents.pkl")
with open(docs_file, 'wb') as f:
pickle.dump(self.documents, f)
meta_file = os.path.join(self.index_path, "metadata.pkl")
with open(meta_file, 'wb') as f:
pickle.dump(self.metadata, f)
except Exception as e:
logger.error(f"Failed to save index: {e}")
def index_gaia_validation(self, validation_file: str):
"""
Index the GAIA validation file for experience retrieval.
Args:
validation_file: Path to gaia-validation.jsonl
"""
if not os.path.exists(validation_file):
logger.error(f"Validation file not found: {validation_file}")
return
logger.info(f"Indexing GAIA validation data from {validation_file}")
try:
with open(validation_file, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
try:
data = json.loads(line)
# Extract relevant information
question = data.get('Question', '')
answer = data.get('Final answer', '')
level = data.get('Level', 0)
metadata = data.get('Annotator Metadata', {})
# Create experience document
experience = {
'task_id': data.get('task_id', f'gaia_{line_num}'),
'question': question,
'answer': answer,
'level': level,
'approach': self._extract_approach(metadata),
'tools_used': self._extract_tools(metadata),
'steps': metadata.get('Steps', ''),
'num_steps': metadata.get('Number of steps', '0'),
'source': 'gaia_validation'
}
# Add to index
self.add_experience(question, experience)
except json.JSONDecodeError as e:
logger.error(f"Failed to parse line {line_num}: {e}")
except Exception as e:
logger.error(f"Error processing line {line_num}: {e}")
# Save the index after bulk indexing
self._save_index()
logger.info(f"Successfully indexed {len(self.documents)} experiences from GAIA validation")
except Exception as e:
logger.error(f"Failed to index validation file: {e}")
def _extract_approach(self, metadata: Dict[str, Any]) -> str:
"""
Extract approach from metadata steps.
Args:
metadata: Annotator metadata
Returns:
Summarized approach
"""
steps = metadata.get('Steps', '')
if not steps:
return ""
# Extract key actions from steps
lines = steps.split('\n')
key_actions = []
for line in lines[:3]: # Take first 3 steps as approach
if line.strip():
# Remove step numbers
clean_line = line.strip()
if clean_line[0].isdigit():
clean_line = clean_line.split('.', 1)[-1].strip()
key_actions.append(clean_line)
return " → ".join(key_actions) if key_actions else steps[:200]
def _extract_tools(self, metadata: Dict[str, Any]) -> List[str]:
"""
Extract tools used from metadata.
Args:
metadata: Annotator metadata
Returns:
List of tools used
"""
tools = metadata.get('Tools', '')
if not tools:
return []
# Parse tools string
tool_list = []
# Handle numbered list format
lines = tools.split('\n')
for line in lines:
line = line.strip()
if line:
# Remove numbering
if '. ' in line:
tool = line.split('. ', 1)[-1].strip()
else:
tool = line
if tool and tool not in ['', 'None']:
tool_list.append(tool)
return tool_list
def add_experience(self, query: str, experience: Dict[str, Any]):
"""
Add an experience to the knowledge base.
Args:
query: The query/question for indexing
experience: The experience data
"""
# Store document
self.documents.append(experience)
self.metadata.append({
'query': query,
'task_id': experience.get('task_id', ''),
'level': experience.get('level', 0)
})
# Create embedding and add to index if encoder is available
if self.encoder and self.index:
try:
embedding = self.encoder.encode([query])
self.index.add(embedding)
except Exception as e:
logger.error(f"Failed to create embedding: {e}")
def search(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]:
"""
Search for relevant experiences.
Args:
query: The search query
top_k: Number of top results to return
Returns:
List of relevant experiences
"""
if not self.documents:
return []
# If we have embeddings, use semantic search
if self.encoder and self.index and self.index.ntotal > 0:
try:
query_embedding = self.encoder.encode([query])
distances, indices = self.index.search(query_embedding, min(top_k, len(self.documents)))
results = []
for idx in indices[0]:
if idx < len(self.documents):
results.append(self.documents[idx])
return results
except Exception as e:
logger.error(f"Semantic search failed, falling back to keyword search: {e}")
# Fallback to simple keyword search
return self._keyword_search(query, top_k)
def _keyword_search(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]:
"""
Simple keyword-based search fallback.
Args:
query: The search query
top_k: Number of results to return
Returns:
List of relevant experiences
"""
query_words = set(query.lower().split())
scored_docs = []
for doc in self.documents:
doc_text = f"{doc.get('question', '')} {doc.get('approach', '')} {' '.join(doc.get('tools_used', []))}"
doc_words = set(doc_text.lower().split())
# Calculate simple overlap score
overlap = len(query_words & doc_words)
if overlap > 0:
scored_docs.append((overlap, doc))
# Sort by score and return top k
scored_docs.sort(key=lambda x: x[0], reverse=True)
return [doc for _, doc in scored_docs[:top_k]]
def get_statistics(self) -> Dict[str, Any]:
"""
Get statistics about the knowledge base.
Returns:
Dictionary with statistics
"""
stats = {
'total_documents': len(self.documents),
'has_embeddings': self.encoder is not None,
'index_size': self.index.ntotal if self.encoder and self.index else 0,
'sources': {}
}
# Count by source
for doc in self.documents:
source = doc.get('source', 'unknown')
stats['sources'][source] = stats['sources'].get(source, 0) + 1
return stats
llm_env.py¶
"""包装层 LLM 环境解析：为 GAIA 经验学习示例提供带 OpenRouter 兜底的配置。
本文件位于 gaia-experience 包装层，**不修改** 上游 AWorld fork。它把散落在各处的
`os.getenv("LLM_API_KEY"/"LLM_BASE_URL"/"LLM_MODEL_NAME")` 读取集中起来，并加入统一兜底：
- 有 LLM_API_KEY / OPENAI_API_KEY -> 直连（沿用 LLM_BASE_URL，可为空即 OpenAI 官方）
- 否则有 OPENROUTER_API_KEY -> 走 OpenRouter（https://openrouter.ai/api/v1），
并把模型名映射到 OpenRouter 命名：
gpt-* -> openai/gpt-*
claude-* -> anthropic/claude-opus-4.8
含 "/" -> 原样透传
其它 -> openai/gpt-5.6-luna
默认模型 gpt-5.6-luna。
"""
import os
DEFAULT_MODEL = "gpt-5.6-luna"
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
def to_openrouter_model(model: str) -> str:
if not model:
return "openai/gpt-5.6-luna"
if "/" in model:
return model
if model.startswith("gpt-"):
return "openai/" + model
if model.startswith("claude-"):
return "anthropic/claude-opus-4.8"
return "openai/gpt-5.6-luna"
def resolve_llm(default_model: str = DEFAULT_MODEL, model_override: str = None) -> dict:
"""返回 AgentConfig 需要的 provider/model/base_url/api_key 四元组（含 OpenRouter 兜底）。
model_override: 若显式给了模型名（如 config.yaml 或 CLI），优先用它。
"""
provider = os.getenv("LLM_PROVIDER", "openai")
model = model_override or os.getenv("LLM_MODEL_NAME", default_model)
base_url = os.getenv("LLM_BASE_URL")
api_key = os.getenv("LLM_API_KEY") or os.getenv("OPENAI_API_KEY")
if not api_key and os.getenv("OPENROUTER_API_KEY"):
api_key = os.getenv("OPENROUTER_API_KEY")
base_url = base_url or OPENROUTER_BASE_URL
model = to_openrouter_model(model)
return {
"llm_provider": provider,
"llm_model_name": model,
"llm_base_url": base_url,
"llm_api_key": api_key,
}
run_with_experience.py¶
"""
Modified GAIA Runner with Learning from Experience
This script extends the original GAIA runner with experience learning capabilities.
"""
import argparse
import json
import logging
import os
import re
import traceback
from pathlib import Path
from typing import Any, Dict, List
from dotenv import load_dotenv
from llm_env import resolve_llm, DEFAULT_MODEL
# The heavy runtime dependencies (AWorld framework + local experience
# components, which pull in sentence-transformers / faiss) are imported lazily
# via _load_runtime() so that `--help` and argument parsing work even when the
# full stack is not installed. The names below are populated on first run.
AgentConfig = TaskConfig = Task = None
system_prompt = None
add_file_path = load_dataset_meta = question_scorer = report_results = None
ExperienceAgent = KnowledgeBase = TrajectorySummarizer = None
def _load_runtime():
"""Import AWorld and experience components lazily (see note above)."""
global AgentConfig, TaskConfig, Task, system_prompt
global add_file_path, load_dataset_meta, question_scorer, report_results
global ExperienceAgent, KnowledgeBase, TrajectorySummarizer
from AWorld.aworld.config.conf import AgentConfig as _AgentConfig
from AWorld.aworld.config.conf import TaskConfig as _TaskConfig
from AWorld.aworld.core.task import Task as _Task
from AWorld.examples.gaia.prompt import system_prompt as _system_prompt
from AWorld.examples.gaia.utils import (
add_file_path as _add_file_path,
load_dataset_meta as _load_dataset_meta,
question_scorer as _question_scorer,
report_results as _report_results,
)
from experience_agent import ExperienceAgent as _ExperienceAgent
from knowledge_base import KnowledgeBase as _KnowledgeBase
from trajectory_summarizer import TrajectorySummarizer as _TrajectorySummarizer
AgentConfig, TaskConfig, Task = _AgentConfig, _TaskConfig, _Task
system_prompt = _system_prompt
add_file_path, load_dataset_meta = _add_file_path, _load_dataset_meta
question_scorer, report_results = _question_scorer, _report_results
ExperienceAgent = _ExperienceAgent
KnowledgeBase = _KnowledgeBase
TrajectorySummarizer = _TrajectorySummarizer
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def parse_arguments():
"""解析命令行参数（Parse command line arguments）。"""
parser = argparse.ArgumentParser(
prog="run_with_experience.py",
description=(
"在 GAIA 基准上运行带“从经验中学习”能力的 Agent（实验 8-1：策略摘要）。\n"
"支持两种模式：学习模式（learning-mode）在任务成功后把轨迹提炼成经验并入库；\n"
"应用模式（apply-experience）在解题前检索最相似的历史经验注入系统提示词。\n"
"使用 --compare 可自动做 A/B 对照，直观展示“复用经验是否提升 GAIA 成绩”。"
),
epilog=(
"示例（Examples）：\n"
" # 1) 仅解析参数、查看帮助（无需 API/数据集）\n"
" python run_with_experience.py --help\n\n"
" # 2) 学习模式：从前 10 道题的成功轨迹中沉淀经验\n"
" python run_with_experience.py --learning-mode --start 0 --end 10\n\n"
" # 3) 应用模式：用已学到的经验解 10~20 题\n"
" python run_with_experience.py --apply-experience --start 10 --end 20\n\n"
" # 4) A/B 对照：同一批题分别在“无经验/有经验”下各跑一次并对比准确率\n"
" # （先用 learning-mode 在其它题上积累经验，再在未见过的题上对照，避免数据泄漏）\n"
" python run_with_experience.py --compare --start 10 --end 20 \\\n"
" --experience-db ./learned_experiences.json\n\n"
" # 5) 指定主 Agent 模型与结果输出路径\n"
" python run_with_experience.py --apply-experience --model gpt-5.6-luna \\\n"
" --output ./results/exp_run.json --start 0 --end 5\n"
),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
# ---- 任务集选择（Task set selection）----
parser.add_argument(
"--start",
type=int,
default=0,
help="数据集起始下标（默认：0）",
)
parser.add_argument(
"--end",
type=int,
default=20,
help="数据集结束下标（不含，默认：20）",
)
parser.add_argument(
"--q",
type=str,
help="指定单个题目下标或 task_id；优先级最高，会覆盖 --start/--end",
)
parser.add_argument(
"--skip",
action="store_true",
help="若题目此前已处理过则跳过",
)
parser.add_argument(
"--split",
type=str,
default="validation",
help="数据集划分，如 validation、test（默认：validation）",
)
parser.add_argument(
"--blacklist_file_path",
type=str,
nargs="?",
help="黑名单文件路径，如 blacklist.txt（其中的 task_id 会被跳过）",
)
# ---- 模型与输出（Model & output）----
parser.add_argument(
"--model",
type=str,
default=None,
help="主 Agent 使用的模型名；缺省时读取环境变量 LLM_MODEL_NAME（默认 gpt-5.6-luna）",
)
parser.add_argument(
"--summary-model",
type=str,
default=DEFAULT_MODEL,
help="用于轨迹总结（经验提炼）的模型（默认：gpt-5.6-luna）",
)
parser.add_argument(
"--embedding-model",
type=str,
default="all-MiniLM-L6-v2",
help="用于语义检索的 sentence-transformers 模型（默认：all-MiniLM-L6-v2）",
)
parser.add_argument(
"--output",
type=str,
default=None,
help="结果 JSON 输出路径；缺省写入 $AWORLD_WORKSPACE/experience_results.json",
)
# ---- 经验学习开关（Experience learning）----
parser.add_argument(
"--learning-mode",
action="store_true",
help="启用学习模式：捕获成功轨迹并总结为可复用经验",
)
parser.add_argument(
"--apply-experience",
action="store_true",
help="启用应用模式：为新任务检索并注入相关历史经验",
)
parser.add_argument(
"--compare",
action="store_true",
help="A/B 对照模式：同一批题分别在“无经验/有经验”下各跑一次并对比准确率",
)
parser.add_argument(
"--preload-kb",
action="store_true",
help="从 gaia-validation.jsonl 预加载知识库（注意：勿在评测同一批题上预加载，以免泄漏答案）",
)
parser.add_argument(
"--kb-path",
type=str,
default="./kb_index",
help="知识库索引存储路径（默认：./kb_index）",
)
parser.add_argument(
"--experience-db",
type=str,
default="./learned_experiences.json",
help="已学习经验的存储路径（默认：./learned_experiences.json）",
)
parser.add_argument(
"--validation-file",
type=str,
default="gaia-validation.jsonl",
help="用于预加载的 gaia-validation.jsonl 路径",
)
return parser.parse_args()
def setup_logging(args):
"""Setup logging configuration."""
workspace = os.getenv("AWORLD_WORKSPACE", ".")
os.makedirs(workspace, exist_ok=True)
log_file_name = f"experience_agent_{args.q}.log" if args.q else f"experience_agent_{args.start}_{args.end}.log"
file_handler = logging.FileHandler(
os.path.join(workspace, log_file_name),
mode="a",
encoding="utf-8",
)
file_handler.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
file_handler.setFormatter(formatter)
logging.getLogger().addHandler(file_handler)
def load_mcp_config(config_path: Path) -> Dict[str, Any]:
"""Load MCP configuration."""
mcp_config = {}
available_servers = []
try:
if config_path.exists():
with open(config_path, mode="r", encoding="utf-8") as f:
mcp_config = json.loads(f.read())
available_servers = list(mcp_config.get("mcpServers", {}).keys())
logger.info(f"🔧 MCP Available Servers: {available_servers}")
except json.JSONDecodeError as e:
logger.error(f"Error loading mcp_collections.json: {e}")
return mcp_config, available_servers
async def run_with_experience(args):
"""Main execution function with experience learning."""
_load_runtime()
# Load dataset
gaia_dataset_path = os.getenv("GAIA_DATASET_PATH", "./gaia_dataset")
full_dataset = load_dataset_meta(gaia_dataset_path, split=args.split)
logger.info(f"Total questions: {len(full_dataset)}")
# Load MCP configuration
mcp_config_path = Path(__file__).parent / "AWorld" / "examples" / "gaia" / "mcp.json"
mcp_config, available_servers = load_mcp_config(mcp_config_path)
# Setup agent configuration. The main agent model can be overridden via the
# --model CLI flag; otherwise it falls back to the LLM_MODEL_NAME env var.
# OpenAI 直连优先，缺 Key 时自动走 OpenRouter 兜底（见 llm_env.resolve_llm）。
agent_config = AgentConfig(
**resolve_llm(model_override=args.model),
llm_temperature=float(os.getenv("LLM_TEMPERATURE", "0.0"))
)
# Initialize knowledge base
knowledge_base = None
if args.apply_experience or args.preload_kb:
logger.info("Initializing knowledge base...")
knowledge_base = KnowledgeBase(
index_path=args.kb_path,
model_name=args.embedding_model
)
# Preload validation data if requested
if args.preload_kb and os.path.exists(args.validation_file):
logger.info(f"Preloading knowledge base from {args.validation_file}")
knowledge_base.index_gaia_validation(args.validation_file)
stats = knowledge_base.get_statistics()
logger.info(f"Knowledge base statistics: {stats}")
# Initialize trajectory summarizer
summarizer = None
if args.learning_mode:
logger.info("Initializing trajectory summarizer...")
summarizer = TrajectorySummarizer(
llm_config=agent_config,
model_name=resolve_llm(model_override=args.summary_model)["llm_model_name"]
)
# Create experience agent
experience_agent = ExperienceAgent(
conf=agent_config,
name="gaia_experience_agent",
system_prompt=system_prompt,
learning_mode=args.learning_mode,
apply_experience=args.apply_experience,
experience_db_path=args.experience_db,
knowledge_base=knowledge_base,
summarizer=summarizer,
mcp_config=mcp_config,
mcp_servers=available_servers,
)
logger.info(f"Experience Agent initialized:")
logger.info(f" - Learning mode: {args.learning_mode}")
logger.info(f" - Apply experience: {args.apply_experience}")
logger.info(f" - Knowledge base: {'Yes' if knowledge_base else 'No'}")
logger.info(f" - Summarizer: {'Yes' if summarizer else 'No'}")
# Load existing results (path overridable via --output)
results_file = args.output or os.path.join(
os.getenv("AWORLD_WORKSPACE", "."), "experience_results.json"
)
results_dir = os.path.dirname(results_file)
if results_dir:
os.makedirs(results_dir, exist_ok=True)
if os.path.exists(results_file):
with open(results_file, "r", encoding="utf-8") as f:
results = json.load(f)
else:
results = []
# Load blacklist
blacklist = set()
if args.blacklist_file_path and os.path.exists(args.blacklist_file_path):
with open(args.blacklist_file_path, "r", encoding="utf-8") as f:
blacklist = set(f.read().splitlines())
try:
# Determine dataset slice
if args.q:
dataset_slice = [
record for record in full_dataset
if record["task_id"] == args.q
]
else:
dataset_slice = full_dataset[args.start:args.end]
# Process each question
for i, dataset_i in enumerate(dataset_slice):
# Check if should skip
if not args.q:
if dataset_i["task_id"] in blacklist:
logger.info(f"Skipping blacklisted task: {dataset_i['task_id']}")
continue
if args.skip and any(
result["task_id"] == dataset_i["task_id"]
for result in results
):
logger.info(f"Skipping already processed task: {dataset_i['task_id']}")
continue
try:
# Log task details
logger.info(f"{'='*60}")
logger.info(f"Processing task {i+1}/{len(dataset_slice)}: {dataset_i['task_id']}")
logger.info(f"Question: {dataset_i['Question']}")
logger.info(f"Level: {dataset_i['Level']}")
logger.info(f"Tools: {dataset_i['Annotator Metadata']['Tools']}")
# Prepare question with file paths
question_data = add_file_path(dataset_i, file_path=gaia_dataset_path, split=args.split)
question = question_data["Question"]
# Create and execute task
task = Task(
input=question,
agent=experience_agent,
conf=TaskConfig()
)
# Execute with experience learning/application
task_response = await experience_agent.execute_task(task)
# Extract answer
answer = None
if task_response and task_response.answer:
match = re.search(r"<answer>(.*?)</answer>", task_response.answer)
if match:
answer = match.group(1)
# Evaluate result
is_correct = False
if answer:
logger.info(f"Agent answer: {answer}")
logger.info(f"Correct answer: {dataset_i['Final answer']}")
is_correct = question_scorer(answer, dataset_i["Final answer"])
if is_correct:
logger.info(f"✓ Question {i} Correct!")
else:
logger.info(f"✗ Incorrect!")
else:
logger.warning("No answer extracted from response")
# Record result
new_result = {
"task_id": dataset_i["task_id"],
"level": dataset_i["Level"],
"question": question,
"answer": dataset_i["Final answer"],
"response": answer or "",
"is_correct": is_correct,
"learning_mode": args.learning_mode,
"applied_experience": args.apply_experience
}
# Update or append result
existing_index = next(
(idx for idx, result in enumerate(results)
if result["task_id"] == dataset_i["task_id"]),
None
)
if existing_index is not None:
results[existing_index] = new_result
else:
results.append(new_result)
# Save intermediate results
with open(results_file, "w", encoding="utf-8") as f:
json.dump(results, f, indent=4, ensure_ascii=False)
except Exception as e:
logger.error(f"Error processing task {dataset_i['task_id']}: {e}")
logger.error(traceback.format_exc())
continue
except KeyboardInterrupt:
logger.info("Interrupted by user")
finally:
# Report final results
report_results(results)
# Save final results
with open(results_file, "w", encoding="utf-8") as f:
json.dump(results, f, indent=4, ensure_ascii=False)
# Log experience learning statistics if enabled
if args.learning_mode:
num_experiences = len(experience_agent.experiences)
logger.info(f"\nLearning Statistics:")
logger.info(f" - Total experiences learned: {num_experiences}")
logger.info(f" - Experience database: {args.experience_db}")
if args.apply_experience and knowledge_base:
stats = knowledge_base.get_statistics()
logger.info(f"\nKnowledge Base Statistics:")
logger.info(f" - Total indexed documents: {stats['total_documents']}")
logger.info(f" - Sources: {stats['sources']}")
async def _evaluate_slice(experience_agent, dataset_slice, gaia_dataset_path, args, pass_label):
"""
Run one evaluation pass over a dataset slice and return per-task results.
The agent's ``apply_experience`` flag is read as currently set on the
``experience_agent`` instance, which lets the caller toggle experience reuse
on/off between passes for an apples-to-apples A/B comparison. Correctness is
computed with GAIA's own ``question_scorer`` — no numbers are fabricated.
Args:
experience_agent: The (already constructed) ExperienceAgent.
dataset_slice: The list of dataset records to evaluate.
gaia_dataset_path: Root path of the GAIA dataset (for attached files).
args: Parsed CLI arguments.
pass_label: Human-readable label for logging (e.g. "baseline").
Returns:
A list of per-task result dicts including an ``is_correct`` flag.
"""
results = []
for i, dataset_i in enumerate(dataset_slice):
try:
logger.info(f"[{pass_label}] Task {i + 1}/{len(dataset_slice)}: {dataset_i['task_id']}")
question_data = add_file_path(dataset_i, file_path=gaia_dataset_path, split=args.split)
question = question_data["Question"]
task = Task(input=question, agent=experience_agent, conf=TaskConfig())
task_response = await experience_agent.execute_task(task)
answer = None
if task_response and task_response.answer:
match = re.search(r"<answer>(.*?)</answer>", task_response.answer)
if match:
answer = match.group(1)
is_correct = bool(answer) and question_scorer(answer, dataset_i["Final answer"])
results.append({
"task_id": dataset_i["task_id"],
"level": dataset_i["Level"],
"question": question,
"answer": dataset_i["Final answer"],
"response": answer or "",
"is_correct": is_correct,
})
except Exception as e:
logger.error(f"[{pass_label}] Error on {dataset_i['task_id']}: {e}")
results.append({
"task_id": dataset_i["task_id"],
"level": dataset_i["Level"],
"is_correct": False,
"error": str(e),
})
return results
def _accuracy(results):
"""Compute accuracy (correct / total) from a list of result dicts."""
total = len(results)
correct = sum(1 for r in results if r.get("is_correct"))
return correct, total, (correct / total if total else 0.0)
async def run_comparison(args):
"""
A/B comparison: evaluate the same task slice twice — once WITHOUT experience
reuse (baseline) and once WITH it — then report the accuracy delta.
This directly demonstrates the experiment's thesis: reusing accumulated
experience improves GAIA performance. All reported numbers are computed from
the actual runs; none are hard-coded.
"""
_load_runtime()
gaia_dataset_path = os.getenv("GAIA_DATASET_PATH", "./gaia_dataset")
full_dataset = load_dataset_meta(gaia_dataset_path, split=args.split)
mcp_config_path = Path(__file__).parent / "AWorld" / "examples" / "gaia" / "mcp.json"
mcp_config, available_servers = load_mcp_config(mcp_config_path)
agent_config = AgentConfig(
**resolve_llm(model_override=args.model),
llm_temperature=float(os.getenv("LLM_TEMPERATURE", "0.0")),
)
# Build the experience knowledge base for the "with experience" pass.
knowledge_base = KnowledgeBase(index_path=args.kb_path, model_name=args.embedding_model)
if args.preload_kb and os.path.exists(args.validation_file):
logger.warning(
"--preload-kb indexes gaia-validation.jsonl; if the evaluated tasks are in "
"that file this leaks their reference solutions. For a fair comparison, "
"accumulate experiences from OTHER tasks (via --learning-mode) instead."
)
knowledge_base.index_gaia_validation(args.validation_file)
experience_agent = ExperienceAgent(
conf=agent_config,
name="gaia_experience_agent",
system_prompt=system_prompt,
learning_mode=False, # keep both passes clean; learn separately
apply_experience=False,
experience_db_path=args.experience_db,
knowledge_base=knowledge_base,
summarizer=None,
mcp_config=mcp_config,
mcp_servers=available_servers,
)
num_learned = len(experience_agent.experiences)
kb_docs = knowledge_base.get_statistics()["total_documents"]
logger.info(
f"Comparison ready: {num_learned} learned experiences, {kb_docs} KB documents available."
)
if num_learned == 0 and kb_docs == 0:
logger.warning(
"No experiences available to reuse. Run --learning-mode first (or pass "
"--preload-kb) so the 'with experience' pass has something to retrieve."
)
dataset_slice = full_dataset[args.start:args.end]
# Pass 1: baseline (no experience reuse)
experience_agent.apply_experience = False
baseline = await _evaluate_slice(experience_agent, dataset_slice, gaia_dataset_path, args, "baseline")
# Pass 2: with experience reuse
experience_agent.apply_experience = True
with_exp = await _evaluate_slice(experience_agent, dataset_slice, gaia_dataset_path, args, "with-experience")
b_correct, b_total, b_acc = _accuracy(baseline)
e_correct, e_total, e_acc = _accuracy(with_exp)
report = {
"split": args.split,
"range": [args.start, args.end],
"num_learned_experiences": num_learned,
"kb_documents": kb_docs,
"baseline": {"correct": b_correct, "total": b_total, "accuracy": b_acc, "results": baseline},
"with_experience": {"correct": e_correct, "total": e_total, "accuracy": e_acc, "results": with_exp},
"delta_accuracy": e_acc - b_acc,
}
out = args.output or os.path.join(os.getenv("AWORLD_WORKSPACE", "."), "comparison_results.json")
out_dir = os.path.dirname(out)
if out_dir:
os.makedirs(out_dir, exist_ok=True)
with open(out, "w", encoding="utf-8") as f:
json.dump(report, f, indent=2, ensure_ascii=False)
# Console summary (computed from real runs)
print("\n" + "=" * 60)
print("A/B COMPARISON: experience reuse vs. baseline")
print("=" * 60)
print(f" Tasks evaluated : {b_total} (split={args.split}, range=[{args.start}, {args.end}))")
print(f" Reusable experiences : {num_learned} learned, {kb_docs} preloaded")
print(f" Baseline accuracy : {b_correct}/{b_total} = {b_acc:.1%}")
print(f" With-experience acc. : {e_correct}/{e_total} = {e_acc:.1%}")
print(f" Delta (with - base) : {e_acc - b_acc:+.1%}")
print(f" Full report written to: {out}")
print("=" * 60)
def main():
"""Main entry point."""
# Parse arguments
args = parse_arguments()
# Load environment
load_dotenv()
# Setup logging
setup_logging(args)
# Log configuration
logger.info("Starting GAIA with Experience Learning")
logger.info(f"Configuration:")
logger.info(f" - Learning mode: {args.learning_mode}")
logger.info(f" - Apply experience: {args.apply_experience}")
logger.info(f" - Preload KB: {args.preload_kb}")
logger.info(f" - Compare mode: {args.compare}")
# Run async main
import asyncio
if args.compare:
asyncio.run(run_comparison(args))
else:
asyncio.run(run_with_experience(args))
if __name__ == "__main__":
main()
test_analyze_action_types_null.py¶
"""Unit-test _analyze_action_types / _format_trajectory with null tool_name,
without importing the full AWorld agent stack (same ast-extraction trick as
test_experience_similar.py)."""
import ast
from pathlib import Path
from typing import Any, Dict, List
def _load_methods(*method_names):
src = Path(__file__).with_name("trajectory_summarizer.py").read_text()
tree = ast.parse(src)
methods = []
for node in tree.body:
if isinstance(node, ast.ClassDef) and node.name == "TrajectorySummarizer":
for item in node.body:
if isinstance(item, ast.FunctionDef) and item.name in method_names:
methods.append(item)
break
assert len(methods) == len(method_names)
class_src = ast.Module(
body=[
ast.ClassDef(
name="TrajectorySummarizer",
bases=[],
keywords=[],
body=methods,
decorator_list=[],
)
],
type_ignores=[],
)
ast.fix_missing_locations(class_src)
ns = {"List": List, "Dict": Dict, "Any": Any}
exec(compile(class_src, "trajectory_summarizer.py", "exec"), ns)
return ns["TrajectorySummarizer"]()
def test_analyze_action_types_null_tool_name():
ts = _load_methods("_analyze_action_types")
# 轨迹步骤显式携带 "tool_name": null（框架序列化 ActionModel 无工具时的形态）
traj = [{"action": {"tool_name": None, "action_name": "x", "params": {}}}]
assert ts._analyze_action_types(traj) == {"other": 1}
def test_analyze_action_types_normal():
ts = _load_methods("_analyze_action_types")
traj = [
{"action": {"tool_name": "mshtools-web_search", "action_name": "", "params": {}}},
{"action": {"tool_name": "mshtools-web_search", "action_name": "", "params": {}}},
{"action": {"tool_name": "mshtools-compute", "action_name": "", "params": {}}},
]
result = ts._analyze_action_types(traj)
assert result["search"] == 2
assert result["computation"] == 1
def test_format_trajectory_null_tool_name():
ts = _load_methods("_format_trajectory", "_truncate_value")
traj = [{"action": {"tool_name": None, "action_name": None, "params": None}}]
text = ts._format_trajectory(traj)
assert "unknown" in text
assert "None" not in text
test_experience_similar.py¶
"""Unit-test _is_similar without importing the full AWorld agent stack."""
import ast
from pathlib import Path
def _load_is_similar():
src = Path(__file__).with_name("experience_agent.py").read_text()
tree = ast.parse(src)
method = None
for node in tree.body:
if isinstance(node, ast.ClassDef) and node.name == "ExperienceAgent":
for item in node.body:
if isinstance(item, ast.FunctionDef) and item.name == "_is_similar":
method = item
break
assert method is not None
# Build a tiny class with only this method
class_src = ast.Module(
body=[
ast.ClassDef(
name="ExperienceAgent",
bases=[],
keywords=[],
body=[method],
decorator_list=[],
)
],
type_ignores=[],
)
ast.fix_missing_locations(class_src)
ns = {}
exec(compile(class_src, "experience_agent.py", "exec"), ns)
return ns["ExperienceAgent"]()
def test_is_similar_none_question():
agent = _load_is_similar()
assert agent._is_similar(None, "what is 2+2") is False
assert agent._is_similar("what is 2+2", None) is False
assert agent._is_similar("what is 2+2", "what is 2+2") is True
trajectory_summarizer.py¶
"""
Trajectory Summarizer for Learning from Experience
This module summarizes successful task trajectories into reusable experiences.
"""
import json
import logging
import re
from typing import Dict, Any, List, Optional
from AWorld.aworld.models.llm import get_llm_model
from AWorld.aworld.config.conf import AgentConfig
logger = logging.getLogger(__name__)
class TrajectorySummarizer:
"""
Summarizes task execution trajectories into natural language experiences.
"""
def __init__(
self,
llm_config: Optional[AgentConfig] = None,
model_name: str = "gpt-5.6-luna",
temperature: float = 0.3
):
"""
Initialize the trajectory summarizer.
Args:
llm_config: LLM configuration
model_name: Model to use for summarization
temperature: Temperature for generation
"""
self.llm_config = llm_config
self.model_name = model_name
self.temperature = temperature
# Initialize LLM for summarization
if llm_config:
self.llm = get_llm_model(
provider=llm_config.llm_provider,
model_name=model_name,
api_key=llm_config.llm_api_key,
base_url=llm_config.llm_base_url
)
else:
self.llm = None
async def summarize(
self,
question: str,
response: Any,
trajectory: List[Dict[str, Any]]
) -> Dict[str, Any]:
"""
Summarize a successful trajectory into reusable experience.
Args:
question: The original question
response: The successful response
trajectory: The execution trajectory
Returns:
Summarized experience dictionary
"""
if not self.llm:
# Fallback to rule-based summarization
return self._rule_based_summary(question, response, trajectory)
try:
# Prepare trajectory for LLM
trajectory_text = self._format_trajectory(trajectory)
# Create summarization prompt
prompt = self._create_summary_prompt(question, response.answer, trajectory_text)
# Get summary from LLM
summary_response = await self.llm.acompletion(
messages=[
{"role": "system", "content": "You are an expert at analyzing task execution trajectories and extracting key insights for future problem-solving."},
{"role": "user", "content": prompt}
],
temperature=self.temperature
)
# Parse the response
summary = self._parse_llm_summary(summary_response)
# Extract tools used from trajectory
tools_used = self._extract_tools_from_trajectory(trajectory)
summary['tools_used'] = tools_used
return summary
except Exception as e:
logger.error(f"LLM summarization failed: {e}")
# Fallback to rule-based
return self._rule_based_summary(question, response, trajectory)
def _create_summary_prompt(self, question: str, answer: str, trajectory_text: str) -> str:
"""
Create a prompt for LLM summarization.
Args:
question: The original question
answer: The final answer
trajectory_text: Formatted trajectory
Returns:
Prompt string
"""
prompt = f"""Analyze this successful task execution and extract key insights for solving similar problems in the future.
Question: {question}
Final Answer: {answer}
Execution Trajectory:
{trajectory_text}
Please provide a concise summary with the following structure:
1. APPROACH: Describe the high-level approach taken to solve this problem (2-3 sentences)
2. KEY INSIGHTS: What are the critical insights or patterns that made this solution successful? (2-3 bullet points)
3. GENERAL STRATEGY: How could this approach be generalized to similar problems? (1-2 sentences)
Format your response as JSON with keys: "summary", "approach", "key_insights" (list), "general_strategy"
"""
return prompt
def _parse_llm_summary(self, response: Any) -> Dict[str, Any]:
"""
Parse LLM response into structured summary.
Args:
response: LLM response object
Returns:
Parsed summary dictionary
"""
try:
# Extract JSON from response
content = response.content if hasattr(response, 'content') else str(response)
# Try to find JSON in the response
json_match = re.search(r'\{.*\}', content, re.DOTALL)
if json_match:
summary_json = json.loads(json_match.group())
return {
'summary': summary_json.get('summary', ''),
'approach': summary_json.get('approach', ''),
'key_insights': summary_json.get('key_insights', []),
'general_strategy': summary_json.get('general_strategy', '')
}
except Exception as e:
logger.error(f"Failed to parse LLM summary: {e}")
# Fallback: treat entire response as summary
return {
'summary': content,
'approach': 'See summary for details',
'key_insights': [],
'general_strategy': ''
}
def _rule_based_summary(
self,
question: str,
response: Any,
trajectory: List[Dict[str, Any]]
) -> Dict[str, Any]:
"""
Create a rule-based summary when LLM is not available.
Args:
question: The original question
response: The successful response
trajectory: The execution trajectory
Returns:
Summary dictionary
"""
# Extract tools used
tools_used = self._extract_tools_from_trajectory(trajectory)
# Analyze trajectory patterns
action_types = self._analyze_action_types(trajectory)
# Create basic summary
summary = {
'summary': f"Successfully answered question using {len(tools_used)} tools with {len(trajectory)} steps.",
'approach': self._infer_approach(action_types, tools_used),
'key_insights': self._extract_key_patterns(trajectory),
'general_strategy': f"Use combination of {', '.join(tools_used[:3])} for similar tasks",
'tools_used': tools_used
}
return summary
def _format_trajectory(self, trajectory: List[Dict[str, Any]]) -> str:
"""
Format trajectory for readability.
Args:
trajectory: Raw trajectory data
Returns:
Formatted trajectory string
"""
formatted_steps = []
for i, step in enumerate(trajectory, 1):
action = step.get('action', {})
# Extract key information
tool_name = action.get('tool_name') or 'unknown'
action_name = action.get('action_name', '')
params = action.get('params', {})
# Format step
step_text = f"Step {i}: {tool_name}"
if action_name:
step_text += f".{action_name}"
if params:
# Simplify params for readability
param_str = ', '.join([f"{k}={self._truncate_value(v)}" for k, v in params.items()])
step_text += f"({param_str})"
formatted_steps.append(step_text)
return '\n'.join(formatted_steps)
def _truncate_value(self, value: Any, max_length: int = 50) -> str:
"""
Truncate long values for display.
Args:
value: Value to truncate
max_length: Maximum length
Returns:
Truncated string representation
"""
str_value = str(value)
if len(str_value) > max_length:
return str_value[:max_length] + "..."
return str_value
def _extract_tools_from_trajectory(self, trajectory: List[Dict[str, Any]]) -> List[str]:
"""
Extract unique tools used from trajectory.
Args:
trajectory: Execution trajectory
Returns:
List of unique tool names
"""
tools = set()
for step in trajectory:
action = step.get('action', {})
tool_name = action.get('tool_name')
if tool_name:
# Clean tool name
base_tool = tool_name.split('__')[0] if '__' in tool_name else tool_name
tools.add(base_tool)
return sorted(list(tools))
def _analyze_action_types(self, trajectory: List[Dict[str, Any]]) -> Dict[str, int]:
"""
Analyze types of actions in trajectory.
Args:
trajectory: Execution trajectory
Returns:
Count of each action type
"""
action_types = {}
for step in trajectory:
action = step.get('action', {})
tool_name = action.get('tool_name') or 'unknown'
# Categorize action
if 'search' in tool_name.lower():
category = 'search'
elif 'browser' in tool_name.lower() or 'web' in tool_name.lower():
category = 'web_interaction'
elif 'file' in tool_name.lower() or 'read' in tool_name.lower():
category = 'file_operation'
elif 'calculate' in tool_name.lower() or 'compute' in tool_name.lower():
category = 'computation'
else:
category = 'other'
action_types[category] = action_types.get(category, 0) + 1
return action_types
def _infer_approach(self, action_types: Dict[str, int], tools_used: List[str]) -> str:
"""
Infer the approach based on action types.
Args:
action_types: Count of action types
tools_used: List of tools used
Returns:
Inferred approach description
"""
# Determine dominant strategy
if action_types.get('search', 0) > 2:
approach = "Information gathering through multiple searches"
elif action_types.get('web_interaction', 0) > 3:
approach = "Web-based research and navigation"
elif action_types.get('file_operation', 0) > 1:
approach = "File analysis and processing"
elif action_types.get('computation', 0) > 0:
approach = "Computational problem solving"
else:
approach = "Multi-step problem decomposition"
# Add tool specifics
if tools_used:
approach += f" using {', '.join(tools_used[:2])}"
return approach
def _extract_key_patterns(self, trajectory: List[Dict[str, Any]]) -> List[str]:
"""
Extract key patterns from trajectory.
Args:
trajectory: Execution trajectory
Returns:
List of key patterns/insights
"""
patterns = []
# Check for search refinement pattern
search_count = sum(1 for step in trajectory
if 'search' in str(step.get('action', {})).lower())
if search_count > 1:
patterns.append("Multiple searches refined the query")
# Check for verification pattern
if len(trajectory) > 5:
patterns.append("Thorough verification of results")
# Check for tool combination
tools = self._extract_tools_from_trajectory(trajectory)
if len(tools) > 2:
patterns.append(f"Combined {len(tools)} different tools effectively")
# Limit to 3 patterns
return patterns[:3] if patterns else ["Direct problem solving approach"]
run.sh¶
#!/bin/bash
# GAIA Experience Learning System Runner
# This script provides convenient commands for running the system
set -e # Exit on error
# Colors for output
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color
# Default values
START_IDX=0
END_IDX=10
SPLIT="validation"
# Function to print colored messages
print_info() {
echo -e "${BLUE}[INFO]${NC} $1"
}
print_success() {
echo -e "${GREEN}[SUCCESS]${NC} $1"
}
print_warning() {
echo -e "${YELLOW}[WARNING]${NC} $1"
}
print_error() {
echo -e "${RED}[ERROR]${NC} $1"
}
# Function to check prerequisites
check_prerequisites() {
print_info "Checking prerequisites..."
# Check Python
if ! command -v python &> /dev/null; then
print_error "Python is not installed"
exit 1
fi
# Check .env file
if [ ! -f .env ]; then
print_warning ".env file not found. Creating from template..."
cat > .env << EOF
# LLM Configuration
LLM_PROVIDER=openai
LLM_MODEL_NAME=gpt-5.6-luna
LLM_API_KEY=your_api_key_here
# LLM_BASE_URL=https://api.openai.com/v1
# Dataset paths
GAIA_DATASET_PATH=./AWorld/examples/gaia/GAIA
AWORLD_WORKSPACE=./workspace
EOF
print_warning "Please edit .env file with your API keys"
exit 1
fi
print_success "Prerequisites checked"
}
# Function to setup environment
setup_environment() {
print_info "Setting up environment..."
# Create necessary directories
mkdir -p kb_index
mkdir -p experiences
mkdir -p logs
mkdir -p workspace
# Check if AWorld is installed
if [ ! -d "AWorld" ]; then
print_error "AWorld directory not found. Please ensure AWorld is cloned in this directory."
exit 1
fi
print_success "Environment setup complete"
}
# Function to show help
show_help() {
cat << EOF
GAIA Experience Learning System Runner
=======================================
Usage: ./run.sh [COMMAND] [OPTIONS]
Commands:
demo Run the interactive demo
learn Run in learning mode (capture experiences)
apply Run with experience application
full Run with both learning and application
compare A/B compare baseline vs. experience reuse on the same tasks
index Index the validation dataset
test Run a specific test case
help Show this help message
Options:
--start N Start index (default: 0)
--end N End index (default: 10)
--split TYPE Dataset split: validation/test (default: validation)
--task-id ID Specific task ID to run
--model NAME Main agent model (overrides LLM_MODEL_NAME)
--output PATH Results JSON output path
--no-preload Don't preload knowledge base
Examples:
./run.sh demo # Run interactive demo
./run.sh learn --start 0 --end 5 # Learn from first 5 questions
./run.sh apply --start 5 --end 10 # Apply experiences to questions 5-10
./run.sh full # Run complete workflow
./run.sh compare --start 10 --end 20 # A/B: baseline vs. experience reuse
./run.sh test --task-id "task-123" # Run specific task
Note: For a fair 'compare', first accumulate experiences on OTHER tasks, e.g.
./run.sh learn --start 0 --end 10
./run.sh compare --start 10 --end 20
EOF
}
# Parse command
COMMAND=${1:-help}
shift || true
# Parse options
while [[ $# -gt 0 ]]; do
case $1 in
--start)
START_IDX="$2"
shift 2
;;
--end)
END_IDX="$2"
shift 2
;;
--split)
SPLIT="$2"
shift 2
;;
--task-id)
TASK_ID="$2"
shift 2
;;
--model)
MODEL="$2"
shift 2
;;
--output)
OUTPUT="$2"
shift 2
;;
--no-preload)
NO_PRELOAD=true
shift
;;
*)
print_error "Unknown option: $1"
show_help
exit 1
;;
esac
done
# Compose optional pass-through args (model / output)
EXTRA_ARGS=""
if [ -n "$MODEL" ]; then
EXTRA_ARGS="$EXTRA_ARGS --model $MODEL"
fi
if [ -n "$OUTPUT" ]; then
EXTRA_ARGS="$EXTRA_ARGS --output $OUTPUT"
fi
# Execute command
case $COMMAND in
demo)
print_info "Running interactive demo..."
check_prerequisites
setup_environment
python demo.py --interactive
;;
learn)
print_info "Running in learning mode..."
print_info "Processing questions $START_IDX to $END_IDX from $SPLIT split"
check_prerequisites
setup_environment
python run_with_experience.py \
--learning-mode \
--start "$START_IDX" \
--end "$END_IDX" \
--split "$SPLIT" \
$EXTRA_ARGS
;;
apply)
print_info "Running with experience application..."
print_info "Processing questions $START_IDX to $END_IDX from $SPLIT split"
check_prerequisites
setup_environment
PRELOAD_FLAG=""
if [ "$NO_PRELOAD" != true ]; then
PRELOAD_FLAG="--preload-kb"
fi
python run_with_experience.py \
--apply-experience \
$PRELOAD_FLAG \
--start "$START_IDX" \
--end "$END_IDX" \
--split "$SPLIT" \
$EXTRA_ARGS
;;
full)
print_info "Running with full experience learning..."
print_info "Processing questions $START_IDX to $END_IDX from $SPLIT split"
check_prerequisites
setup_environment
PRELOAD_FLAG=""
if [ "$NO_PRELOAD" != true ]; then
PRELOAD_FLAG="--preload-kb"
fi
python run_with_experience.py \
--learning-mode \
--apply-experience \
$PRELOAD_FLAG \
--start "$START_IDX" \
--end "$END_IDX" \
--split "$SPLIT" \
$EXTRA_ARGS
;;
compare)
print_info "Running A/B comparison (baseline vs. experience reuse)..."
print_info "Evaluating questions $START_IDX to $END_IDX from $SPLIT split twice"
check_prerequisites
setup_environment
PRELOAD_FLAG=""
if [ "$NO_PRELOAD" == true ]; then
PRELOAD_FLAG=""
fi
python run_with_experience.py \
--compare \
$PRELOAD_FLAG \
--start "$START_IDX" \
--end "$END_IDX" \
--split "$SPLIT" \
$EXTRA_ARGS
;;
index)
print_info "Indexing validation dataset..."
check_prerequisites
setup_environment
python -c "
import asyncio
from knowledge_base import KnowledgeBase
async def index():
kb = KnowledgeBase()
kb.index_gaia_validation('gaia-validation.jsonl')
stats = kb.get_statistics()
print(f'Indexed {stats[\"total_documents\"]} documents')
asyncio.run(index())
"
print_success "Indexing complete"
;;
test)
if [ -z "$TASK_ID" ]; then
print_error "Task ID required. Use --task-id option"
exit 1
fi
print_info "Running test for task: $TASK_ID"
check_prerequisites
setup_environment
python run_with_experience.py \
--learning-mode \
--apply-experience \
--preload-kb \
--q "$TASK_ID"
;;
help|--help|-h)
show_help
;;
*)
print_error "Unknown command: $COMMAND"
show_help
exit 1
;;
esac
print_success "Done!"
config.yaml¶
# Experience Learning Configuration for GAIA
# Learning Mode Settings
learning:
enabled: true # Enable learning from successful trajectories
summarizer:
model: "gpt-5.6-luna" # Model for summarizing trajectories
temperature: 0.3 # Lower temperature for consistent summaries
max_trajectory_steps: 100 # Maximum steps to include in summary
# Experience storage
storage:
path: "./learned_experiences.json"
max_experiences: 1000 # Maximum experiences to store
auto_save: true # Auto-save after each new experience
# Knowledge Base Settings
knowledge_base:
enabled: true # Enable knowledge base for retrieval
preload_validation: true # Preload gaia-validation.jsonl on startup
# Index settings
index:
path: "./kb_index"
embedding_model: "all-MiniLM-L6-v2" # Sentence transformer model
embedding_dim: 384 # Dimension of embeddings
# Search settings
search:
top_k: 3 # Number of similar experiences to retrieve
min_similarity: 0.3 # Minimum similarity threshold
use_semantic: true # Use semantic search (vs keyword only)
# Experience Application Settings
application:
enabled: true # Apply experiences to new tasks
# How to incorporate experiences
strategy: "prompt_enhancement" # Options: prompt_enhancement, context_injection
# Prompt enhancement settings
prompt:
max_experiences: 5 # Maximum experiences to include in prompt
format: "detailed" # Options: detailed, summary, minimal
# Experience filtering
filtering:
by_level: true # Filter experiences by difficulty level
by_tools: true # Filter by similar tools required
recency_bias: 0.1 # Weight recent experiences higher (0-1)
# GAIA Dataset Settings
dataset:
validation_file: "gaia-validation.jsonl"
dataset_path: "./AWorld/examples/gaia/GAIA"
# Processing settings
batch_size: 10 # Process questions in batches
timeout: 300 # Timeout per question (seconds)
# Evaluation settings
evaluation:
strict_matching: false # Use strict answer matching
case_sensitive: false # Case-sensitive comparison
# Agent Settings
agent:
name: "gaia_experience_agent"
# LLM settings (can be overridden by environment variables)
llm:
provider: "openai" # Default provider
model: "gpt-5.6-luna" # Default model
temperature: 0.0 # Temperature for task execution
max_tokens: 4096 # Maximum tokens per response
# Tool settings
tools:
enable_all: true # Enable all available tools
timeout: 60 # Tool execution timeout (seconds)
# Memory settings
memory:
short_term_capacity: 20 # Short-term memory capacity
enable_long_term: true # Enable long-term memory
# Logging Settings
logging:
level: "INFO" # Logging level: DEBUG, INFO, WARNING, ERROR
file: "./logs/experience_agent.log"
# Detailed logging options
log_trajectories: true # Log full trajectories
log_summaries: true # Log generated summaries
log_retrieval: true # Log retrieval results
# Experimental Features
experimental:
# Multi-agent collaboration
multi_agent:
enabled: false # Enable multi-agent mode
share_experiences: false # Share experiences between agents
# Active learning
active_learning:
enabled: false # Request human feedback on uncertain cases
uncertainty_threshold: 0.3 # Threshold for requesting feedback
# Experience evolution
evolution:
enabled: false # Allow experiences to evolve over time
merge_similar: false # Merge similar experiences
prune_outdated: false # Remove outdated experiences
# Performance Optimization
optimization:
# Caching
cache:
enabled: true # Enable caching
embeddings: true # Cache embeddings
llm_responses: false # Cache LLM responses
# Parallel processing
parallel:
enabled: false # Enable parallel processing
max_workers: 4 # Maximum parallel workers
# Resource limits
limits:
max_memory_gb: 8 # Maximum memory usage
max_cpu_percent: 80 # Maximum CPU usage