OmniEmbodied Framework API

This section provides detailed API documentation for all core framework components.

Evaluation Framework

EvaluationManager

The central coordinator for evaluation activities.

class evaluation.evaluation_manager.EvaluationManager(config_file: str, agent_type: str, task_type: str, scenario_selection: Dict[str, Any], custom_suffix: str = None)[source]

Bases: object

评测管理器 - 统一评测管理和场景级并行执行

__init__(config_file, agent_type, task_type, scenario_selection, custom_suffix=None)[source]

Initialize the evaluation manager.

Parameters:

config_file (str) – Configuration file identifier (e.g., “single_agent_config”)
agent_type (str) – Type of agent (“single” or “multi”)
task_type (str) – Type of task (“independent” or “collaborative”)
scenario_selection (dict) – Scenario selection configuration
custom_suffix (str) – Optional suffix for result files

run_evaluation()[source]

Execute the evaluation process.

Returns:: Evaluation results dictionary
Return type:: dict

__init__(config_file: str, agent_type: str, task_type: str, scenario_selection: Dict[str, Any], custom_suffix: str = None)[source]

初始化评测管理器

Parameters:

config_file – 配置文件名
agent_type – 智能体类型 (‘single’, ‘multi’)
task_type – 任务类型 (‘sequential’, ‘combined’, ‘independent’)
scenario_selection – 场景选择配置
custom_suffix – 自定义后缀

run_evaluation() → Dict[str, Any][source]: 运行评测

EvaluationInterface

High-level interface for running evaluations.

evaluation.evaluation_interface.EvaluationInterface.run_evaluation()

Execute evaluation with specified parameters.

Parameters:

config_file (str) – Configuration file identifier
agent_type (str) – Agent type (“single” or “multi”)
task_type (str) – Task type (“independent” or “collaborative”)
scenario_selection (dict) – Scenario selection parameters
custom_suffix (str) – Optional result file suffix

Returns:

Evaluation results

Return type:

dict

ScenarioSelector

Manages scenario selection and filtering.

class evaluation.scenario_selector.ScenarioSelector[source]

Bases: object

Scenario Selector - Simplified implementation

get_scenario_list(config, scenario_selection)[source]

Get filtered list of scenarios.

Parameters:

config (dict) – Configuration dictionary
scenario_selection (dict) – Selection criteria

Returns:

List of selected scenarios

Return type:

list

static get_scenario_list(config: Dict[str, Any], scenario_selection: Dict[str, Any] = None) → Dict[str, Any][source]

Get list of scenarios to evaluate and task filtering information

Parameters:

config – Configuration file
scenario_selection –
Scenario selection configuration {

’mode’: ‘all’, # ‘all’, ‘range’, ‘list’ ‘range’: {‘start’: ‘00001’, ‘end’: ‘00010’}, ‘list’: [‘00001’, ‘00003’, ‘00005’], ‘task_filter’: {

’categories’: [‘direct_command’, ‘attribute_reasoning’] # Task category filtering

}

}

Returns:

Contains scenario list and task filtering information

’scenarios’: Scenario ID list
’task_indices’: Task indices to execute in each scenario

Return type:

Dict[str, Any]

static parse_scenario_selection_string(scenarios_str: str) → Dict[str, Any][source]

Parse scenario selection string

Parameters:: scenarios_str – Scenario selection string - ‘all’: All scenarios - ‘00001-00010’: Range scenarios - ‘00001,00003,00005’: List scenarios - ‘00001’: Single scenario
Returns:: Scenario selection configuration
Return type:: Dict

static get_scenario_count(scenario_selection: Dict[str, Any] = None) → int[source]: 获取场景数量

static validate_scenario_selection(scenario_selection: Dict[str, Any]) → bool[source]: 验证场景选择配置的有效性

TaskExecutor

Executes individual tasks within scenarios.

class evaluation.task_executor.TaskExecutor(simulator, agent_adapter, trajectory_recorder: TrajectoryRecorder)[source]

Bases: object

任务执行器 - 执行单个任务的详细步骤

execute_task(scenario_data, task_config)[source]

Execute a single task.

Parameters:

