Upload 11 files

aworld/core/README.md

@@ -0,0 +1,13 @@
+# Core Components
+
+Common functionality and system components.
+
+- `agent/`: Base agent for sub agents and description of already registered agents.
+- `envs/`: The environment and its tools, as well as the related actions of the tools. It is a three-level and one to
+  many structure.
+- `context`: to be continued
+- `swarm`: Interactive collaboration in the topology structure of multiple agents that interact with the environment tools.
+- `task`:  Structure containing datasets, agents, tools, metrics, outputs, etc.
+- `runner`: Complete a runnable specific workflow and obtain results.
+
+![Architecture](../../readme_assets/framework_arch.png)

aworld/core/__init__.py

@@ -0,0 +1,8 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+try:
+    from aworld.agents import agent_desc
+    from examples.tools import tool_action_desc
+except:
+    pass

aworld/core/common.py

@@ -0,0 +1,92 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+from pydantic import BaseModel
+from typing import Dict, Any, Optional, Union, List
+
+from aworld.config import ConfigDict
+from aworld.core.memory import MemoryItem
+
+Config = Union[Dict[str, Any], ConfigDict, BaseModel]
+
+
+class ActionResult(BaseModel):
+    """Result of executing an action by use tool."""
+    is_done: bool = False
+    success: bool = False
+    content: Any = None
+    error: str = None
+    keep: bool = False
+    action_name: str = None
+    tool_name: str = None
+    # llm tool id
+    tool_id: str = None
+    metadata: Optional[Dict[str, Any]] = {}
+
+
+class Observation(BaseModel):
+    """Observation information is obtained from the tools or transformed from the actions made by agents.
+
+    It can be an agent(as a tool) in the swarm or a tool in the virtual environment.
+    """
+    # default is None, means the main virtual environment or swarm
+    container_id: Optional[str] = None
+    # Observer who obtains observation, default is None for compatible, means an agent name or a tool name
+    observer: Optional[str] = None
+    # default is None for compatible, means with its action/ability name of an agent or a tool
+    # NOTE: The only ability of an agent as a tool is handoffs
+    ability: Optional[str] = None
+    # The agent wants the observation to be created, default is None for compatible.
+    from_agent_name: Optional[str] = None
+    # To which agent should the observation be given, default is None for compatible.
+    to_agent_name: Optional[str] = None
+    # general info for agent
+    content: Optional[Any] = None
+    # dom_tree is a str or DomTree object
+    dom_tree: Optional[Union[str, Any]] = None
+    image: Optional[str] = None  # base64
+    action_result: Optional[List[ActionResult]] = []
+    # for video or image list
+    images: Optional[List[str]] = []
+    # extend key value pair. `done` is an internal key
+    info: Optional[Dict[str, Any]] = {}
+
+
+class StatefulObservation(Observation):
+    """Observations with contextual states."""
+    context: List[MemoryItem]
+
+
+class ParamInfo(BaseModel):
+    name: str | None = None
+    type: str = "str"
+    required: bool = False
+    desc: str = None
+    default_value: Any = None
+
+
+class ToolActionInfo(BaseModel):
+    name: str
+    input_params: Dict[str, ParamInfo] = {}
+    desc: str = None
+
+
+class ActionModel(BaseModel):
+    tool_name: Optional[str] = None
+    tool_id: Optional[str] = None
+    # agent name
+    agent_name: Optional[str] = None
+    # action_name is a tool action name by agent policy.
+    action_name: Optional[str] = None
+    params: Optional[Dict[str, Any]] = {}
+    policy_info: Optional[Any] = None
+
+
+class TaskItem(BaseModel):
+    data: Any
+    msg: str = None
+    stop: bool = False
+    success: bool = False
+    action_name: Optional[str] = None
+    params: Optional[Dict[str, Any]] = {}
+    policy_info: Optional[Any] = None

aworld/core/exceptions.py

@@ -0,0 +1,7 @@
+class AworldException(Exception):
+    """Exception Aworld."""
+
+    message: str
+
+    def __init__(self, message: str):
+        self.message = message

aworld/core/experiment.py

@@ -0,0 +1,7 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+from pydantic import BaseModel
+
+
+class Experiment(BaseModel):
+    pass

aworld/core/factory.py

