Upload 10 files

aworld/models/README.md

@@ -0,0 +1,401 @@
+# AWorld LLM Interface
+
+A unified interface for interacting with various LLM providers through a consistent API.
+
+## Features
+
+- Unified API for multiple LLM providers. Currently, only OpenAI and Anthropic are supported.
+- Synchronous and asynchronous calls with optional initialization control
+- Streaming responses support
+- Tool calls support
+- Unified ModelResponse object for all provider responses
+- Easy extension with custom providers
+
+## Supported Providers
+
+- `openai`: Models supporting OpenAI API protocol (OpenAI, compatible models)
+- `anthropic`: Models supporting Anthropic API protocol (Claude models)
+- `azure_openai`: Azure OpenAI service
+
+## Basic Usage
+
+### Quick Start
+
+```python
+from aworld.config.conf import AgentConfig
+from aworld.models.llm import get_llm_model, call_llm_model, acall_llm_model
+
+# Create configuration
+config = AgentConfig(
+    llm_provider="openai",  # Options: "openai", "anthropic", "azure_openai"
+    llm_model_name="gpt-4o",
+    llm_temperature=0.0,
+    llm_api_key="your_api_key",
+    llm_base_url="your_llm_server_address"
+)
+
+# Initialize the model
+model = get_llm_model(config)
+
+# Prepare messages
+messages = [
+    {"role": "system", "content": "You are a helpful AI assistant."},
+    {"role": "user", "content": "Explain Python in three sentences."}
+]
+
+# Get response
+response = model.completion(messages)
+print(response.content)  # Access content directly from ModelResponse
+```
+
+### Using call_llm_model (Recommended)
+
+```python
+from aworld.models.llm import get_llm_model, call_llm_model
+
+# Initialize model
+model = get_llm_model(
+    llm_provider="openai",
+    model_name="gpt-4o",
+    api_key="your_api_key",
+    base_url="https://api.openai.com/v1"
+)
+
+# Prepare messages
+messages = [
+    {"role": "system", "content": "You are a helpful AI assistant."},
+    {"role": "user", "content": "Write a short poem about programming."}
+]
+
+# Using call_llm_model - returns ModelResponse object
+response = call_llm_model(model, messages)
+print(response.content)  # Access content directly from ModelResponse
+
+# Stream response with call_llm_model
+for chunk in call_llm_model(model, messages, temperature=0.7, stream=True):
+    if chunk.content:
+        print(chunk.content, end="", flush=True)
+```
+
+### Asynchronous Calls with acall_llm_model
+
+```python
+import asyncio
+from aworld.models.llm import get_llm_model, acall_llm_model
+
+async def main():
+    # Initialize model
+    model = get_llm_model(
+        llm_provider="anthropic",
+        model_name="claude-3-5-sonnet-20241022",
+        api_key="your_anthropic_api_key"
+    )
+    
+    # Prepare messages
+    messages = [
+        {"role": "user", "content": "List 3 effective ways to learn programming."}
+    ]
+    
+    # Async call with acall_llm_model
+    response = await acall_llm_model(model, messages)
+    print(response.content)
+    
+    # Async streaming with acall_llm_model
+    print("\nStreaming response:")
+    async for chunk in await acall_llm_model(model, messages, stream=True):
+        if chunk.content:
+            print(chunk.content, end="", flush=True)
+
+# Run async function
+asyncio.run(main())
+```
+
+### Selective Sync/Async Initialization
+
+For performance optimization, you can control whether to initialize synchronous or asynchronous providers:
+By default, both `sync_enabled` and `async_enabled` are set to `True`, which means both synchronous and asynchronous providers will be initialized.
+
+```python
+# Initialize only synchronous provider
+model = get_llm_model(
+    llm_provider="openai",
+    model_name="gpt-4o",
+    sync_enabled=True,    # Initialize sync provider
+    async_enabled=False,  # Don't initialize async provider
+    api_key="your_api_key"
+)
+
+# Initialize only asynchronous provider
+model = get_llm_model(
+    llm_provider="anthropic",
+    model_name="claude-3-5-sonnet-20241022",
+    sync_enabled=False,   # Don't initialize sync provider
+    async_enabled=True,   # Initialize async provider
+    api_key="your_api_key"
+)
+
+# Initialize both (default behavior)
+model = get_llm_model(
+    llm_provider="openai",
+    model_name="gpt-4o",
+    sync_enabled=True,
+    async_enabled=True
+)
+```
+
+### HTTP Client Mode
+
+You can use direct HTTP requests instead of the SDK by specifying `client_type=ClientType.HTTP` parameter:
+
+```python
+from aworld.config.conf import AgentConfig, ClientType
+from aworld.models.llm import get_llm_model, call_llm_model
+
+# Initialize model with HTTP client mode
+model = get_llm_model(
+    llm_provider="openai",
+    model_name="gpt-4o",
+    api_key="your_api_key",
+    base_url="https://api.openai.com/v1",
+    client_type=ClientType.HTTP  # Use HTTP client instead of SDK
+)
+
+# Use it exactly the same way as SDK mode
+messages = [
+    {"role": "system", "content": "You are a helpful AI assistant."},
+    {"role": "user", "content": "Tell me a short joke."}
+]
+
+# The model uses HTTP requests under the hood
+response = call_llm_model(model, messages)
+print(response.content)
+
+# Streaming also works with HTTP client
+for chunk in call_llm_model(model, messages, stream=True):
+    if chunk.content:
+        print(chunk.content, end="", flush=True)
+```
+
+This approach can be useful when:
+- You need more control over the HTTP requests
+- You have compatibility issues with the official SDK
+- You're using a model that follows OpenAI API protocol but isn't fully compatible with the SDK
+
+### Tool Calls Support
+
+```python
+from aworld.models.llm import get_llm_model, call_llm_model
+import json
+
+# Initialize model
+model = get_llm_model(
+    llm_provider="openai",
+    model_name="gpt-4o",
+    api_key="your_api_key"
+)
+
+# Define tools
+tools = [
+    {
+        "type": "function",
+        "function": {
+            "name": "get_weather",
+            "description": "Get the current weather in a given location",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "location": {
+                        "type": "string",
+                        "description": "The city and state, e.g. San Francisco, CA"
+                    }
+                },
+                "required": ["location"]
+            }
+        }
+    }
+]
+
+# Prepare messages
+messages = [
+    {"role": "user", "content": "What's the weather like in San Francisco?"}
+]
+
+# Call model with tools
+response = call_llm_model(model, messages, tools=tools, tool_choice="auto")
+
+# Check for tool calls
+if response.tool_calls:
+    for tool_call in response.tool_calls:
+        print(f"Tool name: {tool_call.name}")
+        print(f"Arguments: {tool_call.arguments}")
+        
+        # Handle tool call
+        if tool_call.name == "get_weather":
+            # Parse arguments
+            args = json.loads(tool_call.arguments)
+            location = args.get("location")
+            
+            # Mock getting weather data
+            weather = "Sunny, 25°C"
+            
+            # Add tool response to messages
+            messages.append(response.message)  # Add assistant message
+            messages.append({
+                "role": "tool",
+                "tool_call_id": tool_call.id,
+                "name": tool_call.name,
+                "content": f"{{\"weather\": \"{weather}\"}}"
+            })
+            
+            # Call model again
+            final_response = call_llm_model(model, messages)
+            print("\nFinal response:", final_response.content)
+else:
+    print("\nResponse content:", response.content)
+```
+
+### Asynchronous Calls
+
+```python
+import asyncio
+from aworld.models.llm import get_llm_model
+
+async def main():
+    # Initialize model
+    model = get_llm_model(
+        llm_provider="anthropic",
+        model_name="claude-3-5-sonnet-20241022",
+        temperature=0.0
+    )
+    
+    # Prepare messages
+    messages = [
+        {"role": "user", "content": "Explain machine learning briefly."}
+    ]
+    
+    # Async call
+    response = await model.acompletion(messages)
+    print(response.content)
+
+# Run async function
+asyncio.run(main())
+```
+
+### Streaming Responses
+
+```python
+# Synchronous streaming
+for chunk in model.stream_completion(messages):
+    print(chunk.content, end="", flush=True)
+    
+# Asynchronous streaming
+async for chunk in model.astream_completion(messages):
+    print(chunk.content, end="", flush=True)
+```
+
+## ModelResponse Object
+
+All responses are encapsulated in a unified `ModelResponse` object with these key attributes:
+
+- `id`: Response ID
+- `model`: Model name used
+- `content`: Generated text content
+- `tool_calls`: List of tool calls (if any)
+- `usage`: Token usage statistics
+- `error`: Error message (if any)
+- `message`: Complete message object for subsequent API calls
+
+Example:
+```python
+response = call_llm_model(model, messages)
+print(f"Content: {response.content}")
+print(f"Model: {response.model}")
+print(f"Total tokens: {response.usage['total_tokens']}")
+
+# Get complete message for next call
+messages.append(response.message)
+```
+
+## API Parameters
+
+Essential parameters for model calls:
+
+- `messages`: List of message dictionaries with `role` and `content` keys
+- `temperature`: Controls response randomness (0.0-1.0)
+- `max_tokens`: Maximum tokens to generate
+- `stop`: List of stopping sequences
+- `tools`: List of tool definitions
+- `tool_choice`: Tool choice strategy
+
+## Automatic Provider Detection
+
+The system can automatically identify the provider based on model name or API endpoint:
+
+```python
+# Detect Anthropic based on model name
+model = get_llm_model(model_name="claude-3-5-sonnet-20241022")
+
+```
+
+## Creating Custom Providers
+
+Implement your own provider by extending `LLMProviderBase`:
+
+```python
+from aworld.models.llm import LLMProviderBase, register_llm_provider
+from aworld.models.model_response import ModelResponse, ToolCall
+
+class CustomProvider(LLMProviderBase):
+    def _init_provider(self):
+        # Initialize your API client
+        return {
+            "api_key": self.api_key,
+            "endpoint": self.base_url
+        }
+    
+    def _init_async_provider(self):
+        # Initialize your asynchronous API client (optional)
+        # If not implemented, async methods will raise NotImplementedError
+        return None
+    
+    def preprocess_messages(self, messages):
+        # Convert standard format to your API format
+        return messages
+    
+    def postprocess_response(self, response):
+        # Convert API response to ModelResponse
+        return ModelResponse(
+            id="response_id",
+            model=self.model_name,
+            content=response.get("text", ""),
+            tool_calls=None  # Parse ToolCall objects if supported
+        )
+    
+    def completion(self, messages, temperature=0.0, **kwargs):
+        # Implement the actual API call
+        processed = self.preprocess_messages(messages)
+        # Call your API here...
+        response = {"text": "Response from custom provider"}
+        return self.postprocess_response(response)
+    
+    async def acompletion(self, messages, temperature=0.0, **kwargs):
+        # Implement async API call
+        # Similar to completion but asynchronous
+        response = {"text": "Async response from custom provider"}
+        return self.postprocess_response(response)
+
+# Register your provider
+register_llm_provider("custom_provider", CustomProvider)
+
+# Use it like any other provider
+model = get_llm_model(llm_provider="custom_provider", model_name="custom-model")
+```
+
+## API Key Management
+
+Keys are retrieved in this order:
+1. Direct `api_key` parameter
+2. Environment variable in `.env` file
+3. System environment variable
+
+Example for OpenAI: `OPENAI_API_KEY` in parameters → `.env` → system env

aworld/models/ant_provider.py