scenario_data (dict) – Scenario information
task_config (dict) – Task configuration

Returns:

Task execution result

Return type:

dict

__init__(simulator, agent_adapter, trajectory_recorder: TrajectoryRecorder)[source]

初始化任务执行器

Parameters:

simulator – 模拟器实例
agent_adapter – 智能体适配器
trajectory_recorder – 轨迹记录器

execute_task(task: Dict[str, Any], task_index: int, max_steps: int = 50) → Dict[str, Any][source]

执行单个任务

Parameters:

task – 任务信息
task_index – 任务索引（从1开始）
max_steps – 最大步数

Returns:

任务执行结果

Return type:

Dict

Agent Framework

Single Agent

LLMAgent for single-agent scenarios.

class modes.single_agent.llm_agent.LLMAgent(*args: Any, **kwargs: Any)

Bases:

__init__(simulator, agent_id, config=None)

Initialize LLM agent.

Parameters:

simulator (SimulationEngine) – Simulation engine instance
agent_id (str) – Unique agent identifier
config (dict) – Optional agent configuration

decide_action(observation, available_actions)

Decide next action based on observation.

Parameters:

observation (str) – Current observation text
available_actions (list) – Available action types

Returns:

Selected action details

Return type:

dict

Multi-Agent System

CentralizedAgent for coordinated multi-agent scenarios.

class modes.centralized.centralized_agent.CentralizedAgent(*args: Any, **kwargs: Any)

Bases:

__init__(simulator, agent_id, config=None)

Initialize centralized agent coordinator.

Parameters:

simulator (SimulationEngine) – Simulation engine instance
agent_id (str) – Coordinator agent identifier
config (dict) – Optional coordination configuration

LLM Integration

LLM Factory

Factory function for creating LLM instances.

The configuration format for different LLM providers:

# OpenAI API configuration
llm_config = {
    "mode": "api",
    "api": {
        "provider": "openai",
        "model": "gpt-4",
        "api_key": "your-api-key"
    }
}

# Anthropic API configuration
llm_config = {
    "mode": "api",
    "api": {
        "provider": "anthropic",
        "model": "claude-3-sonnet",
        "api_key": "your-api-key"
    }
}

BaseLLM

Base class for LLM implementations.

class llm.base_llm.BaseLLM(config: Dict[str, Any])[source]

Bases: ABC

大语言模型基类，定义与LLM交互的通用接口

__init__(config: Dict[str, Any])[source]

初始化LLM

Parameters:: config – 配置字典，包含模型参数

abstractmethod generate(prompt: str, system_message: str | None = None, temperature: float | None = None, max_tokens: int | None = None, **kwargs) → str[source]

生成文本响应

Parameters:

prompt – 用户输入提示
system_message – 系统消息，可选
temperature – 温度参数，控制随机性，可选
max_tokens – 最大生成token数，可选
**kwargs – 额外参数

Returns:

生成的文本响应

Return type:

str

abstractmethod generate_chat(messages: List[Dict[str, str]], temperature: float | None = None, max_tokens: int | None = None, **kwargs) → str[source]

生成多轮对话响应

Parameters:

messages – 消息列表，每个消息是包含’role’和’content’的字典
temperature – 温度参数，控制随机性，可选
max_tokens – 最大生成token数，可选
**kwargs – 额外参数

Returns:

生成的文本响应

Return type:

str

get_config() → Dict[str, Any][source]: 获取当前配置

APILLM

API-based LLM client implementation.

class llm.api_llm.APILLM(*args: Any, **kwargs: Any)

Bases:

generate(messages, **kwargs)

Generate response from messages.

Parameters:

messages (list) – List of message dictionaries
kwargs – Additional generation parameters

Returns:

Generated response text

Return type:

str

Configuration System

ConfigManager

Manages configuration loading and merging.

class config.config_manager.ConfigManager(config_root: str | None = None)[source]

Bases: object

统一配置管理器支持配置继承、环境变量替换、命令行参数覆盖等功能

__init__(config_dir=None)[source]

Initialize configuration manager.

Parameters:: config_dir (str) – Directory containing configuration files

load_config(config_name, config_override=None)[source]