@@ -0,0 +1,91 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+from typing import TypeVar, Generic, Dict, Any
+
+from aworld.logs.util import logger
+
+T = TypeVar("T")
+
+
+class Factory(Generic[T]):
+    """The base generic class that is used to define a factory(local) for various objects, with a parameterized types: T."""
+
+    def __init__(self, type_name: str = None):
+        self._type = type_name
+        self._cls: Dict[str, T] = {}
+        self._desc: Dict[str, str] = {}
+        self._asyn: Dict[str, bool] = {}
+        self._ext_info: Dict[str, Dict[Any, Any]] = {}
+
+    def __call__(self, name: str, asyn: bool = False, **kwargs):
+        """Create the special type object instance by name. If not found, raise ValueError or construct object instance.
+
+        Returns:
+            Object instance.
+        """
+        exception = kwargs.pop('except', False)
+        if not name in self._cls:
+            if not exception:
+                return None
+
+            if self._type is None:
+                raise ValueError(f"Unknown factory object type: '{self._type}'")
+            raise ValueError(f"Unknown {self._type}: '{name}'")
+        name = "async_" + name if asyn else name
+        return self._cls[name](**kwargs)
+
+    def __iter__(self):
+        for name in self._cls:
+            yield name
+
+    def __contains__(self, name: str) -> bool:
+        """Whether the name in the factory."""
+        return name in self._cls
+
+    def get_class(self, name: str, asyn: bool = False) -> T | None:
+        """Get the object instance by name."""
+        return self._cls.get(name, None)
+
+    def count(self) -> int:
+        """Total number in the special type object factory."""
+        return len(self._cls)
+
+    def desc(self, name: str, asyn: bool = False) -> str:
+        """Obtain the description by name."""
+        name = "async_" + name if asyn else name
+        return self._desc.get(name, "")
+
+    def get_ext_info(self, name: str, asyn: bool = False) -> Dict[Any, Any]:
+        """Obtain the extent info by name."""
+        name = "async_" + name if asyn else name
+        return self._ext_info.get(name, {})
+
+    def register(self, name: str, desc: str, **kwargs):
+        def func(cls):
+            asyn = kwargs.pop("asyn", False)
+            prefix = "async_" if asyn else ""
+            if len(prefix) > 0:
+                logger.debug(f"{name} has an async type, will add `async_` prefix.")
+
+            if prefix + name in self._cls:
+                equal = True
+                if asyn:
+                    equal = self._asyn[name] == asyn
+                if equal:
+                    logger.warning(f"{name} already in {self._type} factory, will override it.")
+
+            self._asyn[name] = asyn
+            self._cls[prefix + name] = cls
+            self._desc[prefix + name] = desc
+            self._ext_info[prefix + name] = kwargs
+            return cls
+
+        return func
+
+    def unregister(self, name: str):
+        if name in self._cls:
+            logger.warning(f"unregister {name} in the {self._type} factory.")
+            del self._cls[name]
+            del self._desc[name]
+            del self._asyn[name]

aworld/core/llm_provider_base.py

