Upload 5 files

aworld/replay_buffer/README.md

@@ -0,0 +1,111 @@
+# Replay Buffer
+
+A multi-process capable replay buffer system for storing and sampling experience data.
+
+## Features
+
+- **Multi-process Support**: Safe concurrent access using shared memory and locks
+- **Flexible Querying**: Powerful query builder for filtering stored data
+- **Task-based Organization**: Data organized by task_id and agent_id
+- **Capacity Management**: FIFO eviction when reaching max capacity
+- **Custom Sampling**: Implement custom sampling logic through Sampler interface
+- **Data Conversion**: Custom data conversion through Converter interface
+
+## Basic Usage
+
+### Writing Data
+
+```python
+from aworld.replay_buffer import ReplayBuffer, DataRow, ExpMeta, Experience
+from aworld.core.common import ActionModel, Observation
+
+# Create a data row
+data = DataRow(
+    exp_meta=ExpMeta(
+        task_id="task_1",
+        task_name="my_task",
+        agent_id="agent_1",
+        step=1,
+        execute_time=time.time()
+    ),
+    exp_data=Experience(
+        state=Observation(),
+        action=ActionModel()
+    )
+)
+
+# Store data
+replay_buffer.store(data)
+```
+
+### Reading Data
+
+```python
+from aworld.replay_buffer.query_filter import QueryBuilder
+
+# Basic example
+replay_buffer = ReplayBuffer()
+query_condition = QueryBuilder().eq("exp_meta.task_name", "test_task").build()
+data = replay_buffer.sample(sampler=RandomTaskSample(),
+                            query_condition=query_condition,
+                            converter=DefaultConverter(),
+                            batch_size=1000)
+
+# Query Task by task_id
+query = QueryBuilder().eq("exp_meta.task_id", "task_1").build()
+data = replay_buffer.sample_task(query_condition=query, batch_size=10)
+
+# Query Task by agent_id 
+query = QueryBuilder().eq("exp_meta.agent_id", "agent_1").build()
+data = replay_buffer.sample_task(query_condition=query, batch_size=5)
+```
+## Multi-processing Example
+
+```python
+import multiprocessing
+from aworld.replay_buffer.storage.multi_proc_mem import MultiProcMemoryStorage
+
+manager = multiprocessing.Manager()
+replay_buffer = ReplayBuffer(
+    storage=MultiProcMemoryStorage(
+        data_dict=manager.dict(),
+        fifo_queue=manager.list(),
+        lock=manager.Lock(),
+        max_capacity=10000
+    )
+)
+
+# Start writer processes
+processes = [
+    multiprocessing.Process(target=write_processing, args=(replay_buffer, f"task_{i}"))
+    for i in range(4)
+]
+```
+## Query Builder Examples
+
+### Simple Equality
+```python
+QueryBuilder().eq("exp_meta.task_id", "123").build()
+```
+
+### Complex Conditions
+```python
+QueryBuilder()
+    .eq("exp_meta.task_id", "123")
+    .and_()
+    .eq("exp_meta.agent_id", "456")
+    .build()
+```
+### Nested Conditions
+```python
+QueryBuilder()
+    .eq("exp_meta.task_id", "123")
+    .and_()
+    .nested(
+        QueryBuilder()
+            .eq("exp_meta.agent_id", "111")
+            .or_()
+            .eq("exp_meta.agent_id", "222")
+    )
+    .build()
+```

aworld/replay_buffer/__init__.py

@@ -0,0 +1,2 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.

aworld/replay_buffer/base.py