@@ -0,0 +1,866 @@
+import ast
+import asyncio
+import datetime
+import html
+import json
+import os
+import time
+
+from typing import (
+    Any,
+    List,
+    Dict,
+    Generator,
+    AsyncGenerator,
+)
+from binascii import b2a_hex
+
+from aworld.config.conf import ClientType
+from aworld.core.llm_provider_base import LLMProviderBase
+from aworld.models.llm_http_handler import LLMHTTPHandler
+from aworld.models.model_response import ModelResponse, LLMResponseError, ToolCall
+from aworld.logs.util import logger
+from aworld.utils import import_package
+from aworld.models.utils import usage_process
+
+MODEL_NAMES = {
+    "anthropic": ["claude-3-5-sonnet-20241022", "claude-3-5-sonnet-20240620", "claude-3-opus-20240229"],
+    "openai": ["gpt-4o", "gpt-4", "gpt-3.5-turbo", "o3-mini", "gpt-4o-mini"],
+}
+
+
+# Custom JSON encoder to handle ToolCall and other special types
+class CustomJSONEncoder(json.JSONEncoder):
+    """Custom JSON encoder to handle ToolCall objects and other special types."""
+
+    def default(self, obj):
+        # Handle objects with to_dict method
+        if hasattr(obj, 'to_dict') and callable(obj.to_dict):
+            return obj.to_dict()
+
+        # Handle objects with __dict__ attribute (most custom classes)
+        if hasattr(obj, '__dict__'):
+            return obj.__dict__
+
+        # Let the base class handle it (will raise TypeError if not serializable)
+        return super().default(obj)
+
+
+class AntProvider(LLMProviderBase):
+    """Ant provider implementation.
+    """
+
+    def _init_provider(self):
+        """Initialize Ant provider.
+
+        Returns:
+            Ant provider instance.
+        """
+        import_package("Crypto", install_name="pycryptodome")
+
+        # Get API key
+        api_key = self.api_key
+
+        if not api_key:
+            env_var = "ANT_API_KEY"
+            api_key = os.getenv(env_var, "")
+            self.api_key = api_key
+            if not api_key:
+                raise ValueError(
+                    f"ANT API key not found, please set {env_var} environment variable or provide it in the parameters")
+
+        if api_key and api_key.startswith("ak_info:"):
+            ak_info_str = api_key[len("ak_info:"):]
+            try:
+                ak_info = json.loads(ak_info_str)
+                for key, value in ak_info.items():
+                    os.environ[key] = value
+                    if key == "ANT_API_KEY":
+                        api_key = value
+                        self.api_key = api_key
+            except Exception as e:
+                logger.warn(f"Invalid ANT API key startswith ak_info: {api_key}")
+
+        self.stream_api_key = os.getenv("ANT_STREAM_API_KEY", "")
+
+        base_url = self.base_url
+        if not base_url:
+            base_url = os.getenv("ANT_ENDPOINT", "https://zdfmng.alipay.com")
+            self.base_url = base_url
+
+        self.aes_key = os.getenv("ANT_AES_KEY", "")
+
+        self.is_http_provider = True
+        self.kwargs["client_type"] = ClientType.HTTP
+        logger.info(f"Using HTTP provider for Ant")
+        self.http_provider = LLMHTTPHandler(
+            base_url=base_url,
+            api_key=api_key,
+            model_name=self.model_name,
+        )
+        self.is_http_provider = True
+        return self.http_provider
+
+    def _init_async_provider(self):
+        """Initialize async Ant provider.
+
+        Returns:
+            Async Ant provider instance.
+        """
+        # Get API key
+        if not self.provider:
+            provider = self._init_provider()
+            return provider
+
+    @classmethod
+    def supported_models(cls) -> list[str]:
+        return [""]
+
+    def _aes_encrypt(self, data, key):
+        """AES encryption function. If data is not a multiple of 16 [encrypted data must be a multiple of 16!], pad it to a multiple of 16.
+
+        Args:
+            key: Encryption key
+            data: Data to encrypt
+
+        Returns:
+            Encrypted data
+        """
+        from Crypto.Cipher import AES
+
+        iv = "1234567890123456"
+        cipher = AES.new(key.encode('utf-8'), AES.MODE_CBC, iv.encode('utf-8'))
+        block_size = AES.block_size
+
+        # Check if data is a multiple of 16, if not, pad with b'\0'
+        if len(data) % block_size != 0:
+            add = block_size - (len(data) % block_size)
+        else:
+            add = 0
+        data = data.encode('utf-8') + b'\0' * add
+        encrypted = cipher.encrypt(data)
+        result = b2a_hex(encrypted)
+        return result.decode('utf-8')
+
+    def _build_openai_params(self,
+                             messages: List[Dict[str, str]],
+                             temperature: float = 0.0,
+                             max_tokens: int = None,
+                             stop: List[str] = None,
+                             **kwargs) -> Dict[str, Any]:
+        openai_params = {
+            "model": kwargs.get("model_name", self.model_name or ""),
+            "messages": messages,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
+            "stop": stop
+        }
+
+        supported_params = [
+            "frequency_penalty", "logit_bias", "logprobs", "top_logprobs",
+            "presence_penalty", "response_format", "seed", "stream", "top_p",
+            "user", "function_call", "functions", "tools", "tool_choice"
+        ]
+
+        for param in supported_params:
+            if param in kwargs:
+                openai_params[param] = kwargs[param]
+
+        return openai_params
+
+    def _build_claude_params(self,
+                             messages: List[Dict[str, str]],
+                             temperature: float = 0.0,
+                             max_tokens: int = None,
+                             stop: List[str] = None,
+                             **kwargs) -> Dict[str, Any]:
+        claude_params = {
+            "model": kwargs.get("model_name", self.model_name or ""),
+            "messages": messages,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
+            "stop": stop
+        }
+
+        supported_params = [
+            "top_p", "top_k", "reasoning_effort", "tools", "tool_choice"
+        ]
+
+        for param in supported_params:
+            if param in kwargs:
+                claude_params[param] = kwargs[param]
+
+        return claude_params
+
+    def _get_visit_info(self):
+        visit_info = {
+            "visitDomain": self.kwargs.get("ant_visit_domain") or os.getenv("ANT_VISIT_DOMAIN", "BU_general"),
+            "visitBiz": self.kwargs.get("ant_visit_biz") or os.getenv("ANT_VISIT_BIZ", ""),
+            "visitBizLine": self.kwargs.get("ant_visit_biz_line") or os.getenv("ANT_VISIT_BIZ_LINE", "")
+        }
+        if not visit_info["visitBiz"] or not visit_info["visitBizLine"]:
+            return None
+        return visit_info
+
+    def _get_service_param(self,
+                           message_key: str,
+                           output_type: str = "request",
+                           messages: List[Dict[str, str]] = None,
+                           temperature: float = 0.0,
+                           max_tokens: int = None,
+                           stop: List[str] = None,
+                           **kwargs
+                           ) -> Dict[str, Any]:
+        """Get service name from model name.
+        Returns:
+            Service name.
+        """
+        if messages:
+            for message in messages:
+                if message["role"] == "assistant" and "tool_calls" in message and message["tool_calls"]:
+                    if message["content"] is None: message["content"] = ""
+                    processed_tool_calls = []
+                    for tool_call in message["tool_calls"]:
+                        if isinstance(tool_call, dict):
+                            processed_tool_calls.append(tool_call)
+                        elif isinstance(tool_call, ToolCall):
+                            processed_tool_calls.append(tool_call.to_dict())
+                    message["tool_calls"] = processed_tool_calls
+        query_conditions = {
+            "messageKey": message_key,
+        }
+        param = {"cacheInterval": -1, }
+        visit_info = self._get_visit_info()
+        if not visit_info:
+            raise LLMResponseError(
+                f"AntProvider#Invalid visit_info, please set ANT_VISIT_BIZ and ANT_VISIT_BIZ_LINE environment variable or provide it in the parameters",
+                self.model_name or "unknown"
+            )
+        param.update(visit_info)
+        if self.model_name.startswith("claude"):
+            query_conditions.update(self._build_claude_params(messages, temperature, max_tokens, stop, **kwargs))
+            param.update({
+                "serviceName": "amazon_claude_chat_completions_dataview",
+                "queryConditions": query_conditions,
+            })
+        elif output_type == "pull":
+            param.update({
+                "serviceName": "chatgpt_response_query_dataview",
+                "queryConditions": query_conditions
+            })
+        else:
+            query_conditions = {
+                "model": self.model_name,
+                "n": "1",
+                "api_key": self.api_key,
+                "messageKey": message_key,
+                "outputType": "PULL",
+                "messages": messages,
+            }
+            query_conditions.update(self._build_openai_params(messages, temperature, max_tokens, stop, **kwargs))
+            param.update({
+                "serviceName": "asyn_chatgpt_prompts_completions_query_dataview",
+                "queryConditions": query_conditions,
+            })
+        return param
+
+    def _gen_message_key(self):
+        def _timestamp():
+            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")
+            return timestamp
+
+        timestamp = _timestamp()
+        message_key = "llm_call_%s" % (timestamp)
+        return message_key
+
+    def _build_request_data(self, param: Dict[str, Any]):
+        param_data = json.dumps(param)
+        encrypted_param_data = self._aes_encrypt(param_data, self.aes_key)
+        post_data = {"encryptedParam": encrypted_param_data}
+        return post_data
+
+    def _build_chat_query_request_data(self,
+                                       message_key: str,
+                                       messages: List[Dict[str, str]],
+                                       temperature: float = 0.0,
+                                       max_tokens: int = None,
+                                       stop: List[str] = None,
+                                       **kwargs):
+        param = self._get_service_param(message_key, "request", messages, temperature, max_tokens, stop, **kwargs)
+        query_data = self._build_request_data(param)
+        return query_data
+
+    def _post_chat_query_request(self,
+                                 messages: List[Dict[str, str]],
+                                 temperature: float = 0.0,
+                                 max_tokens: int = None,
+                                 stop: List[str] = None,
+                                 **kwargs):
+        message_key = self._gen_message_key()
+        post_data = self._build_chat_query_request_data(message_key,
+                                                        messages,
+                                                        model_name=self.model_name,
+                                                        temperature=temperature,
+                                                        max_tokens=max_tokens,
+                                                        stop=stop,
+                                                        **kwargs)
+        response = self.http_provider.sync_call(post_data, endpoint="commonQuery/queryData")
+        return message_key, response
+
+    def _valid_chat_result(self, body):
+        if "data" not in body or not body["data"]:
+            return False
+        if "values" not in body["data"] or not body["data"]["values"]:
+            return False
+        if "response" not in body["data"]["values"] and "data" not in body["data"]["values"]:
+            return False
+        return True
+
+    def _build_chat_pull_request_data(self, message_key):
+        param = self._get_service_param(message_key, "pull")
+
+        pull_data = self._build_request_data(param)
+        return pull_data
+
+    def _pull_chat_result(self, message_key, response: Dict[str, Any], timeout):
+        if self.model_name.startswith("claude"):
+            if self._valid_chat_result(response):
+                x = response["data"]["values"]["data"]
+                ast_str = ast.literal_eval("'" + x + "'")
+                result = html.unescape(ast_str)
+                data = json.loads(result)
+                return data
+            else:
+                raise LLMResponseError(
+                    f"Invalid response from Ant API, response: {response}",
+                    self.model_name or "unknown"
+                )
+
+        post_data = self._build_chat_pull_request_data(message_key)
+        url = 'commonQuery/queryData'
+        headers = {
+            'Content-Type': 'application/json'
+        }
+
+        # Start polling until valid result or timeout
+        start_time = time.time()
+        elapsed_time = 0
+
+        while elapsed_time < timeout:
+            response = self.http_provider.sync_call(post_data, endpoint=url, headers=headers)
+
+            logger.debug(f"Poll attempt at {elapsed_time}s, response: {response}")
+
+            # Check if valid result is received
+            if self._valid_chat_result(response):
+                x = response["data"]["values"]["response"]
+                ast_str = ast.literal_eval("'" + x + "'")
+                result = html.unescape(ast_str)
+                data = json.loads(result)
+                return data
+            elif (not response.get("success")) or ("data" in response and response["data"]):
+                err_code = response.get("data", {}).get("errorCode", "")
+                err_msg = response.get("data", {}).get("errorMessage", "")
+                if err_code or err_msg:
+                    raise LLMResponseError(
+                        f"Request failed: {response}",
+                        self.model_name or "unknown"
+                    )
+
+            # If no result, wait 1 second and query again
+            time.sleep(1)
+            elapsed_time = time.time() - start_time
+            logger.debug(f"Polling... Elapsed time: {elapsed_time:.1f}s")
+
+        # Timeout handling
+        raise LLMResponseError(
+            f"Timeout after {timeout} seconds waiting for response from Ant API",
+            self.model_name or "unknown"
+        )
+
+    async def _async_pull_chat_result(self, message_key, response: Dict[str, Any], timeout):
+        if self.model_name.startswith("claude"):
+            if self._valid_chat_result(response):
+                x = response["data"]["values"]["data"]
+                ast_str = ast.literal_eval("'" + x + "'")
+                result = html.unescape(ast_str)
+                data = json.loads(result)
+                return data
+            elif (not response.get("success")) or ("data" in response and response["data"]):
+                err_code = response.get("data", {}).get("errorCode", "")
+                err_msg = response.get("data", {}).get("errorMessage", "")
+                if err_code or err_msg:
+                    raise LLMResponseError(
+                        f"Request failed: {response}",
+                        self.model_name or "unknown"
+                    )
+
+        post_data = self._build_chat_pull_request_data(message_key)
+        url = 'commonQuery/queryData'
+        headers = {
+            'Content-Type': 'application/json'
+        }
+
+        # Start polling until valid result or timeout
+        start_time = time.time()
+        elapsed_time = 0
+
+        while elapsed_time < timeout:
+            response = await self.http_provider.async_call(post_data, endpoint=url, headers=headers)
+
+            logger.debug(f"Poll attempt at {elapsed_time}s, response: {response}")
+
+            # Check if valid result is received
+            if self._valid_chat_result(response):
+                x = response["data"]["values"]["response"]
+                ast_str = ast.literal_eval("'" + x + "'")
+                result = html.unescape(ast_str)
+                data = json.loads(result)
+                return data
+            elif (not response.get("success")) or ("data" in response and response["data"]):
+                err_code = response.get("data", {}).get("errorCode", "")
+                err_msg = response.get("data", {}).get("errorMessage", "")
+                if err_code or err_msg:
+                    raise LLMResponseError(
+                        f"Request failed: {response}",
+                        self.model_name or "unknown"
+                    )
+
+            # If no result, wait 1 second and query again
+            await asyncio.sleep(1)
+            elapsed_time = time.time() - start_time
+            logger.debug(f"Polling... Elapsed time: {elapsed_time:.1f}s")
+
+        # Timeout handling
+        raise LLMResponseError(
+            f"Timeout after {timeout} seconds waiting for response from Ant API",
+            self.model_name or "unknown"
+        )
+
+    def _convert_completion_message(self, message: Dict[str, Any], is_finished: bool = False) -> ModelResponse:
+        """Convert Ant completion message to OpenAI format.
+
+        Args:
+            message: Ant completion message.
+
+        Returns:
+            OpenAI format message.
+        """
+        # Generate unique ID
+        response_id = f"ant-{hash(str(message)) & 0xffffffff:08x}"
+
+        # Get content
+        content = message.get("completion", "")
+
+        # Create message object
+        message_dict = {
+            "role": "assistant",
+            "content": content,
+            "is_chunk": True
+        }
+
+        # Keep original contextId and sessionId
+        if "contextId" in message:
+            message_dict["contextId"] = message["contextId"]
+        if "sessionId" in message:
+            message_dict["sessionId"] = message["sessionId"]
+
+        usage = {
+            "completion_tokens": message.get("completionToken", 0),
+            "prompt_tokens": message.get("promptTokens", 0),
+            "total_tokens": message.get("completionToken", 0) + message.get("promptTokens", 0)
+        }
+
+        # process tool calls
+        tool_calls = message.get("toolCalls", [])
+        for tool_call in tool_calls:
+            index = tool_call.get("index", 0)
+            name = tool_call.get("function", {}).get("name")
+            arguments = tool_call.get("function", {}).get("arguments")
+            if index >= len(self.stream_tool_buffer):
+                self.stream_tool_buffer.append({
+                    "id": tool_call.get("id"),
+                    "type": "function",
+                    "function": {
+                        "name": name,
+                        "arguments": arguments
+                    }
+                })
+            else:
+                self.stream_tool_buffer[index]["function"]["arguments"] += arguments
+
+        if is_finished and self.stream_tool_buffer:
+            message_dict["tool_calls"] = self.stream_tool_buffer.copy()
+            processed_tool_calls = []
+            for tool_call in self.stream_tool_buffer:
+                processed_tool_calls.append(ToolCall.from_dict(tool_call))
+            tool_resp = ModelResponse(
+                id=response_id,
+                model=self.model_name or "ant",
+                content=content,
+                tool_calls=processed_tool_calls,
+                usage=usage,
+                raw_response=message,
+                message=message_dict
+            )
+            self.stream_tool_buffer = []
+            return tool_resp
+
+        # Build and return ModelResponse object directly
+        return ModelResponse(
+            id=response_id,
+            model=self.model_name or "ant",
+            content=content,
+            tool_calls=None,  # TODO: add tool calls
+            usage=usage,
+            raw_response=message,
+            message=message_dict
+        )
+
+    def preprocess_stream_call_message(self, messages: List[Dict[str, str]], ext_params: Dict[str, Any]) -> Dict[
+        str, str]:
+        """Preprocess messages, use Ant format directly.
+
+        Args:
+            messages: Ant format message list.
+
+        Returns:
+            Processed message list.
+        """
+        param = {
+            "messages": messages,
+            "sessionId": "TkQUldjzOgYSKyTrpor3TA==",
+            "model": self.model_name,
+            "needMemory": False,
+            "stream": True,
+            "contextId": "contextId_34555fd2d246447fa55a1a259445a427",
+            "platform": "AWorld"
+        }
+        for k in ext_params.keys():
+            if k not in param:
+                param[k] = ext_params[k]
+        return param
+
+    def postprocess_response(self, response: Any) -> ModelResponse:
+        """Process Ant response.
+
+        Args:
+            response: Ant response object.
+
+        Returns:
+            ModelResponse object.
+
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        if ((not isinstance(response, dict) and (not hasattr(response, 'choices') or not response.choices))
+                or (isinstance(response, dict) and not response.get("choices"))):
+            error_msg = ""
+            if hasattr(response, 'error') and response.error and isinstance(response.error, dict):
+                error_msg = response.error.get('message', '')
+            elif hasattr(response, 'msg'):
+                error_msg = response.msg
+
+            raise LLMResponseError(
+                error_msg if error_msg else "Unknown error",
+                self.model_name or "unknown",
+                response
+            )
+
+        return ModelResponse.from_openai_response(response)
+
+    def postprocess_stream_response(self, chunk: Any) -> ModelResponse:
+        """Process Ant stream response chunk.
+
+        Args:
+            chunk: Ant response chunk.
+
+        Returns:
+            ModelResponse object.
+
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        # Check if chunk contains error
+        if hasattr(chunk, 'error') or (isinstance(chunk, dict) and chunk.get('error')):
+            error_msg = chunk.error if hasattr(chunk, 'error') else chunk.get('error', 'Unknown error')
+            raise LLMResponseError(
+                error_msg,
+                self.model_name or "unknown",
+                chunk
+            )
+
+        if isinstance(chunk, dict) and ('completion' in chunk):
+            return self._convert_completion_message(chunk)
+
+        # If chunk is already in OpenAI format, use standard processing method
+        return ModelResponse.from_openai_stream_chunk(chunk)
+
+    def completion(self,
+                   messages: List[Dict[str, str]],
+                   temperature: float = 0.0,
+                   max_tokens: int = None,
+                   stop: List[str] = None,
+                   **kwargs) -> ModelResponse:
+        """Synchronously call Ant to generate response.
+
+        Args:
+            messages: Message list.
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            ModelResponse object.
+
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        if not self.provider:
+            raise RuntimeError(
+                "Sync provider not initialized. Make sure 'sync_enabled' parameter is set to True in initialization.")
+
+        try:
+            start_time = time.time()
+            message_key, response = self._post_chat_query_request(messages, temperature, max_tokens, stop, **kwargs)
+            timeout = kwargs.get("response_timeout", self.kwargs.get("timeout", 180))
+            result = self._pull_chat_result(message_key, response, timeout)
+            logger.info(f"completion cost time: {time.time() - start_time}s.")
+
+            resp = self.postprocess_response(result)
+            usage_process(resp.usage)
+            return resp
+        except Exception as e:
+            if isinstance(e, LLMResponseError):
+                raise e
+            logger.warn(f"Error in Ant completion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "unknown"))
+
+    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 Ant to generate response.
+
+        Args:
+            messages: Message list.
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            ModelResponse object.
+
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        if not self.async_provider:
+            self._init_async_provider()
+
+        start_time = time.time()
+        try:
+            message_key, response = self._post_chat_query_request(messages, temperature, max_tokens, stop, **kwargs)
+            timeout = kwargs.get("response_timeout", self.kwargs.get("timeout", 180))
+            result = await self._async_pull_chat_result(message_key, response, timeout)
+            logger.info(f"completion cost time: {time.time() - start_time}s.")
+
+            resp = self.postprocess_response(result)
+            usage_process(resp.usage)
+            return resp
+
+        except Exception as e:
+            if isinstance(e, LLMResponseError):
+                raise e
+            logger.warn(f"Error in async Ant completion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "unknown"))
+
+    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 Ant to generate streaming response.
+
+        Args:
+            messages: Message list.
+            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.
+        """
+        if not self.provider:
+            raise RuntimeError(
+                "Sync provider not initialized. Make sure 'sync_enabled' parameter is set to True in initialization.")
+
+        start_time = time.time()
+        # Generate message_key
+        timestamp = int(time.time())
+        self.message_key = f"llm_call_{timestamp}"
+        message_key_literal = self.message_key  # Ensure it's a direct string literal
+        self.aes_key = kwargs.get("aes_key", self.aes_key)
+
+        # Add streaming parameter
+        kwargs["stream"] = True
+        processed_messages = self.preprocess_stream_call_message(messages,
+                                                                 self._build_openai_params(temperature, max_tokens,
+                                                                                           stop, **kwargs))
+        if not processed_messages:
+            raise LLMResponseError("Failed to get post data", self.model_name or "unknown")
+
+        usage = {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0
+        }
+
+        try:
+            # Send request
+            # response = self.http_provider.sync_call(processed_messages[0], endpoint="commonQuery/queryData")
+            headers = {
+                "Content-Type": "application/json",
+                "X_ACCESS_KEY": self.stream_api_key
+            }
+            response_stream = self.http_provider.sync_stream_call(processed_messages, endpoint="chat/completions",
+                                                                  headers=headers)
+            if response_stream:
+                for chunk in response_stream:
+                    if not chunk:
+                        continue
+
+                    # Process special markers
+                    if isinstance(chunk, dict) and "status" in chunk:
+                        if chunk["status"] == "done":
+                            # Stream completion marker, can choose to end
+                            logger.info("Received [DONE] marker, stream completed")
+                            yield self._convert_completion_message(chunk, is_finished=True)
+                            yield ModelResponse.from_special_marker("done", self.model_name, chunk)
+                            break
+                        elif chunk["status"] == "revoke":
+                            # Revoke marker, need to notify the frontend to revoke the displayed content
+                            logger.info("Received [REVOKE] marker, content should be revoked")
+                            yield ModelResponse.from_special_marker("revoke", self.model_name, chunk)
+                            continue
+                        elif chunk["status"] == "fail":
+                            # Fail marker
+                            logger.error("Received [FAIL] marker, request failed")
+                            raise LLMResponseError("Request failed", self.model_name or "unknown")
+                        elif chunk["status"] == "cancel":
+                            # Request was cancelled
+                            logger.warning("Received [CANCEL] marker, stream was cancelled")
+                            raise LLMResponseError("Stream was cancelled", self.model_name or "unknown")
+                        continue
+
+                    # Process normal response chunks
+                    resp = self.postprocess_stream_response(chunk)
+                    self._accumulate_chunk_usage(usage, resp.usage)
+                    yield resp
+                usage_process(usage)
+
+            logger.info(f"stream_completion cost time: {time.time() - start_time}s.")
+        except Exception as e:
+            if isinstance(e, LLMResponseError):
+                raise e
+            logger.error(f"Error in Ant stream completion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "unknown"))
+
+    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 Ant to generate streaming response.
+
+        Args:
+            messages: Message list.
+            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.
+        """
+        if not self.async_provider:
+            self._init_async_provider()
+
+        start_time = time.time()
+        # Generate message_key
+        timestamp = int(time.time())
+        self.message_key = f"llm_call_{timestamp}"
+        message_key_literal = self.message_key  # Ensure it's a direct string literal
+        self.aes_key = kwargs.get("aes_key", self.aes_key)
+
+        # Add streaming parameter
+        kwargs["stream"] = True
+        processed_messages = self.preprocess_stream_call_message(messages,
+                                                                 self._build_openai_params(temperature, max_tokens,
+                                                                                           stop, **kwargs))
+        if not processed_messages:
+            raise LLMResponseError("Failed to get post data", self.model_name or "unknown")
+
+        usage = {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0
+        }
+        try:
+            headers = {
+                "Content-Type": "application/json",
+                "X_ACCESS_KEY": self.stream_api_key
+            }
+            logger.info(f"astream_completion request data: {processed_messages}")
+
+            async for chunk in self.http_provider.async_stream_call(processed_messages, endpoint="chat/completions",
+                                                                    headers=headers):
+                if not chunk:
+                    continue
+
+                # Process special markers
+                if isinstance(chunk, dict) and "status" in chunk:
+                    if chunk["status"] == "done":
+                        # Stream completion marker, can choose to end
+                        logger.info("Received [DONE] marker, stream completed")
+                        yield ModelResponse.from_special_marker("done", self.model_name, chunk)
+                        break
+                    elif chunk["status"] == "revoke":
+                        # Revoke marker, need to notify the frontend to revoke the displayed content
+                        logger.info("Received [REVOKE] marker, content should be revoked")
+                        yield ModelResponse.from_special_marker("revoke", self.model_name, chunk)
+                        continue
+                    elif chunk["status"] == "fail":
+                        # Fail marker
+                        logger.error("Received [FAIL] marker, request failed")
+                        raise LLMResponseError("Request failed", self.model_name or "unknown")
+                    elif chunk["status"] == "cancel":
+                        # Request was cancelled
+                        logger.warning("Received [CANCEL] marker, stream was cancelled")
+                        raise LLMResponseError("Stream was cancelled", self.model_name or "unknown")
+                    continue
+
+                # Process normal response chunks
+                resp = self.postprocess_stream_response(chunk)
+                self._accumulate_chunk_usage(usage, resp.usage)
+                yield resp
+            usage_process(usage)
+
+            logger.info(f"astream_completion cost time: {time.time() - start_time}s.")
+        except Exception as e:
+            if isinstance(e, LLMResponseError):
+                raise e
+            logger.warn(f"Error in async Ant stream completion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "unknown"))

aworld/models/anthropic_provider.py

@@ -0,0 +1,333 @@
+import os
+from typing import Any, Dict, List, Generator, AsyncGenerator
+
+from aworld.utils import import_package
+from aworld.logs.util import logger
+from aworld.core.llm_provider_base import LLMProviderBase
+from aworld.models.model_response import ModelResponse, LLMResponseError
+
+
+class AnthropicProvider(LLMProviderBase):
+    """Anthropic provider implementation.
+    """
+
+    def __init__(self,
+                 api_key: str = None,
+                 base_url: str = None,
+                 model_name: str = None,
+                 sync_enabled: bool = None,
+                 async_enabled: bool = None,
+                 **kwargs):
+        super().__init__(api_key, base_url, model_name, sync_enabled, async_enabled, **kwargs)
+        import_package("anthropic")
+
+    def _init_provider(self):
+        """Initialize Anthropic provider.
+
+        Returns:
+            Anthropic provider instance.
+        """
+        from anthropic import Anthropic
+
+        # Get API key
+        api_key = self.api_key
+        if not api_key:
+            env_var = "ANTHROPIC_API_KEY"
+            api_key = os.getenv(env_var, "")
+            if not api_key:
+                raise ValueError(
+                    f"Anthropic API key not found, please set {env_var} environment variable or provide it in the parameters")
+
+        return Anthropic(
+            api_key=api_key,
+            base_url=self.base_url
+        )
+
+    def _init_async_provider(self):
+        """Initialize async Anthropic provider.
+
+        Returns:
+            Async Anthropic provider instance.
+        """
+        from anthropic import Anthropic, AsyncAnthropic
+
+        # Get API key
+        api_key = self.api_key
+        if not api_key:
+            env_var = "ANTHROPIC_API_KEY"
+            api_key = os.getenv(env_var, "")
+            if not api_key:
+                raise ValueError(
+                    f"Anthropic API key not found, please set {env_var} environment variable or provide it in the parameters")
+
+        return AsyncAnthropic(
+            api_key=api_key,
+            base_url=self.base_url
+        )
+
+    @classmethod
+    def supported_models(cls) -> list[str]:
+        return [r"claude-3-.*"]
+
+    def preprocess_messages(self, messages: List[Dict[str, str]]) -> Dict[str, Any]:
+        """Preprocess messages, convert OpenAI format to Anthropic format.
+
+        Args:
+            messages: OpenAI format message list.
+
+        Returns:
+            Converted message dictionary, containing messages and system fields.
+        """
+        anthropic_messages = []
+        system_content = None
+
+        for msg in messages:
+            role = msg.get("role", "")
+            content = msg.get("content", "")
+
+            if role == "system":
+                system_content = content
+            elif role == "user":
+                anthropic_messages.append({"role": "user", "content": content})
+            elif role == "assistant":
+                anthropic_messages.append({"role": "assistant", "content": content})
+
+        return {
+            "messages": anthropic_messages,
+            "system": system_content
+        }
+
+    def postprocess_response(self, response: Any) -> ModelResponse:
+        """Process Anthropic response to unified ModelResponse.
+
+        Args:
+            response: Anthropic response object.
+
+        Returns:
+            ModelResponse object.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        # Check if response is empty or contains error
+        if not response or (isinstance(response, dict) and response.get('error')):
+            error_msg = response.get('error', 'Unknown error') if isinstance(response, dict) else 'Empty response'
+            raise LLMResponseError(error_msg, self.model_name or "claude", response)
+
+        return ModelResponse.from_anthropic_response(response)
+
+    def postprocess_stream_response(self, chunk: Any) -> ModelResponse:
+        """Process Anthropic streaming response chunk.
+
+        Args:
+            chunk: Anthropic response chunk.
+
+        Returns:
+            ModelResponse object.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        # Check if chunk is empty or contains error
+        if not chunk or (isinstance(chunk, dict) and chunk.get('error')):
+            error_msg = chunk.get('error', 'Unknown error') if isinstance(chunk, dict) else 'Empty response'
+            raise LLMResponseError(error_msg, self.model_name or "claude", chunk)
+
+        return ModelResponse.from_anthropic_stream_chunk(chunk)
+
+    def completion(self,
+                   messages: List[Dict[str, str]],
+                   temperature: float = 0.0,
+                   max_tokens: int = None,
+                   stop: List[str] = None,
+                   **kwargs) -> ModelResponse:
+        """Synchronously call Anthropic to generate response.
+        
+        Args:
+            messages: Message list.
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            ModelResponse object.
+        """
+        if not self.provider:
+            raise RuntimeError(
+                "Sync provider not initialized. Make sure 'sync_enabled' parameter is set to True in initialization.")
+
+        try:
+            processed_data = self.preprocess_messages(messages)
+            processed_messages = processed_data["messages"]
+            system_content = processed_data["system"]
+            anthropic_params = self.get_anthropic_params(processed_messages, system_content, temperature, max_tokens,
+                                                         stop, **kwargs)
+            response = self.provider.visited_messages.create(**anthropic_params)
+
+            return self.postprocess_response(response)
+        except Exception as e:
+            logger.warn(f"Error in Anthropic completion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "claude"))
+
+    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 Anthropic to generate streaming response.
+
+        Args:
+            messages: Message list.
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            Generator yielding ModelResponse chunks.
+        """
+        if not self.provider:
+            raise RuntimeError(
+                "Sync provider not initialized. Make sure 'sync_enabled' parameter is set to True in initialization.")
+
+        try:
+            processed_data = self.preprocess_messages(messages)
+            processed_messages = processed_data["messages"]
+            system_content = processed_data["system"]
+            anthropic_params = self.get_anthropic_params(processed_messages, system_content, temperature, max_tokens,
+                                                         stop, **kwargs)
+            anthropic_params["stream"] = True
+            response_stream = self.provider.visited_messages.create(**anthropic_params)
+
+            for chunk in response_stream:
+                if not chunk:
+                    continue
+
+                yield self.postprocess_stream_response(chunk)
+
+        except Exception as e:
+            logger.warn(f"Error in Anthropic stream_completion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "claude"))
+
+    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 Anthropic to generate streaming response.
+        
+        Args:
+            messages: Message list.
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            AsyncGenerator yielding ModelResponse chunks.
+        """
+        if not self.async_provider:
+            raise RuntimeError(
+                "Async provider not initialized. Make sure 'async_enabled' parameter is set to True in initialization.")
+
+        try:
+            processed_data = self.preprocess_messages(messages)
+            processed_messages = processed_data["messages"]
+            system_content = processed_data["system"]
+            anthropic_params = self.get_anthropic_params(processed_messages, system_content, temperature, max_tokens,
+                                                         stop, **kwargs)
+            anthropic_params["stream"] = True
+            response_stream = await self.async_provider.visited_messages.create(**anthropic_params)
+
+            async for chunk in response_stream:
+                if not chunk:
+                    continue
+
+                yield self.postprocess_stream_response(chunk)
+
+        except Exception as e:
+            logger.warn(f"Error in Anthropic astream_completion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "claude"))
+
+    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 Anthropic to generate response.
+        
+        Args:
+            messages: Message list.
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            ModelResponse object.
+        """
+        if not self.async_provider:
+            raise RuntimeError(
+                "Async provider not initialized. Make sure 'async_enabled' parameter is set to True in initialization.")
+
+        try:
+            processed_data = self.preprocess_messages(messages)
+            processed_messages = processed_data["messages"]
+            system_content = processed_data["system"]
+            anthropic_params = self.get_anthropic_params(processed_messages, system_content, temperature, max_tokens,
+                                                         stop, **kwargs)
+            response = await self.async_provider.visited_messages.create(**anthropic_params)
+
+            return self.postprocess_response(response)
+        except Exception as e:
+            logger.warn(f"Error in Anthropic acompletion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "claude"))
+
+    def get_anthropic_params(self,
+                             messages: List[Dict[str, str]],
+                             system: str = None,
+                             temperature: float = 0.0,
+                             max_tokens: int = None,
+                             stop: List[str] = None,
+                             **kwargs) -> Dict[str, Any]:
+        if "tools" in kwargs:
+            openai_tools = kwargs["tools"]
+            claude_tools = []
+
+            for tool in openai_tools:
+                if tool["type"] == "function":
+                    claude_tool = {
+                        "name": tool["name"],
+                        "description": tool["description"],
+                        "input_schema": {
+                            "type": "object",
+                            "properties": tool["parameters"]["properties"],
+                            "required": tool["parameters"].get("required", [])
+                        }
+                    }
+                    claude_tools.append(claude_tool)
+
+            kwargs["tools"] = claude_tools
+
+        anthropic_params = {
+            "model": kwargs.get("model_name", self.model_name or ""),
+            "messages": messages,
+            "system": system,
+            "temperature": temperature,
+            "max_tokens": max_tokens or 4096,
+            "stop_sequences": stop,
+        }
+
+        if "tools" in kwargs and kwargs["tools"]:
+            anthropic_params["tools"] = kwargs["tools"]
+            anthropic_params["tool_choice"] = kwargs.get("tool_choice", "auto")
+
+        for param in ["top_p", "top_k", "metadata", "stream"]:
+            if param in kwargs:
+                anthropic_params[param] = kwargs[param]
+
+        return anthropic_params

aworld/models/llm.py

@@ -0,0 +1,584 @@
+from typing import (
+    List,
+    Dict,
+    Union,
+    Generator,
+    AsyncGenerator,
+)
+from aworld.config import ConfigDict
+from aworld.config.conf import AgentConfig, ClientType
+from aworld.logs.util import logger
+
+from aworld.core.llm_provider_base import LLMProviderBase
+from aworld.models.openai_provider import OpenAIProvider, AzureOpenAIProvider
+from aworld.models.anthropic_provider import AnthropicProvider
+from aworld.models.ant_provider import AntProvider
+from aworld.models.model_response import ModelResponse
+
+# Predefined model names for common providers
+MODEL_NAMES = {
+    "anthropic": ["claude-3-5-sonnet-20241022", "claude-3-5-sonnet-20240620", "claude-3-opus-20240229"],
+    "openai": ["gpt-4o", "gpt-4", "gpt-3.5-turbo", "o3-mini", "gpt-4o-mini"],
+    "azure_openai": ["gpt-4", "gpt-4-turbo", "gpt-4o", "gpt-35-turbo"],
+}
+
+# Endpoint patterns for identifying providers
+ENDPOINT_PATTERNS = {
+    "openai": ["api.openai.com"],
+    "anthropic": ["api.anthropic.com", "claude-api"],
+    "azure_openai": ["openai.azure.com"],
+    "ant": ["zdfmng.alipay.com"],
+}
+
+# Provider class mapping
+PROVIDER_CLASSES = {
+    "openai": OpenAIProvider,
+    "anthropic": AnthropicProvider,
+    "azure_openai": AzureOpenAIProvider,
+    "ant": AntProvider,
+}
+
+
+class LLMModel:
+    """Unified large model interface, encapsulates different model implementations, provides a unified completion method.
+    """
+
+    def __init__(self, conf: Union[ConfigDict, AgentConfig] = None, custom_provider: LLMProviderBase = None, **kwargs):
+        """Initialize unified model interface.
+
+        Args:
+            conf: Agent configuration, if provided, create model based on configuration.
+            custom_provider: Custom LLMProviderBase instance, if provided, use it directly.
+            **kwargs: Other parameters, may include:
+                - base_url: Specify model endpoint.
+                - api_key: API key.
+                - model_name: Model name.
+                - temperature: Temperature parameter.
+        """
+
+        # If custom_provider instance is provided, use it directly
+        if custom_provider is not None:
+            if not isinstance(custom_provider, LLMProviderBase):
+                raise TypeError(
+                    "custom_provider must be an instance of LLMProviderBase")
+            self.provider_name = "custom"
+            self.provider = custom_provider
+            return
+
+        # Get basic parameters
+        base_url = kwargs.get("base_url") or (
+            conf.llm_base_url if conf else None)
+        model_name = kwargs.get("model_name") or (
+            conf.llm_model_name if conf else None)
+        llm_provider = conf.llm_provider if conf_contains_key(
+            conf, "llm_provider") else None
+
+        # Get API key from configuration (if any)
+        if conf and conf.llm_api_key:
+            kwargs["api_key"] = conf.llm_api_key
+
+        # Identify provider
+        self.provider_name = self._identify_provider(
+            llm_provider, base_url, model_name)
+
+        # Fill basic parameters
+        kwargs['base_url'] = base_url
+        kwargs['model_name'] = model_name
+
+        # Fill parameters for llm provider
+        kwargs['sync_enabled'] = conf.llm_sync_enabled if conf_contains_key(
+            conf, "llm_sync_enabled") else True
+        kwargs['async_enabled'] = conf.llm_async_enabled if conf_contains_key(
+            conf, "llm_async_enabled") else True
+        kwargs['client_type'] = conf.llm_client_type if conf_contains_key(
+            conf, "llm_client_type") else ClientType.SDK
+
+        kwargs.update(self._transfer_conf_to_args(conf))
+
+        # Create model provider based on provider_name
+        self._create_provider(**kwargs)
+
+    def _transfer_conf_to_args(self, conf: Union[ConfigDict, AgentConfig] = None) -> dict:
+        """
+        Transfer parameters from conf to args
+
+        Args:
+            conf: config object
+        """
+        if not conf:
+            return {}
+
+        # Get all parameters from conf
+        if type(conf).__name__ == 'AgentConfig':
+            conf_dict = conf.model_dump()
+        else:  # ConfigDict
+            conf_dict = conf
+
+        ignored_keys = ["llm_provider", "llm_base_url", "llm_model_name", "llm_api_key", "llm_sync_enabled",
+                        "llm_async_enabled", "llm_client_type"]
+        args = {}
+        # Filter out used parameters and add remaining parameters to args
+        for key, value in conf_dict.items():
+            if key not in ignored_keys and value is not None:
+                args[key] = value
+
+        return args
+
+    def _identify_provider(self, provider: str = None, base_url: str = None, model_name: str = None) -> str:
+        """Identify LLM provider.
+
+        Identification logic:
+        1. If provider is specified and doesn't need to be overridden, use the specified provider.
+        2. If base_url is provided, try to identify provider based on base_url.
+        3. If model_name is provided, try to identify provider based on model_name.
+        4. If none can be identified, default to "openai".
+
+        Args:
+            provider: Specified provider.
+            base_url: Service URL.
+            model_name: Model name.
+
+        Returns:
+            str: Identified provider.
+        """
+        # Default provider
+        identified_provider = "openai"
+
+        # Identify provider based on base_url
+        if base_url:
+            for p, patterns in ENDPOINT_PATTERNS.items():
+                if any(pattern in base_url for pattern in patterns):
+                    identified_provider = p
+                    logger.info(
+                        f"Identified provider: {identified_provider} based on base_url: {base_url}")
+                    return identified_provider
+
+        # Identify provider based on model_name
+        if model_name and not base_url:
+            for p, models in MODEL_NAMES.items():
+                if model_name in models or any(model_name.startswith(model) for model in models):
+                    identified_provider = p
+                    logger.info(
+                        f"Identified provider: {identified_provider} based on model_name: {model_name}")
+                    break
+
+        if provider and provider in PROVIDER_CLASSES and identified_provider and identified_provider != provider:
+            logger.warning(
+                f"Provider mismatch: {provider} != {identified_provider}, using {provider} as provider")
+            identified_provider = provider
+
+        return identified_provider
+
+    def _create_provider(self, **kwargs):
+        """Return the corresponding provider instance based on provider.
+
+        Args:
+            **kwargs: Parameters, may include:
+                - base_url: Model endpoint.
+                - api_key: API key.
+                - model_name: Model name.
+                - temperature: Temperature parameter.
+                - timeout: Timeout.
+                - max_retries: Maximum number of retries.
+        """
+        self.provider = PROVIDER_CLASSES[self.provider_name](**kwargs)
+
+    @classmethod
+    def supported_providers(cls) -> list[str]:
+        return list(PROVIDER_CLASSES.keys())
+
+    def supported_models(self) -> list[str]:
+        """Get supported models for the current provider.
+        Returns:
+            list: Supported models.
+        """
+        return self.provider.supported_models() if self.provider else []
+
+    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.
+        """
+        # Call provider's acompletion method directly
+        return await self.provider.acompletion(
+            messages=messages,
+            temperature=temperature,
+            max_tokens=max_tokens,
+            stop=stop,
+            **kwargs
+        )
+
+    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.
+        """
+        # Call provider's completion method directly
+        return self.provider.completion(
+            messages=messages,
+            temperature=temperature,
+            max_tokens=max_tokens,
+            stop=stop,
+            **kwargs
+        )
+
+    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.
+        """
+        # Call provider's stream_completion method directly
+        return self.provider.stream_completion(
+            messages=messages,
+            temperature=temperature,
+            max_tokens=max_tokens,
+            stop=stop,
+            **kwargs
+        )
+
+    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, may include:
+                - base_url: Specify model endpoint.
+                - api_key: API key.
+                - model_name: Model name.
+
+        Returns:
+            AsyncGenerator yielding ModelResponse chunks.
+        """
+        # Call provider's astream_completion method directly
+        async for chunk in self.provider.astream_completion(
+                messages=messages,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                stop=stop,
+                **kwargs
+        ):
+            yield chunk
+
+    def speech_to_text(self,
+                       audio_file: str,
+                       language: str = None,
+                       prompt: str = None,
+                       **kwargs) -> ModelResponse:
+        """Convert speech to text.
+
+        Args:
+            audio_file: Path to audio file or file object.
+            language: Audio language, optional.
+            prompt: Transcription prompt, optional.
+            **kwargs: Other parameters.
+
+        Returns:
+            ModelResponse: Unified model response object, with content field containing the transcription result.
+
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+            NotImplementedError: When provider does not support speech to text conversion.
+        """
+        return self.provider.speech_to_text(
+            audio_file=audio_file,
+            language=language,
+            prompt=prompt,
+            **kwargs
+        )
+
+    async def aspeech_to_text(self,
+                              audio_file: str,
+                              language: str = None,
+                              prompt: str = None,
+                              **kwargs) -> ModelResponse:
+        """Asynchronously convert speech to text.
+
+        Args:
+            audio_file: Path to audio file or file object.
+            language: Audio language, optional.
+            prompt: Transcription prompt, optional.
+            **kwargs: Other parameters.
+
+        Returns:
+            ModelResponse: Unified model response object, with content field containing the transcription result.
+
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+            NotImplementedError: When provider does not support speech to text conversion.
+        """
+        return await self.provider.aspeech_to_text(
+            audio_file=audio_file,
+            language=language,
+            prompt=prompt,
+            **kwargs
+        )
+
+
+def register_llm_provider(provider: str, provider_class: type):
+    """Register a custom LLM provider.
+
+    Args:
+        provider: Provider name.
+        provider_class: Provider class, must inherit from LLMProviderBase.
+    """
+    if not issubclass(provider_class, LLMProviderBase):
+        raise TypeError("provider_class must be a subclass of LLMProviderBase")
+    PROVIDER_CLASSES[provider] = provider_class
+
+
+def conf_contains_key(conf: Union[ConfigDict, AgentConfig], key: str) -> bool:
+    """Check if conf contains key.
+    Args:
+        conf: Config object.
+        key: Key to check.
+    Returns:
+        bool: Whether conf contains key.
+    """
+    if not conf:
+        return False
+    if type(conf).__name__ == 'AgentConfig':
+        return hasattr(conf, key)
+    else:
+        return key in conf
+
+
+def get_llm_model(conf: Union[ConfigDict, AgentConfig] = None,
+                  custom_provider: LLMProviderBase = None,
+                  **kwargs) -> Union[LLMModel, 'ChatOpenAI']:
+    """Get a unified LLM model instance.
+
+    Args:
+        conf: Agent configuration, if provided, create model based on configuration.
+        custom_provider: Custom LLMProviderBase instance, if provided, use it directly.
+        **kwargs: Other parameters, may include:
+            - base_url: Specify model endpoint.
+            - api_key: API key.
+            - model_name: Model name.
+            - temperature: Temperature parameter.
+
+    Returns:
+        Unified model interface.
+    """
+    # Create and return LLMModel instance directly
+    llm_provider = conf.llm_provider if conf_contains_key(
+        conf, "llm_provider") else None
+
+    if (llm_provider == "chatopenai"):
+        from langchain_openai import ChatOpenAI
+
+        base_url = kwargs.get("base_url") or (
+            conf.llm_base_url if conf_contains_key(conf, "llm_base_url") else None)
+        model_name = kwargs.get("model_name") or (
+            conf.llm_model_name if conf_contains_key(conf, "llm_model_name") else None)
+        api_key = kwargs.get("api_key") or (
+            conf.llm_api_key if conf_contains_key(conf, "llm_api_key") else None)
+
+        return ChatOpenAI(
+            model=model_name,
+            temperature=kwargs.get("temperature", conf.llm_temperature if conf_contains_key(
+                conf, "llm_temperature") else 0.0),
+            base_url=base_url,
+            api_key=api_key,
+        )
+
+    return LLMModel(conf=conf, custom_provider=custom_provider, **kwargs)
+
+
+def call_llm_model(
+        llm_model: LLMModel,
+        messages: List[Dict[str, str]],
+        temperature: float = 0.0,
+        max_tokens: int = None,
+        stop: List[str] = None,
+        stream: bool = False,
+        **kwargs
+) -> Union[ModelResponse, Generator[ModelResponse, None, None]]:
+    """Convenience function to call LLM model.
+
+    Args:
+        llm_model: LLM model instance.
+        messages: Message list.
+        temperature: Temperature parameter.
+        max_tokens: Maximum number of tokens to generate.
+        stop: List of stop sequences.
+        stream: Whether to return a streaming response.
+        **kwargs: Other parameters.
+
+    Returns:
+        Model response or response generator.
+    """
+    if stream:
+        return llm_model.stream_completion(
+            messages=messages,
+            temperature=temperature,
+            max_tokens=max_tokens,
+            stop=stop,
+            **kwargs
+        )
+    else:
+        return llm_model.completion(
+            messages=messages,
+            temperature=temperature,
+            max_tokens=max_tokens,
+            stop=stop,
+            **kwargs
+        )
+
+
+async def acall_llm_model(
+        llm_model: LLMModel,
+        messages: List[Dict[str, str]],
+        temperature: float = 0.0,
+        max_tokens: int = None,
+        stop: List[str] = None,
+        stream: bool = False,
+        **kwargs
+) -> ModelResponse:
+    """Convenience function to asynchronously call LLM model.
+
+    Args:
+        llm_model: LLM model instance.
+        messages: Message list.
+        temperature: Temperature parameter.
+        max_tokens: Maximum number of tokens to generate.
+        stop: List of stop sequences.
+        stream: Whether to return a streaming response.
+        **kwargs: Other parameters.
+
+    Returns:
+        Model response or response generator.
+    """
+    return await llm_model.acompletion(
+        messages=messages,
+        temperature=temperature,
+        max_tokens=max_tokens,
+        stop=stop,
+        **kwargs
+    )
+
+
+async def acall_llm_model_stream(
+        llm_model: LLMModel,
+        messages: List[Dict[str, str]],
+        temperature: float = 0.0,
+        max_tokens: int = None,
+        stop: List[str] = None,
+        **kwargs
+) -> AsyncGenerator[ModelResponse, None]:
+    async for chunk in llm_model.astream_completion(
+            messages=messages,
+            temperature=temperature,
+            max_tokens=max_tokens,
+            stop=stop,
+            **kwargs
+    ):
+        yield chunk
+
+
+def speech_to_text(
+        llm_model: LLMModel,
+        audio_file: str,
+        language: str = None,
+        prompt: str = None,
+        **kwargs
+) -> ModelResponse:
+    """Convenience function to convert speech to text.
+
+    Args:
+        llm_model: LLM model instance.
+        audio_file: Path to audio file or file object.
+        language: Audio language, optional.
+        prompt: Transcription prompt, optional.
+        **kwargs: Other parameters.
+
+    Returns:
+        ModelResponse: Unified model response object, with content field containing the transcription result.
+    """
+    if llm_model.provider_name != "openai":
+        raise NotImplementedError(
+            f"Speech-to-text functionality is currently only supported for OpenAI compatible provider, current provider: {llm_model.provider_name}")
+
+    return llm_model.speech_to_text(
+        audio_file=audio_file,
+        language=language,
+        prompt=prompt,
+        **kwargs
+    )
+
+
+async def aspeech_to_text(
+        llm_model: LLMModel,
+        audio_file: str,
+        language: str = None,
+        prompt: str = None,
+        **kwargs
+) -> ModelResponse:
+    """Convenience function to asynchronously convert speech to text.
+
+    Args:
+        llm_model: LLM model instance.
+        audio_file: Path to audio file or file object.
+        language: Audio language, optional.
+        prompt: Transcription prompt, optional.
+        **kwargs: Other parameters.
+
+    Returns:
+        ModelResponse: Unified model response object, with content field containing the transcription result.
+    """
+    if llm_model.provider_name != "openai":
+        raise NotImplementedError(
+            f"Speech-to-text functionality is currently only supported for OpenAI compatible provider, current provider: {llm_model.provider_name}")
+
+    return await llm_model.aspeech_to_text(
+        audio_file=audio_file,
+        language=language,
+        prompt=prompt,
+        **kwargs
+    )

aworld/models/llm_http_handler.py

@@ -0,0 +1,397 @@
+"""HTTP handler for LLM providers.
+
+This module provides a generic HTTP handler for making requests to LLM providers
+when direct SDK usage is not desired.
+"""
+
+import json
+import asyncio
+import random
+import time
+from typing import Any, Dict, List, Optional, Union, Generator, AsyncGenerator
+import requests
+from requests import HTTPError
+
+from aworld.logs.util import logger
+from aworld.utils import import_package
+
+class LLMHTTPHandler:
+    """HTTP handler for LLM providers.
+
+    This class provides methods to make HTTP requests to LLM providers
+    instead of using their SDKs directly.
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        api_key: str,
+        model_name: str,
+        headers: Optional[Dict[str, str]] = None,
+        timeout: int = 180,
+        max_retries: int = 3,
+    ) -> None:
+        """Initialize the HTTP handler.
+
+        Args:
+            base_url: Base URL for the LLM API.
+            api_key: API key for authentication.
+            model_name: Name of the model to use.
+            headers: Additional headers to include in requests.
+            timeout: Request timeout in seconds.
+            max_retries: Maximum number of retries for failed requests.
+        """
+        import_package("aiohttp")
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self.model_name = model_name
+        self.timeout = timeout
+        self.max_retries = max_retries
+
+        # Set up default headers
+        self.headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {api_key}",
+        }
+        if headers:
+            self.headers.update(headers)
+
+    def _parse_sse_line(self, line: bytes) -> Optional[Dict[str, Any]]:
+        """Parse a Server-Sent Events (SSE) line.
+
+        Args:
+            line: Raw SSE line.
+
+        Returns:
+            Parsed JSON data if successful, None otherwise.
+        """
+        try:
+            # Remove 'data: ' prefix if present
+            line_str = line.decode('utf-8').strip()
+            if line_str.startswith('data: '):
+                line_str = line_str[6:]
+
+            # Skip empty lines
+            if not line_str:
+                return None
+
+            return json.loads(line_str)
+        except (json.JSONDecodeError, UnicodeDecodeError) as e:
+            logger.warning(f"Failed to parse SSE line: {line}, error: {str(e)}")
+            return None
+
+    def _make_request(
+        self,
+        endpoint: str,
+        data: Dict[str, Any],
+        stream: bool = False,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Dict[str, Any], Generator[Dict[str, Any], None, None]]:
+        """Make a synchronous HTTP request.
+
+        Args:
+            endpoint: API endpoint to call.
+            data: Request data to send.
+            stream: Whether to stream the response.
+
+        Returns:
+            Response data or generator of response chunks.
+
+        Raises:
+            requests.exceptions.RequestException: If the request fails.
+        """
+        url = f"{self.base_url}/{endpoint.lstrip('/')}"
+        request_headers = self.headers.copy()
+        if headers:
+            request_headers.update(headers)
+
+
+        try:
+            if stream:
+                    response = requests.post(
+                        url,
+                        headers=request_headers,
+                        json=data,
+                        stream=True,
+                        timeout=self.timeout,
+                    )
+                    response.raise_for_status()
+
+                    def generate_chunks():
+                        for line in response.iter_lines():
+                            if line:
+                                line_str = line.decode('utf-8').strip()
+                                if line_str.startswith('data: '):
+                                    line_content = line_str[6:]
+
+                                    if line_content == "[DONE]":
+                                        yield {"status": "done", "message": "Stream completed"}
+                                        break
+                                    elif line_content == "[REVOKE]":
+                                        yield {"status": "revoke", "message": "Content should be revoked"}
+                                        continue
+                                    elif line_content == "[FAIL]":
+                                        yield {"status": "fail", "message": "Request failed"}
+                                        break
+                                    elif line_content.startswith("[FAIL]_stream was reset: CANCEL"):
+                                        yield {"status": "cancel", "message": "Stream was cancelled"}
+                                        break
+
+                                chunk = self._parse_sse_line(line)
+                                if chunk is not None:
+                                    yield chunk
+                    return generate_chunks()
+            else:
+                response = requests.post(
+                    url,
+                    headers=request_headers,
+                    json=data,
+                    timeout=self.timeout,
+                )
+                response.raise_for_status()
+                return response.json()
+        except Exception as e:
+            logger.error(f"Error in HttpHandler: {str(e)}")
+            raise
+
+    async def _make_async_request_stream(
+        self,
+        endpoint: str,
+        data: Dict[str, Any],
+        headers: Optional[Dict[str, str]] = None,
+    ) -> AsyncGenerator[Dict[str, Any], None]:
+        """Make an asynchronous streaming HTTP request.
+
+        Args:
+            endpoint: API endpoint to call.
+            data: Request data to send.
+
+        Yields:
+            Response chunks.
+
+        Raises:
+            aiohttp.ClientError: If the request fails.
+        """
+        import aiohttp
+        url = f"{self.base_url}/{endpoint.lstrip('/')}"
+        request_headers = self.headers.copy()
+        if headers:
+            request_headers.update(headers)
+
+        # Create an independent session and keep it open
+        session = aiohttp.ClientSession()
+        try:
+            response = await session.post(
+                url,
+                headers=request_headers,
+                json=data,
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+
+            # Implement async generator directly
+            async for line in response.content:
+                if line:
+                    line_str = line.decode('utf-8').strip()
+                    if line_str.startswith('data: '):
+                        line_content = line_str[6:]
+
+                        if line_content == "[DONE]":
+                            yield {"status": "done", "message": "Stream completed"}
+                            break
+                        elif line_content == "[REVOKE]":
+                            yield {"status": "revoke", "message": "Content should be revoked"}
+                            continue
+                        elif line_content == "[FAIL]":
+                            yield {"status": "fail", "message": "Request failed"}
+                            break
+                        elif line_content.startswith("[FAIL]_stream was reset: CANCEL"):
+                            yield {"status": "cancel", "message": "Stream was cancelled"}
+                            break
+
+                    chunk = self._parse_sse_line(line)
+                    if chunk is not None:
+                        yield chunk
+        except Exception as e:
+            logger.error(f"Error in stream: {str(e)}")
+            raise
+        finally:
+            # Ensure the session is eventually closed
+            await session.close()
+
+    async def _make_async_request(
+        self,
+        endpoint: str,
+        data: Dict[str, Any],
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Dict[str, Any]:
+        """Make an asynchronous non-streaming HTTP request.
+
+        Args:
+            endpoint: API endpoint to call.
+            data: Request data to send.
+
+        Returns:
+            Response data.
+
+        Raises:
+            aiohttp.ClientError: If the request fails.
+        """
+        import aiohttp
+        url = f"{self.base_url}/{endpoint.lstrip('/')}"
+        request_headers = self.headers.copy()
+        if headers:
+            request_headers.update(headers)
+
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                url,
+                headers=request_headers,
+                json=data,
+                timeout=self.timeout,
+            ) as response:
+                response.raise_for_status()
+                return await response.json()
+
+    def sync_call(
+        self,
+        data: Dict[str, Any],
+        endpoint: str = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Dict[str, Any]:
+        """Make a synchronous completion request.
+
+        Args:
+            data: Request data.
+
+        Returns:
+            Response data.
+        """
+        logger.debug(f"sync_call request data: {data}")
+
+        if not endpoint:
+            endpoint = "chat/completions"
+
+        retries = 0
+        while retries < self.max_retries:
+            try:
+                response = self._make_request(endpoint, data, headers=headers)
+                return response
+            except Exception as e:
+                last_error = e
+                retries += 1
+                if retries < self.max_retries:
+                    logger.warning(f"Request failed, retrying ({retries}/{self.max_retries}): {str(e)}")
+                    # Exponential backoff with jitter
+                    backoff = min(2 ** retries + random.uniform(0, 1), 10)
+                    time.sleep(backoff)
+                else:
+                    logger.error(f"Request failed after {self.max_retries} retries: {str(e)}")
+                    raise last_error
+
+    async def async_call(
+        self,
+        data: Dict[str, Any],
+        endpoint: str = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Dict[str, Any]:
+        """Make an asynchronous completion request.
+
+        Args:
+            data: Request data.
+
+        Returns:
+            Response data.
+        """
+        import aiohttp
+        logger.info(f"async_call request data: {data}")
+
+        retries = 0
+        last_error = None
+        if not endpoint:
+            endpoint = "chat/completions"
+
+        while retries < self.max_retries:
+            try:
+                response = await self._make_async_request(endpoint, data, headers=headers)
+                return response
+            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+                last_error = e
+                retries += 1
+                if retries < self.max_retries:
+                    logger.warning(f"Request failed, retrying ({retries}/{self.max_retries}): {str(e)}")
+                    # Exponential backoff with jitter
+                    backoff = min(2 ** retries + random.uniform(0, 1), 10)
+                    await asyncio.sleep(backoff)
+                else:
+                    logger.error(f"Request failed after {self.max_retries} retries: {str(e)}")
+                    raise last_error
+
+    def sync_stream_call(
+        self,
+        data: Dict[str, Any],
+        endpoint: str = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Generator[Dict[str, Any], None, None]:
+        """Make a synchronous streaming completion request.
+
+        Args:
+            data: Request data.
+
+        Yields:
+            Response chunks.
+        """
+        data["stream"] = True
+        logger.info(f"sync_stream_call request data: {data}")
+        retries = 0
+
+        while retries < self.max_retries:
+            try:
+                for chunk in self._make_request(endpoint or "chat/completions", data, stream=True, headers=headers):
+                    yield chunk
+                return  # Exit after completing stream processing
+            except Exception as e:
+                last_error = e
+                retries += 1
+                if retries < self.max_retries:
+                    logger.warning(f"Stream connection failed, retrying ({retries}/{self.max_retries}): {str(e)}")
+                else:
+                    logger.error(f"Stream connection failed after {self.max_retries} retries: {str(e)}")
+                    raise last_error
+
+
+    async def async_stream_call(
+        self,
+        data: Dict[str, Any],
+        endpoint: str = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> AsyncGenerator[Dict[str, Any], None]:
+        """Make an asynchronous streaming completion request.
+
+        Args:
+            data: Request data.
+
+        Yields:
+            Response chunks.
+        """
+        import aiohttp
+        data["stream"] = True
+        logger.info(f"async_stream_call request data: {data}")
+
+        retries = 0
+        last_error = None
+
+        while retries < self.max_retries:
+            try:
+                async for chunk in self._make_async_request_stream(endpoint or "chat/completions", data, headers=headers):
+                    yield chunk
+                return  # Exit after completing stream processing
+            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+                last_error = e
+                retries += 1
+                if retries < self.max_retries:
+                    logger.warning(f"Stream connection failed, retrying ({retries}/{self.max_retries}): {str(e)}")
+                    await asyncio.sleep(1)  # Wait one second before retrying
+                else:
+                    logger.error(f"Stream connection failed after {self.max_retries} retries: {str(e)}")
+                    raise last_error

aworld/models/model_response.py

@@ -0,0 +1,631 @@
+from typing import Any, Dict, List, Optional
+import json
+from pydantic import BaseModel
+
+
+class LLMResponseError(Exception):
+    """Represents an error in LLM response.
+    
+    Attributes:
+        message: Error message
+        model: Model name
+        response: Original response object
+    """
+
+    def __init__(self, message: str, model: str = "unknown", response: Any = None):
+        """
+        Initialize LLM response error
+        
+        Args:
+            message: Error message
+            model: Model name
+            response: Original response object
+        """
+        self.message = message
+        self.model = model
+        self.response = response
+        super().__init__(f"LLM Error ({model}): {message}. Response: {response}")
+
+
+class Function(BaseModel):
+    """
+    Represents a function call made by a model
+    """
+    name: str
+    arguments: str = None
+
+
+class ToolCall(BaseModel):
+    """
+    Represents a tool call made by a model
+    """
+
+    id: str
+    type: str = "function"
+    function: Function = None
+
+    # name: str = None
+    # arguments: str = None
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'ToolCall':
+        """
+        Create ToolCall from dictionary representation
+
+        Args:
+            data: Dictionary containing tool call data
+
+        Returns:
+            ToolCall object
+        """
+        if not data:
+            return None
+
+        tool_id = data.get('id', f"call_{hash(str(data)) & 0xffffffff:08x}")
+        tool_type = data.get('type', 'function')
+
+        function_data = data.get('function', {})
+        name = function_data.get('name')
+
+        arguments = function_data.get('arguments')
+        # Ensure arguments is a string
+        if arguments is not None and not isinstance(arguments, str):
+            arguments = json.dumps(arguments, ensure_ascii=False)
+
+        function = Function(name=name, arguments=arguments)
+
+        return cls(
+            id=tool_id,
+            type=tool_type,
+            function=function,
+            # name=name,
+            # arguments=arguments,
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Convert ToolCall to dictionary representation
+
+        Returns:
+            Dictionary representation
+        """
+        return {
+            "id": self.id,
+            "type": self.type,
+            "function": {
+                "name": self.function.name,
+                "arguments": self.function.arguments
+            }
+        }
+
+    def __repr__(self):
+        return json.dumps(self.to_dict(), ensure_ascii=False)
+
+    def __iter__(self):
+        """
+        Make ToolCall dict-like for JSON serialization
+        """
+        yield from self.to_dict().items()
+
+
+class ModelResponse:
+    """
+    Unified model response class for encapsulating responses from different LLM providers
+    """
+
+    def __init__(
+            self,
+            id: str,
+            model: str,
+            content: str = None,
+            tool_calls: List[ToolCall] = None,
+            usage: Dict[str, int] = None,
+            error: str = None,
+            raw_response: Any = None,
+            message: Dict[str, Any] = None
+    ):
+        """
+        Initialize ModelResponse object
+
+        Args:
+            id: Response ID
+            model: Model name used
+            content: Generated text content
+            tool_calls: List of tool calls
+            usage: Usage statistics (token counts, etc.)
+            error: Error message (if any)
+            raw_response: Original response object
+            message: Complete message object, can be used for subsequent API calls
+        """
+        self.id = id
+        self.model = model
+        self.content = content
+        self.tool_calls = tool_calls
+        self.usage = usage or {
+            "completion_tokens": 0,
+            "prompt_tokens": 0,
+            "total_tokens": 0
+        }
+        self.error = error
+        self.raw_response = raw_response
+
+        # If message is not provided, construct one from other fields
+        if message is None:
+            self.message = {
+                "role": "assistant",
+                "content": content
+            }
+
+            if tool_calls:
+                self.message["tool_calls"] = [tool_call.to_dict() for tool_call in tool_calls]
+        else:
+            self.message = message
+
+    @classmethod
+    def from_openai_response(cls, response: Any) -> 'ModelResponse':
+        """
+        Create ModelResponse from OpenAI response object
+
+        Args:
+            response: OpenAI response object
+
+        Returns:
+            ModelResponse object
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs
+        """
+        # Handle error cases
+        if hasattr(response, 'error') or (isinstance(response, dict) and response.get('error')):
+            error_msg = response.error if hasattr(response, 'error') else response.get('error', 'Unknown error')
+            raise LLMResponseError(
+                error_msg,
+                response.model if hasattr(response, 'model') else response.get('model', 'unknown'),
+                response
+            )
+
+        # Normal case
+        message = None
+        if hasattr(response, 'choices') and response.choices:
+            message = response.choices[0].message
+        elif isinstance(response, dict) and response.get('choices'):
+            message = response['choices'][0].get('message', {})
+
+        if not message:
+            raise LLMResponseError(
+                "No message found in response",
+                response.model if hasattr(response, 'model') else response.get('model', 'unknown'),
+                response
+            )
+
+        # Extract usage information
+        usage = {}
+        if hasattr(response, 'usage'):
+            usage = {
+                "completion_tokens": response.usage.completion_tokens if hasattr(response.usage,
+                                                                                 'completion_tokens') else 0,
+                "prompt_tokens": response.usage.prompt_tokens if hasattr(response.usage, 'prompt_tokens') else 0,
+                "total_tokens": response.usage.total_tokens if hasattr(response.usage, 'total_tokens') else 0
+            }
+        elif isinstance(response, dict) and response.get('usage'):
+            usage = response['usage']
+
+        # Build message object
+        message_dict = {}
+        if hasattr(message, '__dict__'):
+            # Convert object to dictionary
+            for key, value in message.__dict__.items():
+                if not key.startswith('_'):
+                    message_dict[key] = value
+        elif isinstance(message, dict):
+            message_dict = message
+        else:
+            # Extract common properties
+            message_dict = {
+                "role": "assistant",
+                "content": message.content if hasattr(message, 'content') else "",
+                "tool_calls": message.tool_calls if hasattr(message, 'tool_calls') else None,
+            }
+
+        message_dict["content"] = '' if message_dict.get('content') is None else message_dict.get('content', '')
+
+        # Process tool calls
+        processed_tool_calls = []
+        raw_tool_calls = message.tool_calls if hasattr(message, 'tool_calls') else message_dict.get('tool_calls')
+        if raw_tool_calls:
+            for tool_call in raw_tool_calls:
+                if isinstance(tool_call, dict):
+                    processed_tool_calls.append(ToolCall.from_dict(tool_call))
+                else:
+                    # Handle OpenAI object
+                    tool_call_dict = {
+                        "id": tool_call.id if hasattr(tool_call,
+                                                      'id') else f"call_{hash(str(tool_call)) & 0xffffffff:08x}",
+                        "type": tool_call.type if hasattr(tool_call, 'type') else "function"
+                    }
+
+                    if hasattr(tool_call, 'function'):
+                        function = tool_call.function
+                        tool_call_dict["function"] = {
+                            "name": function.name if hasattr(function, 'name') else None,
+                            "arguments": function.arguments if hasattr(function, 'arguments') else None
+                        }
+                    processed_tool_calls.append(ToolCall.from_dict(tool_call_dict))
+
+        if message_dict and processed_tool_calls:
+            message_dict["tool_calls"] = [tool_call.to_dict() for tool_call in processed_tool_calls]
+
+        # Create and return ModelResponse
+        return cls(
+            id=response.id if hasattr(response, 'id') else response.get('id', 'unknown'),
+            model=response.model if hasattr(response, 'model') else response.get('model', 'unknown'),
+            content=message.content if hasattr(message, 'content') else message.get('content') or "",
+            tool_calls=processed_tool_calls or None,
+            usage=usage,
+            raw_response=response,
+            message=message_dict
+        )
+
+    @classmethod
+    def from_openai_stream_chunk(cls, chunk: Any) -> 'ModelResponse':
+        """
+        Create ModelResponse from OpenAI stream response chunk
+
+        Args:
+            chunk: OpenAI stream chunk
+
+        Returns:
+            ModelResponse object
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs
+        """
+        # Handle error cases
+        if hasattr(chunk, 'error') or (isinstance(chunk, dict) and chunk.get('error')):
+            error_msg = chunk.error if hasattr(chunk, 'error') else chunk.get('error', 'Unknown error')
+            raise LLMResponseError(
+                error_msg,
+                chunk.model if hasattr(chunk, 'model') else chunk.get('model', 'unknown'),
+                chunk
+            )
+
+        # Handle finish reason chunk (end of stream)
+        if hasattr(chunk, 'choices') and chunk.choices and chunk.choices[0].finish_reason:
+            return cls(
+                id=chunk.id if hasattr(chunk, 'id') else chunk.get('id', 'unknown'),
+                model=chunk.model if hasattr(chunk, 'model') else chunk.get('model', 'unknown'),
+                content=None,
+                raw_response=chunk,
+                message={"role": "assistant", "content": "", "finish_reason": chunk.choices[0].finish_reason}
+            )
+
+        # Normal chunk with delta content
+        content = None
+        processed_tool_calls = []
+
+        if hasattr(chunk, 'choices') and chunk.choices:
+            delta = chunk.choices[0].delta
+            if hasattr(delta, 'content') and delta.content:
+                content = delta.content
+            if hasattr(delta, 'tool_calls') and delta.tool_calls:
+                raw_tool_calls = delta.tool_calls
+                for tool_call in raw_tool_calls:
+                    if isinstance(tool_call, dict):
+                        processed_tool_calls.append(ToolCall.from_dict(tool_call))
+                    else:
+                        # Handle OpenAI object
+                        tool_call_dict = {
+                            "id": tool_call.id if hasattr(tool_call,
+                                                          'id') else f"call_{hash(str(tool_call)) & 0xffffffff:08x}",
+                            "type": tool_call.type if hasattr(tool_call, 'type') else "function"
+                        }
+
+                        if hasattr(tool_call, 'function'):
+                            function = tool_call.function
+                            tool_call_dict["function"] = {
+                                "name": function.name if hasattr(function, 'name') else None,
+                                "arguments": function.arguments if hasattr(function, 'arguments') else None
+                            }
+
+                        processed_tool_calls.append(ToolCall.from_dict(tool_call_dict))
+        elif isinstance(chunk, dict) and chunk.get('choices'):
+            delta = chunk['choices'][0].get('delta', {})
+            if not delta:
+                delta = chunk['choices'][0].get('message', {})
+            content = delta.get('content')
+            raw_tool_calls = delta.get('tool_calls')
+            if raw_tool_calls:
+                for tool_call in raw_tool_calls:
+                    processed_tool_calls.append(ToolCall.from_dict(tool_call))
+
+        # Extract usage information
+        usage = {}
+        if hasattr(chunk, 'usage'):
+            usage = {
+                "completion_tokens": chunk.usage.completion_tokens if hasattr(chunk.usage, 'completion_tokens') else 0,
+                "prompt_tokens": chunk.usage.prompt_tokens if hasattr(chunk.usage, 'prompt_tokens') else 0,
+                "total_tokens": chunk.usage.total_tokens if hasattr(chunk.usage, 'total_tokens') else 0
+            }
+        elif isinstance(chunk, dict) and chunk.get('usage'):
+            usage = chunk['usage']
+
+        # Create message object
+        message = {
+            "role": "assistant",
+            "content": content or "",
+            "tool_calls": [tool_call.to_dict() for tool_call in processed_tool_calls] if processed_tool_calls else None,
+            "is_chunk": True
+        }
+
+        # Create and return ModelResponse
+        return cls(
+            id=chunk.id if hasattr(chunk, 'id') else chunk.get('id', 'unknown'),
+            model=chunk.model if hasattr(chunk, 'model') else chunk.get('model', 'unknown'),
+            content=content,
+            tool_calls=processed_tool_calls or None,
+            usage=usage,
+            raw_response=chunk,
+            message=message
+        )
+
+    @classmethod
+    def from_anthropic_stream_chunk(cls, chunk: Any) -> 'ModelResponse':
+        """
+        Create ModelResponse from Anthropic stream response chunk
+
+        Args:
+            chunk: Anthropic stream chunk
+
+        Returns:
+            ModelResponse object
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs
+        """
+        try:
+            # Handle error cases
+            if not chunk or (isinstance(chunk, dict) and chunk.get('error')):
+                error_msg = chunk.get('error', 'Unknown error') if isinstance(chunk, dict) else 'Empty response'
+                raise LLMResponseError(
+                    error_msg,
+                    chunk.model if hasattr(chunk, 'model') else chunk.get('model', 'unknown'),
+                    chunk)
+
+            # Handle stop reason (end of stream)
+            if hasattr(chunk, 'stop_reason') and chunk.stop_reason:
+                return cls(
+                    id=chunk.id if hasattr(chunk, 'id') else 'unknown',
+                    model=chunk.model if hasattr(chunk, 'model') else 'claude',
+                    content=None,
+                    raw_response=chunk,
+                    message={"role": "assistant", "content": "", "stop_reason": chunk.stop_reason}
+                )
+
+            # Handle delta content
+            content = None
+            processed_tool_calls = []
+
+            if hasattr(chunk, 'delta') and chunk.delta:
+                delta = chunk.delta
+                if hasattr(delta, 'text') and delta.text:
+                    content = delta.text
+                elif hasattr(delta, 'tool_use') and delta.tool_use:
+                    tool_call_dict = {
+                        "id": f"call_{delta.tool_use.id}",
+                        "type": "function",
+                        "function": {
+                            "name": delta.tool_use.name,
+                            "arguments": delta.tool_use.input if isinstance(delta.tool_use.input, str) else json.dumps(
+                                delta.tool_use.input, ensure_ascii=False)
+                        }
+                    }
+                    processed_tool_calls.append(ToolCall.from_dict(tool_call_dict))
+
+            # Create message object
+            message = {
+                "role": "assistant",
+                "content": content or "",
+                "tool_calls": [tool_call.to_dict() for tool_call in
+                               processed_tool_calls] if processed_tool_calls else None,
+                "is_chunk": True
+            }
+
+            # Create and return ModelResponse
+            return cls(
+                id=chunk.id if hasattr(chunk, 'id') else 'unknown',
+                model=chunk.model if hasattr(chunk, 'model') else 'claude',
+                content=content,
+                tool_calls=processed_tool_calls or None,
+                raw_response=chunk,
+                message=message
+            )
+
+        except Exception as e:
+            if isinstance(e, LLMResponseError):
+                raise e
+            raise LLMResponseError(
+                f"Error processing Anthropic stream chunk: {str(e)}",
+                chunk.model if hasattr(chunk, 'model') else chunk.get('model', 'unknown'),
+                chunk)
+
+    @classmethod
+    def from_anthropic_response(cls, response: Any) -> 'ModelResponse':
+        """
+        Create ModelResponse from Anthropic original response object
+
+        Args:
+            response: Anthropic response object
+
+        Returns:
+            ModelResponse object
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs
+        """
+        try:
+            # Handle error cases
+            if not response or (isinstance(response, dict) and response.get('error')):
+                error_msg = response.get('error', 'Unknown error') if isinstance(response, dict) else 'Empty response'
+                raise LLMResponseError(
+                    error_msg,
+                    response.model if hasattr(response, 'model') else response.get('model', 'unknown'),
+                    response)
+
+            # Build message content
+            message = {
+                "content": "",
+                "role": "assistant",
+                "tool_calls": None,
+            }
+
+            processed_tool_calls = []
+
+            if hasattr(response, 'content') and response.content:
+                for content_block in response.content:
+                    if content_block.type == "text":
+                        message["content"] = content_block.text
+                    elif content_block.type == "tool_use":
+                        tool_call_dict = {
+                            "id": f"call_{content_block.id}",
+                            "type": "function",
+                            "function": {
+                                "name": content_block.name,
+                                "arguments": content_block.input if isinstance(content_block.input,
+                                                                               str) else json.dumps(content_block.input)
+                            }
+                        }
+                        processed_tool_calls.append(ToolCall.from_dict(tool_call_dict))
+            else:
+                message["content"] = ""
+
+            if processed_tool_calls:
+                message["tool_calls"] = [tool_call.to_dict() for tool_call in processed_tool_calls]
+
+            # Extract usage information
+            usage = {
+                "completion_tokens": 0,
+                "prompt_tokens": 0,
+                "total_tokens": 0
+            }
+
+            if hasattr(response, 'usage'):
+                if hasattr(response.usage, 'output_tokens'):
+                    usage["completion_tokens"] = response.usage.output_tokens
+                if hasattr(response.usage, 'input_tokens'):
+                    usage["prompt_tokens"] = response.usage.input_tokens
+                if hasattr(response.usage, 'input_tokens') and hasattr(response.usage, 'output_tokens'):
+                    usage["total_tokens"] = response.usage.input_tokens + response.usage.output_tokens
+
+            # Create ModelResponse
+            return cls(
+                id=response.id if hasattr(response,
+                                          'id') else f"chatcmpl-anthropic-{hash(str(response)) & 0xffffffff:08x}",
+                model=response.model if hasattr(response, 'model') else "claude",
+                content=message["content"],
+                tool_calls=processed_tool_calls or None,
+                usage=usage,
+                raw_response=response,
+                message=message
+            )
+        except Exception as e:
+            if isinstance(e, LLMResponseError):
+                raise e
+            raise LLMResponseError(
+                f"Error processing Anthropic response: {str(e)}",
+                response.model if hasattr(response, 'model') else response.get('model', 'unknown'),
+                response)
+
+    @classmethod
+    def from_error(cls, error_msg: str, model: str = "unknown") -> 'ModelResponse':
+        """
+        Create ModelResponse from error message
+
+        Args:
+            error_msg: Error message
+            model: Model name
+
+        Returns:
+            ModelResponse object
+        """
+        return cls(
+            id="error",
+            model=model,
+            error=error_msg,
+            message={"role": "assistant", "content": f"Error: {error_msg}"}
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Convert ModelResponse to dictionary representation
+
+        Returns:
+            Dictionary representation
+        """
+        tool_calls_dict = None
+        if self.tool_calls:
+            tool_calls_dict = [tool_call.to_dict() for tool_call in self.tool_calls]
+
+        return {
+            "id": self.id,
+            "model": self.model,
+            "content": self.content,
+            "tool_calls": tool_calls_dict,
+            "usage": self.usage,
+            "error": self.error,
+            "message": self.message
+        }
+
+    def get_message(self) -> Dict[str, Any]:
+        """
+        Return message object that can be directly used for subsequent API calls
+
+        Returns:
+            Message object dictionary
+        """
+        return self.message
+
+    def serialize_tool_calls(self) -> List[Dict[str, Any]]:
+        """
+        Convert tool call objects to JSON format, handling OpenAI object types
+
+        Returns:
+            List[Dict[str, Any]]: Tool calls list in JSON format
+        """
+        if not self.tool_calls:
+            return []
+
+        result = []
+        for tool_call in self.tool_calls:
+            if hasattr(tool_call, 'to_dict'):
+                result.append(tool_call.to_dict())
+            elif isinstance(tool_call, dict):
+                result.append(tool_call)
+            else:
+                result.append(str(tool_call))
+        return result
+
+    def __repr__(self):
+        return json.dumps(self.to_dict(), ensure_ascii=False, indent=None,
+                          default=lambda obj: obj.to_dict() if hasattr(obj, 'to_dict') else str(obj))
+
+    def _serialize_message(self) -> Dict[str, Any]:
+        """
+        Serialize message object
+
+        Returns:
+            Dict[str, Any]: Serialized message dictionary
+        """
+        if not self.message:
+            return {}
+
+        result = {}
+
+        # Copy basic fields
+        for key, value in self.message.items():
+            if key == 'tool_calls':
+                # Handle tool_calls
+                result[key] = self.serialize_tool_calls()
+            else:
+                result[key] = value
+
+        return result

aworld/models/openai_provider.py

@@ -0,0 +1,633 @@
+import os
+from typing import Any, Dict, List, Generator, AsyncGenerator
+
+from openai import OpenAI, AsyncOpenAI
+
+from aworld.config.conf import ClientType
+from aworld.core.llm_provider_base import LLMProviderBase
+from aworld.models.llm_http_handler import LLMHTTPHandler
+from aworld.models.model_response import ModelResponse, LLMResponseError
+from aworld.logs.util import logger
+from aworld.models.utils import usage_process
+
+
+class OpenAIProvider(LLMProviderBase):
+    """OpenAI provider implementation.
+    """
+
+    def _init_provider(self):
+        """Initialize OpenAI provider.
+        
+        Returns:
+            OpenAI provider instance.
+        """
+        # Get API key
+        api_key = self.api_key
+        if not api_key:
+            env_var = "OPENAI_API_KEY"
+            api_key = os.getenv(env_var, "")
+            if not api_key:
+                raise ValueError(
+                    f"OpenAI API key not found, please set {env_var} environment variable or provide it in the parameters")
+        base_url = self.base_url
+        if not base_url:
+            base_url = os.getenv("OPENAI_ENDPOINT", "https://api.openai.com/v1")
+
+        self.is_http_provider = False
+        if self.kwargs.get("client_type", ClientType.SDK) == ClientType.HTTP:
+            logger.info(f"Using HTTP provider for OpenAI")
+            self.http_provider = LLMHTTPHandler(
+                base_url=base_url,
+                api_key=api_key,
+                model_name=self.model_name,
+                max_retries=self.kwargs.get("max_retries", 3)
+            )
+            self.is_http_provider = True
+            return self.http_provider
+        else:
+            return OpenAI(
+                api_key=api_key,
+                base_url=base_url,
+                timeout=self.kwargs.get("timeout", 180),
+                max_retries=self.kwargs.get("max_retries", 3)
+            )
+
+    def _init_async_provider(self):
+        """Initialize async OpenAI provider.
+
+        Returns:
+            Async OpenAI provider instance.
+        """
+        # Get API key
+        api_key = self.api_key
+        if not api_key:
+            env_var = "OPENAI_API_KEY"
+            api_key = os.getenv(env_var, "")
+            if not api_key:
+                raise ValueError(
+                    f"OpenAI API key not found, please set {env_var} environment variable or provide it in the parameters")
+        base_url = self.base_url
+        if not base_url:
+            base_url = os.getenv("OPENAI_ENDPOINT", "https://api.openai.com/v1")
+
+        return AsyncOpenAI(
+            api_key=api_key,
+            base_url=base_url,
+            timeout=self.kwargs.get("timeout", 180),
+            max_retries=self.kwargs.get("max_retries", 3)
+        )
+
+    @classmethod
+    def supported_models(cls) -> list[str]:
+        return ["gpt-4o", "gpt-4", "gpt-3.5-turbo", "o3-mini", "gpt-4o-mini", "deepseek-chat", "deepseek-reasoner",
+                r"qwq-.*", r"qwen-.*"]
+
+    def preprocess_messages(self, messages: List[Dict[str, str]]) -> List[Dict[str, str]]:
+        """Preprocess messages, use OpenAI format directly.
+
+        Args:
+            messages: OpenAI format message list.
+
+        Returns:
+            Processed message list.
+        """
+        for message in messages:
+            if message["role"] == "assistant" and "tool_calls" in message and message["tool_calls"]:
+                if message["content"] is None: message["content"] = ""
+                for tool_call in message["tool_calls"]:
+                    if "function" not in tool_call and "name" in tool_call and "arguments" in tool_call:
+                        tool_call["function"] = {"name": tool_call["name"], "arguments": tool_call["arguments"]}
+
+        return messages
+
+    def postprocess_response(self, response: Any) -> ModelResponse:
+        """Process OpenAI response.
+
+        Args:
+            response: OpenAI response object.
+
+        Returns:
+            ModelResponse object.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        if ((not isinstance(response, dict) and (not hasattr(response, 'choices') or not response.choices))
+                or (isinstance(response, dict) and not response.get("choices"))):
+            error_msg = ""
+            if hasattr(response, 'error') and response.error and isinstance(response.error, dict):
+                error_msg = response.error.get('message', '')
+            elif hasattr(response, 'msg'):
+                error_msg = response.msg
+
+            raise LLMResponseError(
+                error_msg if error_msg else "Unknown error",
+                self.model_name or "unknown",
+                response
+            )
+
+        return ModelResponse.from_openai_response(response)
+
+    def postprocess_stream_response(self, chunk: Any) -> ModelResponse:
+        """Process OpenAI streaming response chunk.
+
+        Args:
+            chunk: OpenAI response chunk.
+
+        Returns:
+            ModelResponse object.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        # Check if chunk contains error
+        if hasattr(chunk, 'error') or (isinstance(chunk, dict) and chunk.get('error')):
+            error_msg = chunk.error if hasattr(chunk, 'error') else chunk.get('error', 'Unknown error')
+            raise LLMResponseError(
+                error_msg,
+                self.model_name or "unknown",
+                chunk
+            )
+
+        # process tool calls
+        if (hasattr(chunk, 'choices') and chunk.choices and chunk.choices[0].delta and chunk.choices[0].delta.tool_calls) or (
+                isinstance(chunk, dict) and chunk.get("choices") and chunk["choices"] and chunk["choices"][0].get("delta", {}).get("tool_calls")):
+            tool_calls = chunk.choices[0].delta.tool_calls if hasattr(chunk, 'choices') else chunk["choices"][0].get("delta", {}).get("tool_calls")
+
+            for tool_call in tool_calls:
+                index = tool_call.index if hasattr(tool_call, 'index') else tool_call["index"]
+                func_name = tool_call.function.name if hasattr(tool_call, 'function') else tool_call.get("function", {}).get("name")
+                func_args = tool_call.function.arguments if hasattr(tool_call, 'function') else tool_call.get("function", {}).get("arguments")
+                if index >= len(self.stream_tool_buffer):
+                    self.stream_tool_buffer.append({
+                        "id": tool_call.id if hasattr(tool_call, 'id') else tool_call.get("id"),
+                        "type": "function",
+                        "function": {
+                            "name": func_name,
+                            "arguments": func_args
+                        }
+                    })
+                else:
+                    self.stream_tool_buffer[index]["function"]["arguments"] += func_args
+            processed_chunk = chunk
+            if hasattr(processed_chunk, 'choices'):
+                processed_chunk.choices[0].delta.tool_calls = None
+            else:
+                processed_chunk["choices"][0]["delta"]["tool_calls"] = None
+            resp = ModelResponse.from_openai_stream_chunk(processed_chunk)
+            if (not resp.content and not resp.usage.get("total_tokens", 0)):
+                return None
+        if (hasattr(chunk, 'choices') and chunk.choices and chunk.choices[0].finish_reason) or (
+                isinstance(chunk, dict) and chunk.get("choices") and chunk["choices"] and chunk["choices"][0].get(
+            "finish_reason")):
+            finish_reason = chunk.choices[0].finish_reason if hasattr(chunk, 'choices') else chunk["choices"][0].get(
+                "finish_reason")
+            if self.stream_tool_buffer:
+                tool_call_chunk = {
+                    "id": chunk.id if hasattr(chunk, 'id') else chunk.get("id"),
+                    "model": chunk.model if hasattr(chunk, 'model') else chunk.get("model"),
+                    "object": chunk.object if hasattr(chunk, 'object') else chunk.get("object"),
+                    "choices": [
+                        {
+                            "delta": {
+                                "role": "assistant",
+                                "content": "",
+                                "tool_calls": self.stream_tool_buffer
+                            }
+                        }
+                    ]
+                }
+                self.stream_tool_buffer = []
+                return ModelResponse.from_openai_stream_chunk(tool_call_chunk)
+
+        return ModelResponse.from_openai_stream_chunk(chunk)
+
+    def completion(self,
+                   messages: List[Dict[str, str]],
+                   temperature: float = 0.0,
+                   max_tokens: int = None,
+                   stop: List[str] = None,
+                   **kwargs) -> ModelResponse:
+        """Synchronously call OpenAI to generate response.
+        
+        Args:
+            messages: Message list.
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            ModelResponse object.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        if not self.provider:
+            raise RuntimeError(
+                "Sync provider not initialized. Make sure 'sync_enabled' parameter is set to True in initialization.")
+
+        processed_messages = self.preprocess_messages(messages)
+
+        try:
+            openai_params = self.get_openai_params(processed_messages, temperature, max_tokens, stop, **kwargs)
+            if self.is_http_provider:
+                response = self.http_provider.sync_call(openai_params)
+            else:
+                response = self.provider.chat.completions.create(**openai_params)
+
+            if (hasattr(response, 'code') and response.code != 0) or (
+                    isinstance(response, dict) and response.get("code", 0) != 0):
+                error_msg = getattr(response, 'msg', 'Unknown error')
+                logger.warn(f"API Error: {error_msg}")
+                raise LLMResponseError(error_msg, kwargs.get("model_name", self.model_name or "unknown"), response)
+
+            if not response:
+                raise LLMResponseError("Empty response", kwargs.get("model_name", self.model_name or "unknown"))
+
+            resp = self.postprocess_response(response)
+            usage_process(resp.usage)
+            return resp
+        except Exception as e:
+            if isinstance(e, LLMResponseError):
+                raise e
+            logger.warn(f"Error in OpenAI completion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "unknown"))
+
+    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 OpenAI to generate streaming response.
+        
+        Args:
+            messages: Message list.
+            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.
+        """
+        if not self.provider:
+            raise RuntimeError(
+                "Sync provider not initialized. Make sure 'sync_enabled' parameter is set to True in initialization.")
+
+        processed_messages = self.preprocess_messages(messages)
+        usage={
+            "completion_tokens": 0,
+            "prompt_tokens": 0,
+            "total_tokens": 0
+        }
+
+        try:
+            openai_params = self.get_openai_params(processed_messages, temperature, max_tokens, stop, **kwargs)
+            openai_params["stream"] = True
+            if self.is_http_provider:
+                response_stream = self.http_provider.sync_stream_call(openai_params)
+            else:
+                response_stream = self.provider.chat.completions.create(**openai_params)
+
+            for chunk in response_stream:
+                if not chunk:
+                    continue
+                resp = self.postprocess_stream_response(chunk)
+                if resp:
+                    self._accumulate_chunk_usage(usage, resp.usage)
+                    yield resp
+            usage_process(usage)
+
+        except Exception as e:
+            logger.warn(f"Error in stream_completion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "unknown"))
+
+    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 OpenAI to generate streaming response.
+        
+        Args:
+            messages: Message list.
+            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.
+        """
+        if not self.async_provider:
+            raise RuntimeError(
+                "Async provider not initialized. Make sure 'async_enabled' parameter is set to True in initialization.")
+
+        processed_messages = self.preprocess_messages(messages)
+        usage = {
+            "completion_tokens": 0,
+            "prompt_tokens": 0,
+            "total_tokens": 0
+        }
+
+        try:
+            openai_params = self.get_openai_params(processed_messages, temperature, max_tokens, stop, **kwargs)
+            openai_params["stream"] = True
+
+            if self.is_http_provider:
+                async for chunk in self.http_provider.async_stream_call(openai_params):
+                    if not chunk:
+                        continue
+                    resp = self.postprocess_stream_response(chunk)
+                    self._accumulate_chunk_usage(usage, resp.usage)
+                    yield resp
+            else:
+                response_stream = await self.async_provider.chat.completions.create(**openai_params)
+                async for chunk in response_stream:
+                    if not chunk:
+                        continue
+                    resp = self.postprocess_stream_response(chunk)
+                    if resp:
+                        self._accumulate_chunk_usage(usage, resp.usage)
+                        yield resp
+            usage_process(usage)
+
+        except Exception as e:
+            logger.warn(f"Error in astream_completion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "unknown"))
+
+    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 OpenAI to generate response.
+        
+        Args:
+            messages: Message list.
+            temperature: Temperature parameter.
+            max_tokens: Maximum number of tokens to generate.
+            stop: List of stop sequences.
+            **kwargs: Other parameters.
+
+        Returns:
+            ModelResponse object.
+            
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        if not self.async_provider:
+            raise RuntimeError(
+                "Async provider not initialized. Make sure 'async_enabled' parameter is set to True in initialization.")
+
+        processed_messages = self.preprocess_messages(messages)
+
+        try:
+            openai_params = self.get_openai_params(processed_messages, temperature, max_tokens, stop, **kwargs)
+            if self.is_http_provider:
+                response = await self.http_provider.async_call(openai_params)
+            else:
+                response = await self.async_provider.chat.completions.create(**openai_params)
+
+            if (hasattr(response, 'code') and response.code != 0) or (
+                    isinstance(response, dict) and response.get("code", 0) != 0):
+                error_msg = getattr(response, 'msg', 'Unknown error')
+                logger.warn(f"API Error: {error_msg}")
+                raise LLMResponseError(error_msg, kwargs.get("model_name", self.model_name or "unknown"), response)
+
+            if not response:
+                raise LLMResponseError("Empty response", kwargs.get("model_name", self.model_name or "unknown"))
+
+            resp = self.postprocess_response(response)
+            usage_process(resp.usage)
+            return resp
+        except Exception as e:
+            if isinstance(e, LLMResponseError):
+                raise e
+            logger.warn(f"Error in acompletion: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model_name", self.model_name or "unknown"))
+
+    def get_openai_params(self,
+                          messages: List[Dict[str, str]],
+                          temperature: float = 0.0,
+                          max_tokens: int = None,
+                          stop: List[str] = None,
+                          **kwargs) -> Dict[str, Any]:
+        openai_params = {
+            "model": kwargs.get("model_name", self.model_name or ""),
+            "messages": messages,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
+            "stop": stop
+        }
+
+        supported_params = [
+            "max_completion_tokens", "meta_data", "modalities", "n", "parallel_tool_calls",
+            "prediction", "reasoning_effort", "service_tier", "stream_options", "web_search_options"
+            "frequency_penalty", "logit_bias", "logprobs", "top_logprobs",
+            "presence_penalty", "response_format", "seed", "stream", "top_p",
+            "user", "function_call", "functions", "tools", "tool_choice"
+        ]
+
+        for param in supported_params:
+            if param in kwargs:
+                openai_params[param] = kwargs[param]
+
+        return openai_params
+
+    def speech_to_text(self,
+                       audio_file: str,
+                       language: str = None,
+                       prompt: str = None,
+                       **kwargs) -> ModelResponse:
+        """Convert speech to text.
+
+        Uses OpenAI's speech-to-text API to convert audio files to text.
+
+        Args:
+            audio_file: Path to audio file or file object.
+            language: Audio language, optional.
+            prompt: Transcription prompt, optional.
+            **kwargs: Other parameters, may include:
+                - model: Transcription model name, defaults to "whisper-1".
+                - response_format: Response format, defaults to "text".
+                - temperature: Sampling temperature, defaults to 0.
+
+        Returns:
+            ModelResponse: Unified model response object, with content field containing the transcription result.
+
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        if not self.provider:
+            raise RuntimeError(
+                "Sync provider not initialized. Make sure 'sync_enabled' parameter is set to True in initialization.")
+
+        try:
+            # Prepare parameters
+            transcription_params = {
+                "model": kwargs.get("model", "whisper-1"),
+                "response_format": kwargs.get("response_format", "text"),
+                "temperature": kwargs.get("temperature", 0)
+            }
+
+            # Add optional parameters
+            if language:
+                transcription_params["language"] = language
+            if prompt:
+                transcription_params["prompt"] = prompt
+
+            # Open file (if path is provided)
+            if isinstance(audio_file, str):
+                with open(audio_file, "rb") as file:
+                    transcription_response = self.provider.audio.transcriptions.create(
+                        file=file,
+                        **transcription_params
+                    )
+            else:
+                # If already a file object
+                transcription_response = self.provider.audio.transcriptions.create(
+                    file=audio_file,
+                    **transcription_params
+                )
+
+            # Create ModelResponse
+            return ModelResponse(
+                id=f"stt-{hash(str(transcription_response)) & 0xffffffff:08x}",
+                model=transcription_params["model"],
+                content=transcription_response.text if hasattr(transcription_response, 'text') else str(
+                    transcription_response),
+                raw_response=transcription_response,
+                message={
+                    "role": "assistant",
+                    "content": transcription_response.text if hasattr(transcription_response, 'text') else str(
+                        transcription_response)
+                }
+            )
+        except Exception as e:
+            logger.warn(f"Speech-to-text error: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model", "whisper-1"))
+
+    async def aspeech_to_text(self,
+                              audio_file: str,
+                              language: str = None,
+                              prompt: str = None,
+                              **kwargs) -> ModelResponse:
+        """Asynchronously convert speech to text.
+
+        Uses OpenAI's speech-to-text API to convert audio files to text.
+
+        Args:
+            audio_file: Path to audio file or file object.
+            language: Audio language, optional.
+            prompt: Transcription prompt, optional.
+            **kwargs: Other parameters, may include:
+                - model: Transcription model name, defaults to "whisper-1".
+                - response_format: Response format, defaults to "text".
+                - temperature: Sampling temperature, defaults to 0.
+
+        Returns:
+            ModelResponse: Unified model response object, with content field containing the transcription result.
+
+        Raises:
+            LLMResponseError: When LLM response error occurs.
+        """
+        if not self.async_provider:
+            raise RuntimeError(
+                "Async provider not initialized. Make sure 'async_enabled' parameter is set to True in initialization.")
+
+        try:
+            # Prepare parameters
+            transcription_params = {
+                "model": kwargs.get("model", "whisper-1"),
+                "response_format": kwargs.get("response_format", "text"),
+                "temperature": kwargs.get("temperature", 0)
+            }
+
+            # Add optional parameters
+            if language:
+                transcription_params["language"] = language
+            if prompt:
+                transcription_params["prompt"] = prompt
+
+            # Open file (if path is provided)
+            if isinstance(audio_file, str):
+                with open(audio_file, "rb") as file:
+                    transcription_response = await self.async_provider.audio.transcriptions.create(
+                        file=file,
+                        **transcription_params
+                    )
+            else:
+                # If already a file object
+                transcription_response = await self.async_provider.audio.transcriptions.create(
+                    file=audio_file,
+                    **transcription_params
+                )
+
+            # Create ModelResponse
+            return ModelResponse(
+                id=f"stt-{hash(str(transcription_response)) & 0xffffffff:08x}",
+                model=transcription_params["model"],
+                content=transcription_response.text if hasattr(transcription_response, 'text') else str(
+                    transcription_response),
+                raw_response=transcription_response,
+                message={
+                    "role": "assistant",
+                    "content": transcription_response.text if hasattr(transcription_response, 'text') else str(
+                        transcription_response)
+                }
+            )
+        except Exception as e:
+            logger.warn(f"Async speech-to-text error: {e}")
+            raise LLMResponseError(str(e), kwargs.get("model", "whisper-1"))
+
+
+class AzureOpenAIProvider(OpenAIProvider):
+    """Azure OpenAI provider implementation.
+    """
+
+    def _init_provider(self):
+        """Initialize Azure OpenAI provider.
+
+        Returns:
+            Azure OpenAI provider instance.
+        """
+        from langchain_openai import AzureChatOpenAI
+
+        # Get API key
+        api_key = self.api_key
+        if not api_key:
+            env_var = "AZURE_OPENAI_API_KEY"
+            api_key = os.getenv(env_var, "")
+            if not api_key:
+                raise ValueError(
+                    f"Azure OpenAI API key not found, please set {env_var} environment variable or provide it in the parameters")
+
+        # Get API version
+        api_version = self.kwargs.get("api_version", "") or os.getenv("AZURE_OPENAI_API_VERSION", "2025-01-01-preview")
+
+        # Get endpoint
+        azure_endpoint = self.base_url
+        if not azure_endpoint:
+            azure_endpoint = os.getenv("AZURE_OPENAI_ENDPOINT", "")
+            if not azure_endpoint:
+                raise ValueError(
+                    "Azure OpenAI endpoint not found, please set AZURE_OPENAI_ENDPOINT environment variable or provide it in the parameters")
+
+        return AzureChatOpenAI(
+            model=self.model_name or "gpt-4o",
+            temperature=self.kwargs.get("temperature", 0.0),
+            api_version=api_version,
+            azure_endpoint=azure_endpoint,
+            api_key=api_key
+        )

aworld/models/openai_tokenizer.py

@@ -0,0 +1,237 @@
+# Copyright 2024 AWorld Team. All rights reserved.
+# 
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+# 
+#    http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tokenization classes for OpenAI models."""
+
+import base64
+import unicodedata
+from pathlib import Path
+from typing import Collection, Dict, List, Set, Union
+from aworld.logs.util import logger
+from aworld.utils import import_package
+import_package("tiktoken")
+import tiktoken
+
+VOCAB_FILES_NAMES = {'vocab_file': 'cl100k_base.tiktoken'}
+
+# OpenAI GPT tokenizer pattern
+PAT_STR = r"""(?i:'s|'t|'re|'ve|'m|'ll|'d)|[^\r\n\p{L}\p{N}]?\p{L}+|\p{N}| ?[^\s\p{L}\p{N}]+[\r\n]*|\s*[\r\n]+|\s+(?!\S)|\s+"""
+
+# OpenAI special tokens
+ENDOFTEXT = '<|endoftext|>'
+SPECIAL_TOKENS = {
+    ENDOFTEXT: 100256,
+}
+
+
+def _load_tiktoken_bpe(tiktoken_bpe_file: str) -> Dict[bytes, int]:
+    """Load tiktoken BPE file similar to qwen_tokenizer."""
+    with open(tiktoken_bpe_file, 'rb') as f:
+        contents = f.read()
+    return {
+        base64.b64decode(token): int(rank) for token, rank in (line.split() for line in contents.splitlines() if line)
+    }
+
+
+class OpenAITokenizer:
+    """OpenAI tokenizer using local tiktoken file."""
+
+    vocab_files_names = VOCAB_FILES_NAMES
+
+    def __init__(
+        self,
+        vocab_file=None,
+        errors='replace',
+        extra_vocab_file=None,
+    ):
+        if not vocab_file:
+            vocab_file = VOCAB_FILES_NAMES['vocab_file']
+        self._decode_use_source_tokenizer = False
+
+        # how to handle errors in decoding UTF-8 byte sequences
+        # use ignore if you are in streaming inference
+        self.errors = errors
+
+        self.mergeable_ranks = _load_tiktoken_bpe(vocab_file)  # type: Dict[bytes, int]
+        self.special_tokens = SPECIAL_TOKENS.copy()
+
+        # try load extra vocab from file
+        if extra_vocab_file is not None:
+            used_ids = set(self.mergeable_ranks.values()) | set(self.special_tokens.values())
+            extra_mergeable_ranks = _load_tiktoken_bpe(extra_vocab_file)
+            for token, index in extra_mergeable_ranks.items():
+                if token in self.mergeable_ranks:
+                    logger.info(f'extra token {token} exists, skipping')
+                    continue
+                if index in used_ids:
+                    logger.info(f'the index {index} for extra token {token} exists, skipping')
+                    continue
+                self.mergeable_ranks[token] = index
+            # the index may be sparse after this, but don't worry tiktoken.Encoding will handle this
+
+        enc = tiktoken.Encoding(
+            'cl100k_base',
+            pat_str=PAT_STR,
+            mergeable_ranks=self.mergeable_ranks,
+            special_tokens=self.special_tokens,
+        )
+        assert len(self.mergeable_ranks) + len(
+            self.special_tokens
+        ) == enc.n_vocab, f'{len(self.mergeable_ranks) + len(self.special_tokens)} != {enc.n_vocab} in encoding'
+
+        self.decoder = {v: k for k, v in self.mergeable_ranks.items()}  # type: dict[int, bytes|str]
+        self.decoder.update({v: k for k, v in self.special_tokens.items()})
+
+        self.tokenizer = enc  # type: tiktoken.Encoding
+
+        self.eod_id = self.special_tokens[ENDOFTEXT]
+
+    def __getstate__(self):
+        # for pickle lovers
+        state = self.__dict__.copy()
+        del state['tokenizer']
+        return state
+
+    def __setstate__(self, state):
+        # tokenizer is not python native; don't pass it; rebuild it
+        self.__dict__.update(state)
+        enc = tiktoken.Encoding(
+            'cl100k_base',
+            pat_str=PAT_STR,
+            mergeable_ranks=self.mergeable_ranks,
+            special_tokens=self.special_tokens,
+        )
+        self.tokenizer = enc
+
+    def __len__(self) -> int:
+        return self.tokenizer.n_vocab
+
+    def get_vocab(self) -> Dict[bytes, int]:
+        return self.mergeable_ranks
+
+    def convert_tokens_to_ids(self, tokens: Union[bytes, str, List[Union[bytes, str]]]) -> List[int]:
+        ids = []
+        if isinstance(tokens, (str, bytes)):
+            if tokens in self.special_tokens:
+                return self.special_tokens[tokens]
+            else:
+                return self.mergeable_ranks.get(tokens)
+        for token in tokens:
+            if token in self.special_tokens:
+                ids.append(self.special_tokens[token])
+            else:
+                ids.append(self.mergeable_ranks.get(token))
+        return ids
+
+    def tokenize(
+            self,
+            text: str,
+            allowed_special: Union[Set, str] = 'all',
+            disallowed_special: Union[Collection, str] = (),
+    ) -> List[Union[bytes, str]]:
+        """
+        Converts a string in a sequence of tokens.
+
+        Args:
+            text (`str`):
+                The sequence to be encoded.
+            allowed_special (`Literal["all"]` or `set`):
+                The surface forms of the tokens to be encoded as special tokens in regular texts.
+                Default to "all".
+            disallowed_special (`Literal["all"]` or `Collection`):
+                The surface forms of the tokens that should not be in regular texts and trigger errors.
+                Default to an empty tuple.
+
+        Returns:
+            `List[bytes|str]`: The list of tokens.
+        """
+        tokens = []
+        if text is None:
+            return tokens
+        text = unicodedata.normalize('NFC', text)
+
+        # this implementation takes a detour: text -> token id -> token surface forms
+        for t in self.tokenizer.encode(text, allowed_special=allowed_special, disallowed_special=disallowed_special):
+            tokens.append(self.decoder[t])
+        return tokens
+
+    def convert_tokens_to_string(self, tokens: List[Union[bytes, str]]) -> str:
+        """
+        Converts a sequence of tokens in a single string.
+        """
+        text = ''
+        temp = b''
+        for t in tokens:
+            if isinstance(t, str):
+                if temp:
+                    text += temp.decode('utf-8', errors=self.errors)
+                    temp = b''
+                text += t
+            elif isinstance(t, bytes):
+                temp += t
+            else:
+                raise TypeError('token should only be of type types or str')
+        if temp:
+            text += temp.decode('utf-8', errors=self.errors)
+        return text
+
+    @property
+    def vocab_size(self):
+        return self.tokenizer.n_vocab
+
+    def _decode(
+        self,
+        token_ids: Union[int, List[int]],
+        skip_special_tokens: bool = False,
+        errors: str = None,
+    ) -> str:
+        if isinstance(token_ids, int):
+            token_ids = [token_ids]
+        if skip_special_tokens:
+            token_ids = [i for i in token_ids if i < self.eod_id]
+        return self.tokenizer.decode(token_ids, errors=errors or self.errors)
+
+    def encode(self, text: str) -> List[int]:
+        return self.tokenizer.encode(text)
+
+    def decode(self, token_ids: Union[int, List[int]], errors: str = None) -> str:
+        return self._decode(token_ids, errors=errors)
+
+    def count_tokens(self, text: str) -> int:
+        return len(self.encode(text))
+
+    def truncate(self, text: str, max_token: int, start_token: int = 0, keep_both_sides: bool = False) -> str:
+        max_token = int(max_token)
+        token_ids = self.encode(text)[start_token:]
+        if len(token_ids) <= max_token:
+            return self.decode(token_ids)
+
+        if keep_both_sides:
+            ellipsis_tokens = self.encode("...")
+            ellipsis_len = len(ellipsis_tokens)
+            available = max_token - ellipsis_len
+            if available <= 0: # Degenerate case: not enough space even for "..."
+                return self.decode(token_ids[:max_token])
+
+            left_len = available // 2
+            right_len = available - left_len
+            token_ids = token_ids[:left_len] + ellipsis_tokens + token_ids[-right_len:]
+        else:
+            token_ids = token_ids[:max_token]
+
+        return self.decode(token_ids)
+
+
+# Default tokenizer instance using local cl100k_base.tiktoken
+openai_tokenizer = OpenAITokenizer(Path(__file__).resolve().parent.parent / 'config' / 'cl100k_base.tiktoken')

aworld/models/qwen_tokenizer.py

@@ -0,0 +1,245 @@
+# Copyright 2023 The Qwen team, Alibaba Group. All rights reserved.
+# 
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+# 
+#    http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tokenization classes for QWen."""
+
+import base64
+import unicodedata
+from pathlib import Path
+from typing import Collection, Dict, List, Set, Union
+from aworld.logs.util import logger
+from aworld.utils import import_package
+import_package("tiktoken")
+import tiktoken
+
+VOCAB_FILES_NAMES = {'vocab_file': 'qwen.tiktoken'}
+
+PAT_STR = r"""(?i:'s|'t|'re|'ve|'m|'ll|'d)|[^\r\n\p{L}\p{N}]?\p{L}+|\p{N}| ?[^\s\p{L}\p{N}]+[\r\n]*|\s*[\r\n]+|\s+(?!\S)|\s+"""
+ENDOFTEXT = '<|endoftext|>'
+IMSTART = '<|im_start|>'
+IMEND = '<|im_end|>'
+# as the default behavior is changed to allow special tokens in
+# regular texts, the surface forms of special tokens need to be
+# as different as possible to minimize the impact
+EXTRAS = tuple((f'<|extra_{i}|>' for i in range(205)))
+# changed to use actual index to avoid misconfiguration with vocabulary expansion
+SPECIAL_START_ID = 151643
+SPECIAL_TOKENS = tuple(enumerate(
+    ((
+        ENDOFTEXT,
+        IMSTART,
+        IMEND,
+    ) + EXTRAS),
+    start=SPECIAL_START_ID,
+))
+SPECIAL_TOKENS_SET = set(t for i, t in SPECIAL_TOKENS)
+
+
+def _load_tiktoken_bpe(tiktoken_bpe_file: str) -> Dict[bytes, int]:
+    with open(tiktoken_bpe_file, 'rb') as f:
+        contents = f.read()
+    return {
+        base64.b64decode(token): int(rank) for token, rank in (line.split() for line in contents.splitlines() if line)
+    }
+
+
+class QWenTokenizer:
+    """QWen tokenizer."""
+
+    vocab_files_names = VOCAB_FILES_NAMES
+
+    def __init__(
+        self,
+        vocab_file=None,
+        errors='replace',
+        extra_vocab_file=None,
+    ):
+        if not vocab_file:
+            vocab_file = VOCAB_FILES_NAMES['vocab_file']
+        self._decode_use_source_tokenizer = False
+
+        # how to handle errors in decoding UTF-8 byte sequences
+        # use ignore if you are in streaming inference
+        self.errors = errors
+
+        self.mergeable_ranks = _load_tiktoken_bpe(vocab_file)  # type: Dict[bytes, int]
+        self.special_tokens = {token: index for index, token in SPECIAL_TOKENS}
+
+        # try load extra vocab from file
+        if extra_vocab_file is not None:
+            used_ids = set(self.mergeable_ranks.values()) | set(self.special_tokens.values())
+            extra_mergeable_ranks = _load_tiktoken_bpe(extra_vocab_file)
+            for token, index in extra_mergeable_ranks.items():
+                if token in self.mergeable_ranks:
+                    logger.info(f'extra token {token} exists, skipping')
+                    continue
+                if index in used_ids:
+                    logger.info(f'the index {index} for extra token {token} exists, skipping')
+                    continue
+                self.mergeable_ranks[token] = index
+            # the index may be sparse after this, but don't worry tiktoken.Encoding will handle this
+
+        enc = tiktoken.Encoding(
+            'Qwen',
+            pat_str=PAT_STR,
+            mergeable_ranks=self.mergeable_ranks,
+            special_tokens=self.special_tokens,
+        )
+        assert len(self.mergeable_ranks) + len(
+            self.special_tokens
+        ) == enc.n_vocab, f'{len(self.mergeable_ranks) + len(self.special_tokens)} != {enc.n_vocab} in encoding'
+
+        self.decoder = {v: k for k, v in self.mergeable_ranks.items()}  # type: dict[int, bytes|str]
+        self.decoder.update({v: k for k, v in self.special_tokens.items()})
+
+        self.tokenizer = enc  # type: tiktoken.Encoding
+
+        self.eod_id = self.tokenizer.eot_token
+        self.im_start_id = self.special_tokens[IMSTART]
+        self.im_end_id = self.special_tokens[IMEND]
+
+    def __getstate__(self):
+        # for pickle lovers
+        state = self.__dict__.copy()
+        del state['tokenizer']
+        return state
+
+    def __setstate__(self, state):
+        # tokenizer is not python native; don't pass it; rebuild it
+        self.__dict__.update(state)
+        enc = tiktoken.Encoding(
+            'Qwen',
+            pat_str=PAT_STR,
+            mergeable_ranks=self.mergeable_ranks,
+            special_tokens=self.special_tokens,
+        )
+        self.tokenizer = enc
+
+    def __len__(self) -> int:
+        return self.tokenizer.n_vocab
+
+    def get_vocab(self) -> Dict[bytes, int]:
+        return self.mergeable_ranks
+
+    def convert_tokens_to_ids(self, tokens: Union[bytes, str, List[Union[bytes, str]]]) -> List[int]:
+        ids = []
+        if isinstance(tokens, (str, bytes)):
+            if tokens in self.special_tokens:
+                return self.special_tokens[tokens]
+            else:
+                return self.mergeable_ranks.get(tokens)
+        for token in tokens:
+            if token in self.special_tokens:
+                ids.append(self.special_tokens[token])
+            else:
+                ids.append(self.mergeable_ranks.get(token))
+        return ids
+
+    def tokenize(
+            self,
+            text: str,
+            allowed_special: Union[Set, str] = 'all',
+            disallowed_special: Union[Collection, str] = (),
+    ) -> List[Union[bytes, str]]:
+        """
+        Converts a string in a sequence of tokens.
+
+        Args:
+            text (`str`):
+                The sequence to be encoded.
+            allowed_special (`Literal["all"]` or `set`):
+                The surface forms of the tokens to be encoded as special tokens in regular texts.
+                Default to "all".
+            disallowed_special (`Literal["all"]` or `Collection`):
+                The surface forms of the tokens that should not be in regular texts and trigger errors.
+                Default to an empty tuple.
+
+        Returns:
+            `List[bytes|str]`: The list of tokens.
+        """
+        tokens = []
+        if text is None:
+            return tokens
+        text = unicodedata.normalize('NFC', text)
+
+        # this implementation takes a detour: text -> token id -> token surface forms
+        for t in self.tokenizer.encode(text, allowed_special=allowed_special, disallowed_special=disallowed_special):
+            tokens.append(self.decoder[t])
+        return tokens
+
+    def convert_tokens_to_string(self, tokens: List[Union[bytes, str]]) -> str:
+        """
+        Converts a sequence of tokens in a single string.
+        """
+        text = ''
+        temp = b''
+        for t in tokens:
+            if isinstance(t, str):
+                if temp:
+                    text += temp.decode('utf-8', errors=self.errors)
+                    temp = b''
+                text += t
+            elif isinstance(t, bytes):
+                temp += t
+            else:
+                raise TypeError('token should only be of type types or str')
+        if temp:
+            text += temp.decode('utf-8', errors=self.errors)
+        return text
+
+    @property
+    def vocab_size(self):
+        return self.tokenizer.n_vocab
+
+    def _decode(
+        self,
+        token_ids: Union[int, List[int]],
+        skip_special_tokens: bool = False,
+        errors: str = None,
+    ) -> str:
+        if isinstance(token_ids, int):
+            token_ids = [token_ids]
+        if skip_special_tokens:
+            token_ids = [i for i in token_ids if i < self.eod_id]
+        return self.tokenizer.decode(token_ids, errors=errors or self.errors)
+
+    def encode(self, text: str) -> List[int]:
+        return self.convert_tokens_to_ids(self.tokenize(text))
+
+    def count_tokens(self, text: str) -> int:
+        return len(self.tokenize(text))
+
+    def truncate(self, text: str, max_token: int, start_token: int = 0, keep_both_sides: bool = False) -> str:
+        max_token = int(max_token)
+        token_list = self.tokenize(text)[start_token:]
+        if len(token_list) <= max_token:
+            return self.convert_tokens_to_string(token_list)
+
+        if keep_both_sides:
+            ellipsis_tokens = self.tokenize("...")
+            ellipsis_len = len(ellipsis_tokens)
+            available = max_token - ellipsis_len
+            if available <= 0: # Degenerate case: not enough space even for "..."
+                return self.convert_tokens_to_string(token_list[:max_token])
+
+            left_len = available // 2
+            right_len = available - left_len
+            token_list = token_list[:left_len] + ellipsis_tokens + token_list[-right_len:]
+        else:
+            token_list = token_list[:max_token]
+
+        return self.convert_tokens_to_string(token_list)
+
+
+qwen_tokenizer = QWenTokenizer(Path(__file__).resolve().parent.parent / 'config' / 'qwen.tiktoken')

aworld/models/utils.py

@@ -0,0 +1,195 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import copy
+import inspect
+import os.path
+from typing import Dict, Any, List, Union
+
+from aworld.core.context.base import Context
+from aworld.logs.util import logger
+from aworld.models.qwen_tokenizer import qwen_tokenizer
+from aworld.models.openai_tokenizer import openai_tokenizer
+from aworld.utils import import_package
+
+
+def usage_process(usage: Dict[str, Union[int, Dict[str, int]]] = {}, context: Context = None):
+    if not context:
+        context = Context.instance()
+
+    stacks = inspect.stack()
+    index = 0
+    for idx, stack in enumerate(stacks):
+        index = idx + 1
+        file = os.path.basename(stack.filename)
+        # supported use `llm.py` utility function only
+        if 'call_llm_model' in stack.function and file == 'llm.py':
+            break
+
+    if index >= len(stacks):
+        logger.warning("not category usage find to count")
+    else:
+        instance = stacks[index].frame.f_locals.get('self')
+        name = getattr(instance, "_name", "unknown")
+        usage[name] = copy.copy(usage)
+    # total usage
+    context.add_token(usage)
+
+
+def num_tokens_from_messages(messages, model="gpt-4o"):
+    """Return the number of tokens used by a list of messages."""
+    import_package("tiktoken")
+    import tiktoken
+
+    if model.lower() == "qwen":
+        encoding = qwen_tokenizer
+    elif model.lower() == "openai":
+        encoding = openai_tokenizer
+    else:
+        try:
+            encoding = tiktoken.encoding_for_model(model)
+        except KeyError:
+            logger.warning(f"{model} model not found. Using cl100k_base encoding.")
+            encoding = tiktoken.get_encoding("cl100k_base")
+
+    tokens_per_message = 3
+    tokens_per_name = 1
+
+    num_tokens = 0
+    for message in messages:
+        num_tokens += tokens_per_message
+        if isinstance(message, str):
+            num_tokens += len(encoding.encode(message))
+        else:
+            for key, value in message.items():
+                num_tokens += len(encoding.encode(str(value)))
+                if key == "name":
+                    num_tokens += tokens_per_name
+    num_tokens += 3
+    return num_tokens
+
+def truncate_tokens_from_messages(messages: List[Dict[str, Any]], max_tokens: int, keep_both_sides: bool = False, model: str = "gpt-4o"):
+    import_package("tiktoken")
+    import tiktoken
+
+    if model.lower() == "qwen":
+        return qwen_tokenizer.truncate(messages, max_tokens, keep_both_sides)
+    elif model.lower() == "openai":
+        return openai_tokenizer.truncate(messages, max_tokens, keep_both_sides)
+    
+    try:
+        encoding = tiktoken.encoding_for_model(model)
+    except KeyError:
+        logger.warning(f"{model} model not found. Using cl100k_base encoding.")
+        encoding = tiktoken.get_encoding("cl100k_base")
+        
+    return encoding.truncate(messages, max_tokens, keep_both_sides)
+    
+def agent_desc_transform(agent_dict: Dict[str, Any],
+                         agents: List[str] = None,
+                         provider: str = 'openai',
+                         strategy: str = 'min') -> List[Dict[str, Any]]:
+    """Default implement transform framework standard protocol to openai protocol of agent description.
+
+    Args:
+        agent_dict: Dict of descriptions of agents that are registered in the agent factory.
+        agents: Description of special agents to use.
+        provider: Different descriptions formats need to be processed based on the provider.
+        strategy: The value is `min` or `max`, when no special agents are provided, `min` indicates no content returned,
+                 `max` means get all agents' descriptions.
+    """
+    agent_as_tools = []
+    if not agents and strategy == 'min':
+        return agent_as_tools
+
+    if provider and 'openai' in provider:
+        for agent_name, agent_info in agent_dict.items():
+            if agents and agent_name not in agents:
+                logger.debug(f"{agent_name} can not supported in {agents}, you can set `tools` params to support it.")
+                continue
+
+            for action in agent_info["abilities"]:
+                # Build parameter properties
+                properties = {}
+                required = []
+                for param_name, param_info in action["params"].items():
+                    properties[param_name] = {
+                        "description": param_info["desc"],
+                        "type": param_info["type"] if param_info["type"] != "str" else "string"
+                    }
+                    if param_info.get("required", False):
+                        required.append(param_name)
+
+                openai_function_schema = {
+                    "name": f'{agent_name}__{action["name"]}',
+                    "description": action["desc"],
+                    "parameters": {
+                        "type": "object",
+                        "properties": properties,
+                        "required": required
+                    }
+                }
+
+                agent_as_tools.append({
+                    "type": "function",
+                    "function": openai_function_schema
+                })
+    return agent_as_tools
+
+
+def tool_desc_transform(tool_dict: Dict[str, Any],
+                        tools: List[str] = None,
+                        black_tool_actions: Dict[str, List[str]] = {},
+                        provider: str = 'openai',
+                        strategy: str = 'min') -> List[Dict[str, Any]]:
+    """Default implement transform framework standard protocol to openai protocol of tool description.
+
+    Args:
+        tool_dict: Dict of descriptions of tools that are registered in the agent factory.
+        tools: Description of special tools to use.
+        provider: Different descriptions formats need to be processed based on the provider.
+        strategy: The value is `min` or `max`, when no special tools are provided, `min` indicates no content returned,
+                 `max` means get all tools' descriptions.
+    """
+    openai_tools = []
+    if not tools and strategy == 'min':
+        return openai_tools
+
+    if black_tool_actions is None:
+        black_tool_actions = {}
+
+    if provider and 'openai' in provider:
+        for tool_name, tool_info in tool_dict.items():
+            if tools and tool_name not in tools:
+                logger.debug(f"{tool_name} can not supported in {tools}, you can set `tools` params to support it.")
+                continue
+
+            black_actions = black_tool_actions.get(tool_name, [])
+            for action in tool_info["actions"]:
+                if action['name'] in black_actions:
+                    continue
+                # Build parameter properties
+                properties = {}
+                required = []
+                for param_name, param_info in action["params"].items():
+                    properties[param_name] = {
+                        "description": param_info["desc"],
+                        "type": param_info["type"] if param_info["type"] != "str" else "string"
+                    }
+                    if param_info.get("required", False):
+                        required.append(param_name)
+
+                openai_function_schema = {
+                    "name": f'{tool_name}__{action["name"]}',
+                    "description": action["desc"],
+                    "parameters": {
+                        "type": "object",
+                        "properties": properties,
+                        "required": required
+                    }
+                }
+
+                openai_tools.append({
+                    "type": "function",
+                    "function": openai_function_schema
+                })
+    return openai_tools