Load configuration by name.

Parameters:

config_name (str) – Configuration identifier (e.g., “single_agent_config”)
config_override (dict) – Optional override values

Returns:

Merged configuration dictionary

Return type:

dict

__init__(config_root: str | None = None)[source]

初始化配置管理器

Parameters:: config_root – 配置根目录，默认为当前文件所在目录

load_config(config_name: str, force_reload: bool = False) → Dict[str, Any][source]

加载配置文件

Parameters:

config_name – 配置名称
force_reload – 是否强制重新加载

Returns:

配置字典

Return type:

Dict[str, Any]

get_config(config_name: str, reload: bool = False) → Dict[str, Any][source]

获取配置

Parameters:

config_name – 配置名称
reload – 是否重新加载

Returns:

配置字典

Return type:

Dict[str, Any]

set_runtime_override(config_name: str, key: str, value: Any)[source]

设置运行时配置覆盖

Parameters:

config_name – 配置名称
key – 配置键，支持点号分隔的嵌套路径
value – 配置值

set_runtime_overrides_from_dict(config_name: str, overrides: Dict[str, Any])[source]

从字典设置运行时配置覆盖

Parameters:

config_name – 配置名称
overrides – 覆盖配置字典

clear_runtime_overrides(config_name: str | None = None)[source]

清除运行时配置覆盖

Parameters:: config_name – 配置名称，如果为None则清除所有覆盖

get_config_section(config_name: str, section: str) → Any[source]

获取配置的特定部分

Parameters:

config_name – 配置名称
section – 配置节名称，支持点号分隔的嵌套路径

Returns:

配置值

Return type:

Any

Raises:

KeyError – 当配置项不存在时

update_config(config_name: str, updates: Dict[str, Any]) → Dict[str, Any][source]

更新配置（仅内存中）

Parameters:

config_name – 配置名称
updates – 更新内容

Returns:

更新后的配置字典

Return type:

Dict[str, Any]

save_config(config_name: str, config_dir: str | None = None) → bool[source]

保存配置到文件

Parameters:

config_name – 配置名称
config_dir – 保存目录，如果未指定则使用原目录

Returns:

是否成功保存

Return type:

bool

get_data_dir(config_name: str, dataset_name: str) → str[source]

获取数据集目录路径

Parameters:

config_name – 配置名称
dataset_name – 数据集名称（必需）

Returns:

数据集目录的绝对路径

Return type:

str

get_scene_dir(config_name: str, dataset_name: str) → str[source]

获取场景目录路径

Parameters:

config_name – 配置名称
dataset_name – 数据集名称（必需）

Returns:

场景目录的绝对路径

Return type:

str

get_task_dir(config_name: str, dataset_name: str) → str[source]

获取任务目录路径

Parameters:

config_name – 配置名称
dataset_name – 数据集名称（必需）

Returns:

任务目录的绝对路径

Return type:

str

list_datasets(config_name: str) → List[str][source]

列出所有可用的数据集

Parameters:: config_name – 配置名称
Returns:: 数据集名称列表
Return type:: List[str]

get_subdir_name(config_name: str, subdir_type: str) → str[source]

获取子目录名称

Parameters:

config_name – 配置名称
subdir_type – 子目录类型 (‘scene’, ‘task’)

Returns:

子目录名称

Return type:

str

list_configs() → List[str][source]: 列出所有可用的配置文件

clear_cache()[source]: 清空配置缓存

Configuration structure for different components:

# Agent configuration
agent_config:
  max_history: 20
  temperature: 0.7
  action_timeout: 30

# LLM configuration
llm_config:
  mode: "api"
  api:
    provider: "openai"
    model: "gpt-4"

# Evaluation configuration
evaluation:
  max_steps: 50
  timeout: 300
  parallel_scenarios: 5

Utilities

Logger

Basic logging utilities.

class utils.logger.ColorizedFormatter(*args: Any, **kwargs: Any)[source]: Bases:

utils.logger.get_logger(name: str) → Logger[source]

Get a logger with the specified name.

Parameters:: name – Name of the logger
Returns:: Logger instance

Get logger instance by name.