@@ -0,0 +1,217 @@
+import abc
+from collections import Counter
+from typing import (
+    Any,
+    List,
+    Dict,
+    Generator,
+    AsyncGenerator,
+)
+
+from aworld.models.model_response import ModelResponse
+
+
+class LLMProviderBase(abc.ABC):
+    """Base class for large language model providers, defines unified interface."""
+
+    def __init__(self,
+                 api_key: str = None,
+                 base_url: str = None,
+                 model_name: str = None,
+                 sync_enabled: bool = None,
+                 async_enabled: bool = None,
+                 **kwargs):
+        """Initialize provider.
+
+        Args:
+            api_key: API key.
+            base_url: Service URL.
+            model_name: Model name.
+            **kwargs: Other parameters.
+        """
+        self.api_key = api_key
+        self.base_url = base_url
+        self.model_name = model_name
+        self.kwargs = kwargs
+        # Determine whether to initialize sync and async providers
+        self.need_sync = sync_enabled if sync_enabled is not None else async_enabled is not True
+        self.need_async = async_enabled if async_enabled is not None else sync_enabled is not True
+
+        # Initialize providers based on flags
+        self.provider = self._init_provider() if self.need_sync else None
+        self.async_provider = self._init_async_provider() if self.need_async else None
+        self.stream_tool_buffer=[]
+
+    @abc.abstractmethod
+    def _init_provider(self):
+        """Initialize specific provider instance, to be implemented by subclasses.
+        Returns:
+            Provider instance.
+        """
+
+    def _init_async_provider(self):
+        """Initialize async provider instance. Optional for subclasses that don't need async support.
+        Only called if async provider is needed.
+
+        Returns:
+            Async provider instance.
+        """
+        return None
+
+    @classmethod
+    def supported_models(cls) -> list[str]:
+        return []
+
+    def preprocess_messages(self, messages: List[Dict[str, str]]) -> Any:
+        """Preprocess messages, convert OpenAI format messages to specific provider required format.
+
+        Args:
+            messages: OpenAI format message list [{"role": "system", "content": "..."}, ...].
+
+        Returns:
+            Converted messages, format determined by specific provider.
+        """
+        return messages
+
+    @abc.abstractmethod
+    def postprocess_response(self, response: Any) -> ModelResponse:
+        """Post-process response, convert provider response to unified ModelResponse.
+
+        Args:
+            response: Original response from provider.
+
+        Returns:
+            ModelResponse: Unified format response object.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+
+    def postprocess_stream_response(self, chunk: Any) -> ModelResponse:
+        """Post-process streaming response chunk, convert provider chunk to unified ModelResponse.
+
+        Args:
+            chunk: Original response chunk from provider.
+            
+        Returns:
+            ModelResponse: Unified format response object for the chunk.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+
+    async def acompletion(self,
+                          messages: List[Dict[str, str]],
+                          temperature: float = 0.0,
+                          max_tokens: int = None,
+                          stop: List[str] = None,
+                          **kwargs) -> ModelResponse:
+        """Asynchronously call model to generate response.
+        
+        Args:
+            messages: Message list, format is [{"role": "system", "content": "..."}, {"role": "user", "content": "..."}].
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            ModelResponse: Unified model response object.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+            RuntimeError: When async provider is not initialized.
+        """
+        if not self.async_provider:
+            raise RuntimeError(
+                "Async provider not initialized. Make sure 'async_enabled' parameter is set to True in initialization.")
+
+
+    @abc.abstractmethod
+    def completion(self,
+                   messages: List[Dict[str, str]],
+                   temperature: float = 0.0,
+                   max_tokens: int = None,
+                   stop: List[str] = None,
+                   **kwargs) -> ModelResponse:
+        """Synchronously call model to generate response.
+
+        Args:
+            messages: Message list, format is [{"role": "system", "content": "..."}, {"role": "user", "content": "..."}].
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            ModelResponse: Unified model response object.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+            RuntimeError: When sync provider is not initialized.
+        """
+
+    def stream_completion(self,
+                          messages: List[Dict[str, str]],
+                          temperature: float = 0.0,
+                          max_tokens: int = None,
+                          stop: List[str] = None,
+                          **kwargs) -> Generator[ModelResponse, None, None]:
+        """Synchronously call model to generate streaming response.
+
+        Args:
+            messages: Message list, format is [{"role": "system", "content": "..."}, {"role": "user", "content": "..."}].
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            Generator yielding ModelResponse chunks.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+            RuntimeError: When sync provider is not initialized.
+        """
+
+    async def astream_completion(self,
+                                 messages: List[Dict[str, str]],
+                                 temperature: float = 0.0,
+                                 max_tokens: int = None,
+                                 stop: List[str] = None,
+                                 **kwargs) -> AsyncGenerator[ModelResponse, None]:
+        """Asynchronously call model to generate streaming response.
+
+        Args:
+            messages: Message list, format is [{"role": "system", "content": "..."}, {"role": "user", "content": "..."}].
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            AsyncGenerator yielding ModelResponse chunks.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+            RuntimeError: When async provider is not initialized.
+        """
+
+    def _accumulate_chunk_usage(self, usage: Dict[str, int], chunk_usage: Dict[str, int]):
+        """Accumulate usage statistics from chunk into the main usage dictionary.
+
+        Args:
+            usage: Dictionary to accumulate usage into (will be modified)
+            chunk_usage: Usage statistics from the current chunk
+        """
+        if not chunk_usage:
+            return
+
+        usage.update(dict(Counter(usage) + Counter(chunk_usage)))
+
+    def speech_to_text(self, audio_file, language, prompt, **kwargs) -> ModelResponse:
+        pass
+
+    async def aspeech_to_text(self, audio_file, language, prompt, **kwargs) -> ModelResponse:
+        pass
+

aworld/core/memory.py