@@ -0,0 +1,409 @@
+import random
+import uuid
+from dataclasses import dataclass, field
+from typing import Dict, List, TypeVar
+from abc import ABC, abstractmethod
+from math import ceil
+
+from aworld.core.common import ActionModel, Observation
+from aworld.replay_buffer.query_filter import QueryCondition, QueryFilter
+from aworld.logs.util import logger
+
+
+T = TypeVar('T')
+
+
+@dataclass
+class Experience:
+    '''
+    Experience of agent.
+    '''
+    state: Observation
+    actions: List[ActionModel]
+    reward_t: float = None
+    adv_t: float = None
+    v_t: float = None
+    messages: List[Dict] = None
+
+    def to_dict(self):
+        return {
+            "state": self.state,
+            "actions": self.actions,
+            "reward_t": self.reward_t,
+            "adv_t": self.adv_t,
+            "v_t": self.v_t,
+            "messages": self.messages
+        }
+
+
+@dataclass
+class ExpMeta:
+    '''
+    Experience meta data.
+    '''
+    task_id: str
+    task_name: str
+    agent_id: str
+    step: int
+    execute_time: float
+    pre_agent: str
+
+    def to_dict(self):
+        return {
+            "task_id": self.task_id,
+            "task_name": self.task_name,
+            "agent_id": self.agent_id,
+            "step": self.step,
+            "execute_time": self.execute_time,
+            "pre_agent": self.pre_agent
+        }
+@dataclass
+class DataRow:
+    '''
+    Data row for storing data.
+    '''
+    exp_meta: ExpMeta
+    exp_data: Experience
+    id: str = field(default_factory=lambda: str(uuid.uuid4()))
+
+    def to_dict(self):
+        return {
+            "exp_meta": self.exp_meta.to_dict(),
+            "exp_data": self.exp_data.to_dict(),
+            "id": self.id
+        }
+
+
+class Storage(ABC):
+    '''
+    Storage for storing and sampling data.
+    '''
+
+    @abstractmethod
+    def add(self, data: DataRow):
+        '''
+        Add data to the storage.
+        Args:
+            data (DataRow): Data to add.
+        '''
+
+    @abstractmethod
+    def add_batch(self, data_batch: List[DataRow]):
+        '''
+        Add batch of data to the storage.
+        Args:
+            data_batch (List[DataRow]): List of data to add.
+        '''
+
+    @abstractmethod
+    def size(self, query_condition: QueryCondition = None) -> int:
+        '''
+        Get the size of the storage.
+        Returns:
+            int: Size of the storage.
+        '''
+
+    @abstractmethod
+    def get_paginated(self, page: int, page_size: int, query_condition: QueryCondition = None) -> List[DataRow]:
+        '''
+        Get paginated data from the storage.
+        Args:
+            page (int): Page number.
+            page_size (int): Number of data per page.
+        Returns:
+            List[DataRow]: List of data.
+        '''
+
+    @abstractmethod
+    def get_all(self, query_condition: QueryCondition = None) -> List[DataRow]:
+        '''
+        Get all data from the storage.
+        Returns:
+            List[DataRow]: List of data.
+        '''
+
+    @abstractmethod
+    def get_by_task_id(self, task_id: str) -> List[DataRow]:
+        '''
+        Get data by task_id from the storage.
+        Args:
+            task_id (str): Task id.
+        Returns:
+            List[DataRow]: List of data.
+        '''
+
+    @abstractmethod
+    def get_bacth_by_task_ids(self, task_ids: List[str]) -> Dict[str, List[DataRow]]:
+        '''
+        Get batch of data by task_ids from the storage.
+        Args:
+            task_ids (List[str]): List of task ids.
+        Returns:
+            Dict[str, List[DataRow]]: Dictionary of data.
+            The key is the task_id and the value is the list of data.
+            The list of data is sorted by step.
+        '''
+
+
+class Sampler(ABC):
+    '''
+    Sample data from the storage.
+    '''
+
+    def sample(self,
+               storage: Storage,
+               batch_size: int,
+               query_condition: QueryCondition = None) -> List[DataRow]:
+        '''
+        Sample data from the storage.
+        Args:
+            storage (Storage): Storage to sample from.
+            batch_size (int): Number of data to sample.
+            query_condition (QueryCondition, optional): Query condition. Defaults to None.
+        Returns:
+            List[DataRow]
+        '''
+
+
+class TaskSampler(Sampler):
+    '''
+    Sample task data from storage, returns Dict[str, List[DataRow]] where:
+    - key is task_id
+    - value is list of task all data rows
+    '''
+
+    def sorted_by_step(self, task_experience: List[DataRow]) -> List[DataRow]:
+        '''
+        Sort the task experience by step and execute_time.
+        Args:
+            task_experience (List[DataRow]): List of task experience.
+        Returns:
+            List[DataRow]: List of task experience sorted by step and execute_time.
+        '''
+        return sorted(task_experience, key=lambda x: (x.exp_meta.step, x.exp_meta.execute_time))
+
+    def sample(self,
+               storage: Storage,
+               batch_size: int,
+               query_condition: QueryCondition = None) -> List[DataRow]:
+        task_ids = self.sample_task_ids(storage, batch_size, query_condition)
+        return storage.get_bacth_by_task_ids(task_ids)
+
+    def sample_tasks(self,
+                     storage: Storage,
+                     batch_size: int,
+                     query_condition: QueryCondition = None) -> Dict[str, List[DataRow]]:
+        '''
+        Sample data from the storage.
+        Args:
+            storage (Storage): Storage to sample from.
+            batch_size (int): Number of data to sample.
+            query_condition (QueryCondition, optional): Query condition. Defaults to None.
+        Returns:
+            Dict[str, List[DataRow]]: Dictionary of sampled data.
+            The key is the task_id and the value is the list of data.
+            The list of data is sorted by step.
+        '''
+        task_ids = self.sample_task_ids(storage, batch_size, query_condition)
+        raws = storage.get_bacth_by_task_ids(task_ids)
+        return {task_id: self.sorted_by_step(raws) for task_id, raws in raws.items()}
+
+    @abstractmethod
+    def sample_task_ids(self,
+                        storage: Storage,
+                        batch_size: int,
+                        query_condition: QueryCondition = None) -> List[str]:
+        '''
+        Sample task_ids from the storage.
+        Args:
+            storage (Storage): Storage to sample from.
+            batch_size (int): Number of task_ids to sample.
+            query_condition (QueryCondition, optional): Query condition. Defaults to None.
+        Returns:
+            List[str]: List of task_ids.
+        '''
+
+
+class Converter(ABC):
+    '''
+    Convert data to dataset row.
+    '''
+
+    @abstractmethod
+    def to_dataset_row(self, task_experience: List[DataRow]) -> T:
+        '''
+        Convert task experience to dataset row.
+        Args:
+            task_experience (List[DataRow]): List of task experience.
+        Returns:
+            T: type of dataset row.
+        '''
+
+
+class InMemoryStorage(Storage):
+    '''
+    In-memory storage for storing and sampling data.
+    '''
+
+    def __init__(self, max_capacity: int = 10000):
+        self._data: Dict[str, List[DataRow]] = {}
+        self._max_capacity = max_capacity
+        self._fifo_queue = []  # (task_id)
+
+    def add(self, data: DataRow):
+        if not data:
+            raise ValueError("Data is required")
+        if not data.exp_meta:
+            raise ValueError("exp_meta is required")
+
+        while self.size() >= self._max_capacity and self._fifo_queue:
+            oldest_task_id = self._fifo_queue.pop(0)
+            if oldest_task_id in self._data:
+                del self._data[oldest_task_id]
+
+        if data.exp_meta.task_id not in self._data:
+            self._data[data.exp_meta.task_id] = []
+        self._data[data.exp_meta.task_id].append(data)
+        self._fifo_queue.append(data.exp_meta.task_id)
+
+        if data.exp_meta.task_id not in self._data:
+            self._data[data.exp_meta.task_id] = []
+        self._data[data.exp_meta.task_id].append(data)
+
+    def add_batch(self, data_batch: List[DataRow]):
+        for data in data_batch:
+            self.add(data)
+
+    def size(self, query_condition: QueryCondition = None) -> int:
+        return len(self.get_all(query_condition))
+
+    def get_paginated(self, page: int, page_size: int, query_condition: QueryCondition = None) -> List[DataRow]:
+        if page < 1:
+            raise ValueError("Page must be greater than 0")
+        if page_size < 1:
+            raise ValueError("Page size must be greater than 0")
+        all_data = self.get_all(query_condition)
+        start_index = (page - 1) * page_size
+        end_index = start_index + page_size
+        return all_data[start_index:end_index]
+
+    def get_all(self, query_condition: QueryCondition = None) -> List[DataRow]:
+        all_data = []
+        query_filter = None
+        if query_condition:
+            query_filter = QueryFilter(query_condition)
+        for data in self._data.values():
+            if query_filter:
+                all_data.extend(query_filter.filter(data))
+            else:
+                all_data.extend(data)
+        return all_data
+
+    def get_by_task_id(self, task_id: str) -> List[DataRow]:
+        return self._data.get(task_id, [])
+
+    def get_bacth_by_task_ids(self, task_ids: List[str]) -> Dict[str, List[DataRow]]:
+        return {task_id: self._data.get(task_id, []) for task_id in task_ids}
+
+    def clear(self):
+        self._data = {}
+        self._fifo_queue = []
+
+
+class RandomTaskSample(TaskSampler):
+    '''
+    Randomly sample data from the storage.
+    '''
+
+    def sample_task_ids(self,
+                        storage: Storage,
+                        batch_size: int,
+                        query_condition: QueryCondition = None) -> List[str]:
+        total_size = storage.size(query_condition)
+        if total_size <= batch_size:
+            return storage.get_all(query_condition)
+
+        sampled_task_ids = set()
+        page_size = min(100, batch_size * 2)
+        total_pages = ceil(total_size/page_size)
+        visited_pages = set()
+        while len(sampled_task_ids) < batch_size and len(visited_pages) < total_pages:
+            page = random.choice(
+                [p for p in range(1, total_pages+1) if p not in visited_pages])
+            visited_pages.add(page)
+
+            current_page = storage.get_paginated(
+                page, page_size, query_condition)
+            if not current_page:
+                continue
+            current_page_task_ids = set(
+                [data.exp_meta.task_id for data in current_page if data.exp_meta.task_id not in sampled_task_ids])
+            sample_count = min(len(current_page_task_ids),
+                               batch_size - len(sampled_task_ids))
+            sampled_task_ids.update(random.sample(
+                list(current_page_task_ids), sample_count))
+
+        return list(sampled_task_ids)
+
+
+class DefaultConverter(Converter):
+    '''
+    Default converter do nothing.
+    '''
+
+    def to_dataset_row(self, task_experience: List[DataRow]) -> List[DataRow]:
+        return task_experience
+
+
+class ReplayBuffer:
+    '''
+    Replay buffer for storing and sampling data.
+    '''
+
+    def __init__(
+        self,
+        storage: Storage = InMemoryStorage()
+    ):
+        self._storage = storage
+
+    def store(self, data: DataRow):
+        '''
+        Store data in the replay buffer.
+        '''
+        if not data:
+            raise ValueError("Data is required")
+        self._storage.add(data)
+
+    def store_batch(self, data_batch: List[DataRow]):
+        '''
+        Store batch of data in the replay buffer.
+        '''
+        if not data_batch:
+            raise ValueError("Data batch is required")
+        self._storage.add_batch(data_batch)
+
+    def sample_task(self,
+                    sampler: TaskSampler = RandomTaskSample(),
+                    query_condition: QueryCondition = None,
+                    converter: Converter = DefaultConverter(),
+                    batch_size: int = 1000) -> List[T]:
+        '''
+        Sample Task from the replay buffer and convert to dataset row.
+        DefaultConverter return List[DataRow]
+        '''
+        sampled_task = sampler.sample_tasks(
+            self._storage, batch_size, query_condition)
+        return [converter.to_dataset_row(task_experiences) for task_experiences in sampled_task.values()]
+
+    def sample(self,
+               sampler: Sampler = RandomTaskSample(),
+               query_condition: QueryCondition = None,
+               converter: Converter = DefaultConverter(),
+               batch_size: int = 1000) -> List[T]:
+        '''
+        Sample data from the replay buffer and convert to dataset row.
+        DefaultConverter return List[DataRow]
+        '''
+        sampled_data = sampler.sample(
+            self._storage, batch_size, query_condition)
+        return converter.to_dataset_row(sampled_data)