Parameters:: name (str) – Logger name
Returns:: Logger instance
Return type:: logging.Logger

utils.logger.setup_logging(name: str, level: str = 'INFO', log_file: str | None = None, console_output: bool = True, format_style: str = 'simple', enable_colors: bool = True) → Logger[source]

设置日志记录器

Parameters:

name – 记录器名称
level – 日志级别 (DEBUG, INFO, WARNING, ERROR, CRITICAL)
log_file – 日志文件路径，如果为None则不写入文件
console_output – 是否输出到控制台
format_style – 格式样式 (“simple”, “detailed”, “debug”)
enable_colors – 是否启用颜色（仅影响控制台输出）

Returns:

配置好的Logger对象

Setup logging configuration.

Parameters:

name (str) – Logger name
level (str) – Logging level
log_file (str) – Optional log file path
console_output (bool) – Enable console output

Returns:

Configured logger

Return type:

logging.Logger

Data Generation Logger

Specialized logging utilities for data generation.

data_generation.utils.logger.get_logger()[source]: Get data generation logger instance.

data_generation.utils.logger.log_raw_response(generator_type: str, item_id: str, thread_id: int, response: str) → None[source]

Log raw LLM responses to separate files.

Log raw LLM responses.

Parameters:

generator_type (str) – Type of generator
item_id (str) – Item identifier
thread_id (int) – Thread identifier
response (str) – Response text

data_generation.utils.logger.log_processing(logger: Logger, message: str)[source]

Log processing message with special formatting.

Log processing messages.

Parameters:

logger (logging.Logger) – Logger instance
message (str) – Message to log

data_generation.utils.logger.log_success(logger: Logger, message: str)[source]

Log success message with special formatting.

Log success messages.

Parameters:

logger (logging.Logger) – Logger instance
message (str) – Success message

Data Generation Framework

BaseGenerator

Base class for data generators.

class data_generation.generators.base_generator.BaseGenerator(*args: Any, **kwargs: Any)

Bases:

__init__(generator_type, config_override=None)

Initialize base generator.

Parameters:

generator_type (str) – Type of generator
config_override (dict) – Optional configuration overrides

generate(num_items)

Generate specified number of items.

Parameters:: num_items (int) – Number of items to generate
Returns:: Generated items
Return type:: list

TaskGenerator

Generates task definitions.

class data_generation.generators.task_generator.TaskGenerator(*args: Any, **kwargs: Any): Bases:

SceneGenerator

Generates scene descriptions.

class data_generation.generators.scene_generator.SceneGenerator(*args: Any, **kwargs: Any): Bases:

Usage Examples

Basic Evaluation

from evaluation.evaluation_interface import EvaluationInterface

# Run single-agent evaluation
result = EvaluationInterface.run_evaluation(
    config_file="single_agent_config",  # ConfigManager resolves full path
    agent_type="single",
    task_type="independent",
    scenario_selection={
        "dataset_type": "single",
        "scenario_range": {"start": "00001", "end": "00050"}
    }
)

print(f"Success rate: {result.get('success_rate', 0):.2%}")

LLM Integration

from llm.llm_factory import create_llm_from_config

# Create LLM instance
llm_config = {
    "mode": "api",
    "api": {
        "provider": "openai",
        "model": "gpt-4",
        "api_key": "your-api-key"
    }
}

llm = create_llm_from_config(llm_config)
response = llm.generate([{"role": "user", "content": "Hello!"}])

Custom Agent

from modes.single_agent.llm_agent import LLMAgent
from OmniSimulator.core.engine import SimulationEngine

# Initialize simulation
engine = SimulationEngine()

# Create custom agent
agent = LLMAgent(
    simulator=engine,
    agent_id="custom_agent",
    config={
        "temperature": 0.7,
        "max_history": 20
    }
)

Configuration Management

from config.config_manager import ConfigManager

# Load configuration
config_manager = ConfigManager()
config = config_manager.load_config(
    "single_agent_config",  # Short name - automatically resolved
    config_override={
        "evaluation": {"max_steps": 100}
    }
)

For more examples and detailed tutorials, see Basic Simulation Examples.