@@ -0,0 +1,305 @@
+import datetime
+import uuid
+from abc import ABC, abstractmethod
+from typing import Optional, Any, Literal, Union
+
+from pydantic import BaseModel, Field, ConfigDict
+
+from aworld.models.llm import LLMModel
+
+
+class MemoryItem(BaseModel):
+    id: str = Field(description="id")
+    content: Any = Field(description="content")
+    created_at: Optional[str] = Field(None, description="created at")
+    updated_at: Optional[str] = Field(None, description="updated at")
+    metadata: dict = Field(
+        description="metadata, use to store additional information, such as user_id, agent_id, run_id, task_id, etc.")
+    tags: list[str] = Field(description="tags")
+    histories: list["MemoryItem"] = Field(default_factory=list)
+    deleted: bool = Field(default=False)
+    memory_type: Literal["init", "message", "summary", "agent_experience", "user_profile"] = Field(default="message")
+    version: int = Field(description="version")
+
+    def __init__(self, **data):
+        # Set default values for optional fields
+        if "id" not in data:
+            data["id"] = str(uuid.uuid4())
+        if "created_at" not in data:
+            data["created_at"] = datetime.datetime.now().isoformat()
+        if "updated_at" not in data:
+            data["updated_at"] = data["created_at"]
+        if "metadata" not in data:
+            data["metadata"] = {}
+        if "tags" not in data:
+            data["tags"] = []
+        if "version" not in data:
+            data["version"] = 1
+
+        super().__init__(**data)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "MemoryItem":
+        """Create a MemoryItem instance from a dictionary.
+
+        Args:
+            data (dict): A dictionary containing the memory item data.
+
+        Returns:
+            MemoryItem: An instance of MemoryItem.
+        """
+        return cls(**data)
+
+
+class MemoryStore(ABC):
+    """
+    Memory store interface for messages history
+    """
+
+    @abstractmethod
+    def add(self, memory_item: MemoryItem):
+        pass
+
+    @abstractmethod
+    def get(self, memory_id) -> Optional[MemoryItem]:
+        pass
+
+    @abstractmethod
+    def get_first(self, filters: dict = None) -> Optional[MemoryItem]:
+        pass
+
+    @abstractmethod
+    def total_rounds(self, filters: dict = None) -> int:
+        pass
+
+    @abstractmethod
+    def get_all(self, filters: dict = None) -> list[MemoryItem]:
+        pass
+
+    @abstractmethod
+    def get_last_n(self, last_rounds, filters: dict = None) -> list[MemoryItem]:
+        pass
+
+    @abstractmethod
+    def update(self, memory_item: MemoryItem):
+        pass
+
+    @abstractmethod
+    def delete(self, memory_id):
+        pass
+
+    @abstractmethod
+    def history(self, memory_id) -> list[MemoryItem] | None:
+        pass
+
+
+class MemoryBase(ABC):
+
+    @abstractmethod
+    def get(self, memory_id) -> Optional[MemoryItem]:
+        """Get item in memory by ID.
+
+        Args:
+            memory_id (str): ID of the memory to retrieve.
+
+        Returns:
+            dict: Retrieved memory.
+        """
+
+    @abstractmethod
+    def get_all(self, filters: dict = None) -> Optional[list[MemoryItem]]:
+        """List all items in memory store.
+
+        Args:
+            filters (dict, optional): Filters to apply to the search. Defaults to None.
+            - user_id (str, optional): ID of the user to search for. Defaults to None.
+            - agent_id (str, optional): ID of the agent to search for. Defaults to None.
+            - session_id (str, optional): ID of the session to search for. Defaults to None.
+
+        Returns:
+            list: List of all memories.
+        """
+
+    @abstractmethod
+    def get_last_n(self, last_rounds, filters: dict = None) -> Optional[list[MemoryItem]]:
+        """get last_rounds memories.
+
+        Args:
+            last_rounds (int): Number of last rounds to retrieve.
+            filters (dict, optional): Filters to apply to the search. Defaults to None.
+            - user_id (str, optional): ID of the user to search for. Defaults to None.
+            - agent_id (str, optional): ID of the agent to search for. Defaults to None.
+            - session_id (str, optional): ID of the session to search for. Defaults to None.
+        Returns:
+            list: List of latest memories.
+        """
+
+    @abstractmethod
+    def search(self, query, limit=100, filters=None) -> Optional[list[MemoryItem]]:
+        """
+        Search for memories.
+        Hybrid search: Retrieve memories from vector store and memory store.
+
+
+        Args:
+            query (str): Query to search for.
+            limit (int, optional): Limit the number of results. Defaults to 100.
+            filters (dict, optional): Filters to apply to the search. Defaults to None.
+            - user_id (str, optional): ID of the user to search for. Defaults to None.
+            - agent_id (str, optional): ID of the agent to search for. Defaults to None.
+            - session_id (str, optional): ID of the session to search for. Defaults to None.
+
+        Returns:
+            list: List of search results.
+        """
+
+    @abstractmethod
+    def add(self, memory_item: MemoryItem, filters: dict = None):
+        """Add memory in the memory store.
+
+        Step 1: Add memory to memory store
+        Step 2: Add memory to vector store
+
+        Args:
+            memory_item (MemoryItem): memory item.
+            metadata (dict, optional): metadata to add.
+             - user_id (str, optional): ID of the user to search for. Defaults to None.
+             - agent_id (str, optional): ID of the agent to search for. Defaults to None.
+             - session_id (str, optional): ID of the session to search for. Defaults to None.
+            tags (list, optional): tags to add.
+            memory_type (str, optional): memory type.
+            version (int, optional): version of the memory.
+        """
+
+    @abstractmethod
+    def update(self, memory_item: MemoryItem):
+        """Update a memory by ID.
+
+        Args:
+            memory_item (MemoryItem): memory item.
+
+        Returns:
+            dict: Updated memory.
+        """
+
+    @abstractmethod
+    async def async_gen_cur_round_summary(self, to_be_summary: MemoryItem, filters: dict, last_rounds: int) -> str:
+        """A tool for reducing the context length of the current round.
+
+        Step 1: Retrieve historical conversation content based on filters and last_rounds
+        Step 2: Extract current round content and most relevant historical content  
+        Step 3: Generate corresponding summary for the current round
+
+        Args:
+            to_be_summary (MemoryItem): msg to summary.
+            filters (dict): filters to get memory list.
+            last_rounds (int): last rounds of memory list.
+
+        Returns:
+            str: summary memory.
+        """
+
+    @abstractmethod
+    async def async_gen_multi_rounds_summary(self, to_be_summary: list[MemoryItem]) -> str:
+        """A tool for summarizing the list of memory item.
+
+        Args:
+            to_be_summary (list[MemoryItem]): the list of memory item.
+        """
+
+    @abstractmethod
+    async def async_gen_summary(self, filters: dict, last_rounds: int) -> str:
+        """A tool for summarizing the conversation history.
+
+        Step 1: Retrieve historical conversation content based on filters and last_rounds
+        Step 2: Generate corresponding summary for conversation history
+
+        Args:
+            filters (dict): filters to get memory list.
+            last_rounds (int): last rounds of memory list.
+        """
+
+    @abstractmethod
+    def delete(self, memory_id):
+        """Delete a memory by ID.
+
+        Args:
+            memory_id (str): ID of the memory to delete.
+        """
+
+
+SUMMARY_PROMPT = """
+You are a helpful assistant that summarizes the conversation history.
+- Summarize the following text in one clear and concise paragraph, capturing the key ideas without missing critical points. 
+- Ensure the summary is easy to understand and avoids excessive detail.
+
+Here are the content: 
+{context}
+"""
+
+
+class MemoryConfig(BaseModel):
+    """Configuration for procedural memory."""
+
+    model_config = ConfigDict(
+        from_attributes=True, validate_default=True, revalidate_instances='always', validate_assignment=True,
+        arbitrary_types_allowed=True
+    )
+
+    # Memory Config
+    provider: Literal['inmemory', 'mem0'] = 'inmemory'
+    enable_summary: bool = Field(default=False, description="enable_summary use llm to create summary memory")
+    summary_rounds: int = Field(default=5, description="rounds of message msg; when the number of messages is greater than the summary_rounds, the summary will be created")
+    summary_single_context_length: int = Field(default=4000, description=" when the content length is greater than the summary_single_context_length, the summary will be created")
+    summary_prompt: str = Field(default=SUMMARY_PROMPT, description="summary prompt")
+
+    # Embedder settings
+    embedder_provider: Literal['openai', 'gemini', 'ollama', 'huggingface'] = 'huggingface'
+    embedder_model: str = Field(min_length=2, default='all-MiniLM-L6-v2')
+    embedder_dims: int = Field(default=384, gt=10, lt=10000)
+
+    # LLM settings - the LLM instance can be passed separately
+    llm_provider: Literal['openai', 'langchain'] = 'langchain'
+    llm_instance: Optional[Union[LLMModel]] = None
+
+    # Vector store settings
+    vector_store_provider: Literal['faiss'] = 'faiss'
+    vector_store_base_path: str = Field(default='/tmp/mem0_aworld')
+
+    @property
+    def vector_store_path(self) -> str:
+        """Returns the full vector store path for the current configuration. e.g. /tmp/mem0_384_faiss"""
+        return f'{self.vector_store_base_path}_{self.embedder_dims}_{self.vector_store_provider}'
+
+    @property
+    def embedder_config_dict(self) -> dict[str, Any]:
+        """Returns the embedder configuration dictionary."""
+        return {
+            'provider': self.embedder_provider,
+            'config': {'model': self.embedder_model, 'embedding_dims': self.embedder_dims},
+        }
+
+    @property
+    def llm_config_dict(self) -> dict[str, Any]:
+        """Returns the LLM configuration dictionary."""
+        return {'provider': self.llm_provider, 'config': {'model': self.llm_instance}}
+
+    @property
+    def vector_store_config_dict(self) -> dict[str, Any]:
+        """Returns the vector store configuration dictionary."""
+        return {
+            'provider': self.vector_store_provider,
+            'config': {
+                'embedding_model_dims': self.embedder_dims,
+                'path': self.vector_store_path,
+            },
+        }
+
+    @property
+    def full_config_dict(self) -> dict[str, dict[str, Any]]:
+        """Returns the complete configuration dictionary for Mem0."""
+        return {
+            'embedder': self.embedder_config_dict,
+            'llm': self.llm_config_dict,
+            'vector_store': self.vector_store_config_dict,
+        }