aworld/replay_buffer/processor.py

@@ -0,0 +1,190 @@
+# coding: utf-8
+"""
+processor.py
+Used to clean raw trace data into standard storage structure for reinforcement learning training.
+"""
+import json
+import os
+import datetime
+from typing import Any
+import threading
+
+from aworld.utils import import_package
+from aworld.replay_buffer.base import DataRow, Experience, ExpMeta
+from aworld.logs.util import logger
+from aworld.utils.common import get_local_ip
+
+
+class ReplayBufferExporter:
+    def __init__(self):
+        """Initialize ReplayBufferExporter instance"""
+        self._file_locks = {}
+        self._lock_dict_lock = threading.Lock()
+        self._task_output_paths = {}
+
+    def _get_file_lock(self, file_path):
+        """Get the lock for the specified file"""
+        with self._lock_dict_lock:
+            if file_path not in self._file_locks:
+                self._file_locks[file_path] = threading.Lock()
+            return self._file_locks[file_path]
+
+    def replay_buffer_exporter(self, spans: list[dict[str, Any]], output_dir: str):
+        """
+        Process spans, only process spans with 'step_execution_' prefix, and group by task_id to output to different files
+
+        Args:
+            spans: span data list
+            output_dir: output directory path
+        """
+        # Ensure output directory exists
+        import_package("oss2")
+        import oss2
+
+        os.makedirs(output_dir, exist_ok=True)
+
+        # Get OSS credentials from environment variables
+        enable_oss_export = os.getenv("EXPORT_REPLAY_TRACE_TO_OSS", "false").lower() == "true"
+        access_key_id = os.getenv('OSS_ACCESS_KEY_ID')
+        access_key_secret = os.getenv('OSS_ACCESS_KEY_SECRET')
+        endpoint = os.getenv('OSS_ENDPOINT')
+        bucket_name = os.getenv('OSS_BUCKET_NAME')
+        bucket = None
+
+        if not all([access_key_id, access_key_secret, endpoint, bucket_name]):
+            enable_oss_export = False
+            logger.warn("Missing required OSS environment variables")
+        else:
+            try:
+                # Initialize OSS client
+                auth = oss2.Auth(access_key_id, access_key_secret)
+                bucket = oss2.Bucket(auth, endpoint, bucket_name)
+            except Exception as e:
+                enable_oss_export = False
+                logger.warn(f"Failed to initialize OSS client, endpoint: {endpoint}, bucket: {bucket_name}. Error: {str(e)}")
+
+        # Group by task_id
+        task_groups = {}
+
+        for span_data in spans:
+            # Only process spans with 'step_execution_' prefix
+            if not span_data['name'].startswith('step_execution_'):
+                continue
+
+            attr = span_data.get('attributes', {})
+            exp_id = attr.get('exp_id')
+            task_id = attr.get('task_id', '')
+
+            if not exp_id or not task_id:
+                continue
+
+            if task_id not in task_groups:
+                task_groups[task_id] = {}
+
+            if exp_id not in task_groups[task_id]:
+                task_groups[task_id][exp_id] = {
+                    'exp_meta': None,
+                    'exp_data': None
+                }
+
+            # Process step_execution span
+            task_name = attr.get('task_name', '')
+            agent_id = attr.get('agent_id', '')
+            step = attr.get('step', 0)
+            execute_time = float(span_data.get('start_time', 0).split('.')[0].replace(' ', '').replace('-', '').replace(':', ''))
+
+            observation = {}
+            action = []
+            messages = []
+            pre_agent = None
+            if 'observation' in attr:
+                try:
+                    observation = json.loads(attr['observation'])
+                except:
+                    observation = attr['observation']
+
+            if 'actions' in attr:
+                try:
+                    action = json.loads(attr['actions'])
+                except:
+                    action = attr['actions']
+
+            if 'messages' in attr:
+                try:
+                    messages = json.loads(attr['messages'])
+                except:
+                    messages = attr['messages']
+
+            pre_agent = attr.get('pre_agent', '')
+            reward = attr.get('reward', 0.0)
+            adv = attr.get('adv_t', 0.0)
+            v = attr.get('v_t', 0.0)
+
+            exp_meta = ExpMeta(task_id, task_name, agent_id, step, execute_time, pre_agent)
+            exp_data = Experience(observation, action, reward, adv, v, messages)
+
+            task_groups[task_id][exp_id]['exp_meta'] = exp_meta
+            task_groups[task_id][exp_id]['exp_data'] = exp_data
+
+        # Process data for each task_id
+        for task_id, exp_groups in task_groups.items():
+            # Merge data and generate final Experience object
+            data_rows = []
+
+            # Read existing data (if any)
+            output_path = self._task_output_paths.get(task_id)
+            if not output_path:
+                timestamp = datetime.datetime.now().strftime("%Y%m%d")
+                replay_dir = os.path.join(output_dir or "./trace_data", timestamp, get_local_ip(), "replays")
+                replay_dataset_path = os.getenv("REPLAY_TRACE_DATASET_PATH", replay_dir)
+                export_dir = os.path.abspath(replay_dataset_path)
+                os.makedirs(export_dir, exist_ok=True)
+                output_path = os.path.join(export_dir, f"task_replay_{task_id}.json")
+                self._task_output_paths[task_id] = output_path
+
+            # Use thread lock to protect read and write operations
+            file_lock = self._get_file_lock(output_path)
+            with file_lock:
+                if os.path.exists(output_path):
+                    try:
+                        with open(output_path, 'r', encoding='utf-8') as f:
+                            existing_data = json.load(f)
+                            data_rows.extend([DataRow(
+                                ExpMeta(**row['exp_meta']),
+                                Experience(**row['exp_data']),
+                                row['id']
+                            ) for row in existing_data])
+                    except Exception as e:
+                        print(f"Failed to read existing file {output_path}: {str(e)}")
+
+                # Add new data
+                for exp_id, group in exp_groups.items():
+                    if group['exp_meta'] and group['exp_data']:
+                        row = DataRow(group['exp_meta'], group['exp_data'], exp_id)
+                        data_rows.append(row)
+
+                # Sort by execute_time
+                data_rows.sort(key=lambda x: x.exp_meta.execute_time)
+
+                # Export to json
+                with open(output_path, 'w', encoding='utf-8') as f:
+                    json.dump([row.to_dict() for row in data_rows], f, ensure_ascii=False, indent=2)
+                logger.info(f"Processing completed, exported {len(data_rows)} experiences to {output_path}")
+
+                if enable_oss_export:
+                    # Upload to OSS
+                    try:
+                        # Get the relative path
+                        abs_path = os.path.abspath(output_path)
+                        path_parts = abs_path.split(os.sep)
+                        if len(path_parts) >= 4:
+                            # Get the last 4 parts of the path
+                            relative_path = os.sep.join(path_parts[-4:])
+                            oss_key = relative_path
+                        else:
+                            oss_key = f"replay_buffer/{os.path.basename(output_path)}"
+                        bucket.put_object_from_file(oss_key, output_path)
+                        logger.info(f"Successfully uploaded {output_path} to OSS: {oss_key}")
+                    except Exception as e:
+                        logger.warn(f"Failed to upload {output_path} to OSS: {str(e)}")
+