aworld/core/runtime_engine.py

@@ -0,0 +1,216 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import abc
+import inspect
+import os
+import asyncio
+from concurrent.futures import Future
+from concurrent.futures.process import ProcessPoolExecutor
+from types import MethodType
+from typing import List, Callable, Any, Dict
+
+from aworld.config import RunConfig, ConfigDict
+from aworld.logs.util import logger
+from aworld.utils.common import sync_exec
+
+LOCAL = "local"
+SPARK = "spark"
+RAY = "ray"
+K8S = "k8s"
+
+
+class RuntimeEngine(object):
+    """Lightweight wrapper of computing engine runtime."""
+
+    __metaclass__ = abc.ABCMeta
+
+    def __init__(self, conf: RunConfig):
+        """Engine runtime instance initialize."""
+        self.conf = ConfigDict(conf.model_dump())
+        self.runtime = None
+        register(conf.name, self)
+
+        # Initialize clients running on top of distributed computing engines
+        pass
+
+    def build_engine(self) -> 'RuntimeEngine':
+        """Create computing engine runtime.
+
+        If create more times in the same runtime instance, will get the same engine instance, like getOrCreate.
+        """
+        if self.runtime is not None:
+            return self
+        self._build_engine()
+        return self
+
+    @abc.abstractmethod
+    def _build_engine(self) -> None:
+        raise NotImplementedError("Base _build_engine not implemented!")
+
+    @abc.abstractmethod
+    def broadcast(self, data: Any):
+        """Broadcast the data to all workers."""
+
+    @abc.abstractmethod
+    async def execute(self, funcs: List[Callable[..., Any]], *args, **kwargs) -> Dict[str, Any]:
+        """Submission focuses on the execution of stateless tasks on the special engine cluster."""
+        raise NotImplementedError("Base task execute not implemented!")
+
+    def pre_execute(self):
+        """Define the pre execution logic."""
+        pass
+
+    def post_execute(self):
+        """Define the post execution logic."""
+        pass
+
+
+class LocalRuntime(RuntimeEngine):
+    """Local runtime key is 'local', and execute tasks in local machine.
+
+    Local runtime is used to verify or test locally.
+    """
+
+    def _build_engine(self):
+        self.runtime = self
+
+    def func_wrapper(self, func, *args, **kwargs):
+        """Function is used to adapter computing form."""
+
+        if inspect.iscoroutinefunction(func):
+            res = sync_exec(func, *args, **kwargs)
+        else:
+            res = func(*args, **kwargs)
+        return res
+
+    async def execute(self, funcs: List[Callable[..., Any]], *args, **kwargs) -> Dict[str, Any]:
+        # opt of the one task process
+        if len(funcs) == 1:
+            func = funcs[0]
+            if inspect.iscoroutinefunction(func):
+                res = await func(*args, **kwargs)
+            else:
+                res = func(*args, **kwargs)
+            return {res.id: res}
+
+        num_executor = self.conf.get('worker_num', os.cpu_count() - 1)
+        num_process = len(funcs)
+        if num_process > num_executor:
+            num_process = num_executor
+
+        if num_process <= 0:
+            num_process = 1
+
+        futures = []
+        with ProcessPoolExecutor(num_process) as pool:
+            for func in funcs:
+                futures.append(pool.submit(self.func_wrapper, func, *args, **kwargs))
+
+        results = {}
+        for future in futures:
+            future: Future = future
+            res = future.result()
+            results[res.id] = res
+        return results
+
+
+class K8sRuntime(LocalRuntime):
+    """K8s runtime key is 'k8s', and execute tasks in kubernetes cluster."""
+
+
+class KubernetesRuntime(LocalRuntime):
+    """kubernetes runtime key is 'kubernetes', and execute tasks in kubernetes cluster."""
+
+
+class SparkRuntime(RuntimeEngine):
+    """Spark runtime key is 'spark', and execute tasks in spark cluster.
+
+    Note: Spark runtime must in driver end.
+    """
+
+    def __init__(self, engine_options):
+        super(SparkRuntime, self).__init__(engine_options)
+
+    def _build_engine(self):
+        from pyspark.sql import SparkSession
+
+        conf = self.conf
+        is_local = conf.get('is_local', True)
+        logger.info('build runtime is_local:{}'.format(is_local))
+        spark_builder = SparkSession.builder
+        if is_local:
+            if 'PYSPARK_PYTHON' not in os.environ:
+                import sys
+                os.environ['PYSPARK_PYTHON'] = sys.executable
+
+            spark_builder = spark_builder.master('local[1]').config('spark.executor.instances', '1')
+
+        self.runtime = spark_builder.appName(conf.get('job_name', 'aworld_spark_job')).getOrCreate()
+
+    def args_process(self, *args):
+        re_args = []
+        for arg in args:
+            if arg:
+                options = self.runtime.sparkContext.broadcast(arg)
+                arg = options.value
+            re_args.append(arg)
+        return re_args
+
+    async def execute(self, funcs: List[Callable[..., Any]], *args, **kwargs) -> Dict[str, Any]:
+        re_args = self.args_process(*args)
+        res_rdd = self.runtime.sparkContext.parallelize(funcs, len(funcs)).map(
+            lambda func: func(*re_args, **kwargs))
+
+        res_list = res_rdd.collect()
+        results = {res.id: res for res in res_list}
+        return results
+
+
+class RayRuntime(RuntimeEngine):
+    """Ray runtime key is 'ray', and execute tasks in ray cluster.
+
+    Ray runtime in TaskRuntimeBackend only execute function (stateless), can be used to custom
+    resource allocation and communication etc. advanced features.
+    """
+
+    def __init__(self, engine_options):
+        super(RayRuntime, self).__init__(engine_options)
+
+    def _build_engine(self):
+        import ray
+
+        if not ray.is_initialized():
+            ray.init()
+
+        self.runtime = ray
+        self.num_executors = self.conf.get('num_executors', 1)
+        logger.info("ray init finished, executor number {}".format(str(self.num_executors)))
+
+    async def execute(self, funcs: List[Callable[..., Any]], *args, **kwargs) -> Dict[str, Any]:
+        @self.runtime.remote
+        def fn_wrapper(fn, *args):
+            if asyncio.iscoroutinefunction(fn):
+                return sync_exec(fn, *args, **kwargs)
+            else:
+                real_args = [arg for arg in args if not isinstance(arg, MethodType)]
+                return fn(*real_args, **kwargs)
+
+        params = []
+        for arg in args:
+            params.append([arg] * len(funcs))
+
+        ray_map = lambda func, fn: [func.remote(x, *y) for x, *y in zip(fn, *params)]
+        res_list = self.runtime.get(ray_map(fn_wrapper, funcs))
+        return {res.id: res for res in res_list}
+
+
+RUNTIME: Dict[str, RuntimeEngine] = {}
+
+
+def register(key, runtime_backend):
+    if RUNTIME.get(key, None) is not None:
+        logger.debug("{} runtime backend already exists, will reuse the client.".format(key))
+        return
+
+    RUNTIME[key] = runtime_backend
+    logger.info("register {}:{} success".format(key, runtime_backend))

aworld/core/singleton.py

@@ -0,0 +1,63 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+from aworld.logs.util import logger
+
+import threading
+
+
+class SingletonMeta(type):
+    _instances = {}
+    _lock = threading.Lock()
+
+    def __call__(cls, *args, **kwargs):
+        """Create or get the class instance."""
+        with cls._lock:
+            if cls not in cls._instances:
+                instance = super(SingletonMeta, cls).__call__(*args, **kwargs)
+                cls._instances[cls] = instance
+        return cls._instances[cls]
+
+
+class InheritanceSingleton(object, metaclass=SingletonMeta):
+    _local_instances = {}
+
+    @staticmethod
+    def __get_base_class(clazz):
+        if clazz == object:
+            return None
+
+        bases = clazz.__bases__
+        for base in bases:
+            if base == InheritanceSingleton:
+                return clazz
+            else:
+                base_class = InheritanceSingleton.__get_base_class(base)
+                if base_class:
+                    return base_class
+        return None
+
+    def __new__(cls, *args, **kwargs):
+        base = InheritanceSingleton.__get_base_class(cls)
+        if base is None:
+            raise ValueError(f"{cls} singleton base not found")
+
+        return super(InheritanceSingleton, cls).__new__(cls)
+
+    @classmethod
+    def instance(cls, *args, **kwargs):
+        """Each thread has its own singleton instance."""
+
+        if cls.__name__ not in cls._local_instances:
+            cls._local_instances[cls.__name__] = threading.local()
+
+        local_instance = cls._local_instances[cls.__name__]
+        if not hasattr(local_instance, 'instance'):
+            logger.info(f"{threading.current_thread().name} thread create {cls} instance.")
+            local_instance.instance = cls(*args, **kwargs)
+
+        return local_instance.instance
+
+    @classmethod
+    def clear_singleton(cls):
+        base = InheritanceSingleton.__get_base_class(cls)
+        cls._instances.pop(base, None)