aworld/replay_buffer/query_filter.py

@@ -0,0 +1,228 @@
+from typing import Any, List, TypeVar, Union, Literal, TypedDict, Dict
+
+DataRow = TypeVar('DataRow')
+
+
+class BaseCondition(TypedDict):
+    field: str
+    value: Any
+    op: Literal[
+        'eq', 'ne', 'gt', 'gte', 'lt', 'lte',
+        'in', 'not_in', 'like', 'not_like',
+        'is_null', 'is_not_null'
+    ]
+
+
+class LogicalCondition(TypedDict):
+    and_: List['QueryCondition']
+    or_: List['QueryCondition']
+
+
+QueryCondition = Union[BaseCondition, LogicalCondition]
+
+
+class QueryBuilder:
+    '''
+    Query builder for replay buffer. result example:
+    {
+        "and": [
+            {"field": "field1", "value": "value1", "op": "eq"}, 
+            {"or": [{"field": "field2", "value": "value2", "op": "eq"}, {"field": "field3", "value": "value3", "op": "eq"}]}
+        ]
+    }
+    '''
+
+    def __init__(self) -> None:
+        self.conditions: List[Dict[str, any]] = []
+        self.logical_ops: List[str] = []
+
+    def eq(self, field: str, value: any) -> 'QueryBuilder':
+        self.conditions.append({"field": field, "value": value, "op": "eq"})
+        return self
+
+    def ne(self, field: str, value: any) -> 'QueryBuilder':
+        self.conditions.append({"field": field, "value": value, "op": "ne"})
+        return self
+
+    def gt(self, field: str, value: any) -> 'QueryBuilder':
+        self.conditions.append({"field": field, "value": value, "op": "gt"})
+        return self
+
+    def gte(self, field: str, value: any) -> 'QueryBuilder':
+        self.conditions.append({"field": field, "value": value, "op": "gte"})
+        return self
+
+    def lt(self, field: str, value: any) -> 'QueryBuilder':
+        self.conditions.append({"field": field, "value": value, "op": "lt"})
+        return self
+
+    def lte(self, field: str, value: any) -> 'QueryBuilder':
+        self.conditions.append({"field": field, "value": value, "op": "lte"})
+        return self
+
+    def in_(self, field: str, value: any) -> 'QueryBuilder':
+        self.conditions.append({"field": field, "value": value, "op": "in"})
+        return self
+
+    def not_in(self, field: str, value: any) -> 'QueryBuilder':
+        self.conditions.append(
+            {"field": field, "value": value, "op": "not_in"})
+        return self
+
+    def like(self, field: str, value: any) -> 'QueryBuilder':
+        self.conditions.append({"field": field, "value": value, "op": "like"})
+        return self
+
+    def not_like(self, field: str, value: any) -> 'QueryBuilder':
+        self.conditions.append(
+            {"field": field, "value": value, "op": "not_like"})
+        return self
+
+    def is_null(self, field: str) -> 'QueryBuilder':
+        self.conditions.append({"field": field, "op": "is_null"})
+        return self
+
+    def is_not_null(self, field: str) -> 'QueryBuilder':
+        self.conditions.append({"field": field, "op": "is_not_null"})
+        return self
+
+    def and_(self) -> 'QueryBuilder':
+        self.logical_ops.append("and_")
+        return self
+
+    def or_(self) -> 'QueryBuilder':
+        self.logical_ops.append("or_")
+        return self
+
+    def nested(self, builder: 'QueryBuilder') -> 'QueryBuilder':
+        self.conditions.append({"nested": builder.build()})
+        return self
+
+    def build(self) -> QueryCondition:
+        conditions = self.conditions  # all conditions（including nested）
+        operators = self.logical_ops
+
+        # Validate condition and operator counts (n conditions need n-1 operators)
+        if len(operators) != len(conditions) - 1:
+            raise ValueError("Mismatch between condition and operator counts")
+
+        # Use stack to handle operator precedence (simplified version supporting and/or)
+        stack: List[Union[Dict[str, any], str]] = []
+
+        for i, item in enumerate(conditions):
+            if i == 0:
+                # First element goes directly to stack (condition or nested)
+                stack.append(item)
+                continue
+
+            # Pop stack top as left operand
+            left = stack.pop()
+            op = operators[i-1]       # Current operator (and/or)
+            right = item              # Right operand (current condition)
+
+            # Build logical expression: {op: [left, right]}
+            expr = {op: [left, right]}
+            # Push result back to stack for further operations
+            stack.append(expr)
+
+        # Process nested conditions (recursive unfolding)
+        def process_nested(cond: any) -> any:
+            if isinstance(cond, dict):
+                if "nested" in cond:
+                    # Recursively process sub-conditions
+                    return process_nested(cond["nested"])
+                # Recursively process child elements
+                return {k: process_nested(v) for k, v in cond.items()}
+            elif isinstance(cond, list):
+                return [process_nested(item) for item in cond]
+            return cond
+
+        # Final result: only one element left in stack, return after processing nested
+        result = stack[0] if stack else None
+        return process_nested(result) if result else None
+
+
+class QueryFilter:
+    '''
+    Query filter for replay buffer.
+    '''
+
+    def __init__(self, query_condition: QueryCondition) -> None:
+        self.query_condition = query_condition
+
+    def _get_field_value(self, row: DataRow, field: str) -> Any:
+        '''
+        Get field value from row.
+        '''
+        obj = row
+        for part in field.split('.'):
+            obj = getattr(obj, part, None)
+            if obj is None:
+                break
+        return obj
+
+    def _do_check(self, row: DataRow, condition: QueryCondition) -> bool:
+        """
+        check if row match condition
+        """
+        if condition is None:
+            return True
+        if "field" in condition and "op" in condition:
+            field_val = self._get_field_value(row, condition["field"])
+            op = condition["op"]
+            target_val = condition["value"]
+
+            if op == "eq":
+                return field_val == target_val
+            if op == "ne":
+                return field_val != target_val
+            if op == "gt":
+                return field_val > target_val
+            if op == "gte":
+                return field_val >= target_val
+            if op == "lt":
+                return field_val < target_val
+            if op == "lte":
+                return field_val <= target_val
+            if op == "in":
+                return field_val in target_val
+            if op == "not_in":
+                return field_val not in target_val
+            if op == "like":
+                return target_val in field_val
+            if op == "not_like":
+                return target_val not in field_val
+            if op == "is_null":
+                return field_val is None
+            if op == "is_not_null":
+                return field_val is not None
+
+            return False
+
+        elif "and_" in condition or "or_" in condition:
+            if "and_" in condition:
+                return all(self._do_check(row, c) for c in condition["and_"])
+            if "or_" in condition:
+                return any(self._do_check(row, c) for c in condition["or_"])
+            return False
+
+        return False
+
+    def check_condition(self, row: DataRow) -> bool:
+        """
+        check if row match condition
+        """
+        return self._do_check(row, self.query_condition)
+
+    def filter(self, rows: List[DataRow]) -> List[DataRow]:
+        """filter rows by condition
+        Args:
+            rows (List[DataRow]): List of rows to filter.
+            query_condition (QueryCondition): Query condition.
+        Returns:
+            List[DataRow]: List of rows that match the condition.
+        """
+        condition = self.query_condition
+        if not condition:
+            return rows
+        return [row for row in rows if self.check_condition(row)]