aworld/core/task.py

@@ -0,0 +1,92 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import abc
+import uuid
+from dataclasses import dataclass, field
+from typing import Any, Union, List, Dict, Callable, Optional
+
+from pydantic import BaseModel
+
+from aworld.agents.llm_agent import Agent
+from aworld.core.agent.swarm import Swarm
+from aworld.core.common import Config
+from aworld.core.context.base import Context
+from aworld.core.tool.base import Tool, AsyncTool
+from aworld.output.outputs import Outputs, DefaultOutputs
+
+
+@dataclass
+class Task:
+    id: str = uuid.uuid1().hex
+    name: str = uuid.uuid1().hex
+    user_id: str = None
+    session_id: str = None
+    input: Any = None
+    # task config
+    conf: Config = None
+    # global tool instance
+    tools: List[Union[Tool, AsyncTool]] = field(default_factory=list)
+    # global tool names
+    tool_names: List[str] = field(default_factory=list)
+    # custom tool conf
+    tools_conf: Config = field(default_factory=dict)
+    # custom mcp servers conf
+    mcp_servers_conf: Config = field(default_factory=dict)
+    swarm: Optional[Swarm] = None
+    agent: Optional[Agent] = None
+    event_driven: bool = True
+    # for loop detect
+    endless_threshold: int = 3
+    # task_outputs
+    outputs: Outputs = field(default_factory=DefaultOutputs)
+    # task special runner class, for example: package.XXRunner
+    runner_cls: Optional[str] = None
+    # such as: {"start": ["init_tool", "init_context", ...]}
+    hooks: Dict[str, List[str]] = field(default_factory=dict)
+    # task specified context
+    context: 'Context' = None
+
+class TaskResponse(BaseModel):
+    id: str
+    answer: str | None
+    usage: Dict[str, Any] | None = None
+    time_cost: float | None = None
+    success: bool = False
+    msg: str | None = None
+
+
+class Runner(object):
+    __metaclass__ = abc.ABCMeta
+
+    _use_demon: bool = False
+    daemon_target: Callable[..., Any] = None
+    context: Context = None
+
+    async def pre_run(self):
+        pass
+
+    async def post_run(self):
+        pass
+
+    @abc.abstractmethod
+    async def do_run(self, context: Context = None):
+        """Raise exception if not success."""
+
+    async def _daemon_run(self):
+        if self._use_demon and self.daemon_target and callable(self.daemon_target):
+            import threading
+            t = threading.Thread(target=self.daemon_target, name="daemon", daemon=True)
+            t.start()
+
+    async def run(self) -> Any:
+        try:
+            await self.pre_run()
+            await self._daemon_run()
+            ret = await self.do_run(self.context)
+            return 0 if ret is None else ret
+        except BaseException as ex:
+            self._exception = ex
+            # do record or report
+            raise ex
+        finally:
+            await self.post_run()