跳转至

🧬 模型接口

evoagentx.models

LLMOutputParser 

LLMOutputParser(**kwargs)

Bases: Parser

A basic parser for LLM-generated content.

This parser stores the raw text generated by an LLM in the .content attribute and provides methods to extract structured data from this text using different parsing strategies.

Attributes:

Name Type Description
content str

The raw text generated by the LLM.
Source code in evoagentx/core/module.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def __init__(self, **kwargs):
    """
    Initializes a BaseModule instance.

    Args:
        **kwargs (Any): Keyword arguments used to initialize the instance

    Raises:
        ValidationError: When parameter validation fails
        Exception: When other errors occur during initialization
    """

    try:
        for field_name, _ in type(self).model_fields.items():
            field_value = kwargs.get(field_name, None)
            if field_value:
                kwargs[field_name] = self._process_data(field_value)
            # if field_value and isinstance(field_value, dict) and "class_name" in field_value:
            #     class_name = field_value.get("class_name")
            #     sub_cls = MODULE_REGISTRY.get_module(cls_name=class_name)
            #     kwargs[field_name] = sub_cls._create_instance(field_value)
        super().__init__(**kwargs) 
        self.init_module()
    except (ValidationError, Exception) as e:
        exception_handler = callback_manager.get_callback("exception_buffer")
        if exception_handler is None:
            error_message = get_base_module_init_error_message(
                cls=self.__class__, 
                data=kwargs, 
                errors=e
            )
            logger.error(error_message)
            raise
        else:
            exception_handler.add(e)

get_attrs classmethod 

get_attrs(return_type: bool = False) -> List[Union[str, tuple]]

Returns the attributes of the LLMOutputParser class.

Excludes ["class_name", "content"] by default.

Parameters:

Name Type Description Default
return_type bool

Whether to return the type of the attributes along with their names.

 False

Returns:

Type Description
List[Union[str, tuple]]

If return_type is True, returns a list of tuples where each tuple contains

List[Union[str, tuple]]

the attribute name and its type. Otherwise, returns a list of attribute names.
Source code in evoagentx/models/base_model.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@classmethod
def get_attrs(cls, return_type: bool = False) -> List[Union[str, tuple]]:
    """Returns the attributes of the LLMOutputParser class.

    Excludes ["class_name", "content"] by default.

    Args:
        return_type: Whether to return the type of the attributes along with their names.

    Returns:
        If `return_type` is True, returns a list of tuples where each tuple contains 
        the attribute name and its type. Otherwise, returns a list of attribute names.
    """
    attrs = [] 
    exclude_attrs = ["class_name", "content"]
    for field, field_info in cls.model_fields.items():
        if field not in exclude_attrs:
            if return_type:
                field_type = get_type_name(field_info.annotation)
                attrs.append((field, field_type))
            else:
                attrs.append(field)
    return attrs

get_attr_descriptions classmethod 

get_attr_descriptions() -> dict

Returns the attributes and their descriptions.

Returns:

Type Description
dict

A dictionary mapping attribute names to their descriptions.
Source code in evoagentx/models/base_model.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@classmethod
def get_attr_descriptions(cls) -> dict:
    """Returns the attributes and their descriptions.

    Returns:
        A dictionary mapping attribute names to their descriptions.
    """
    attrs = cls.get_attrs()
    results = {} 
    for field_name, field_info in cls.model_fields.items():
        if field_name not in attrs:
            continue
        field_desc = field_info.description if field_info.description is not None else "None"
        results[field_name] = field_desc
    return results

get_content_data classmethod 

get_content_data(content: str, parse_mode: str = 'json', parse_func: Optional[Callable] = None, **kwargs) -> dict

Parses LLM-generated content into a dictionary.

This method takes content from an LLM response and converts it to a structured dictionary based on the specified parsing mode.

Parameters:

Name Type Description Default
content str

The content to parse.

 required
parse_mode str

The mode to parse the content. Must be one of: - 'str': Assigns the raw text content to all attributes of the parser. - 'json': Extracts and parses JSON objects from LLM output. It will return a dictionary parsed from the first valid JSON string. - 'xml': Parses content using XML tags. It will return a dictionary parsed from the XML tags. - 'title': Parses content with Markdown-style headings. - 'custom': Uses custom parsing logic. Requires providing parse_func parameter as a custom parsing function.

 'json'
parse_func Optional[Callable]

The function to parse the content, only valid when parse_mode is 'custom'.

 None
**kwargs Any

Additional arguments passed to the parsing function.

 {}

Returns:

Type Description
dict

The parsed content as a dictionary.

Raises:

Type Description
ValueError

If parse_mode is invalid or if parse_func is not provided when parse_mode is 'custom'.
Source code in evoagentx/models/base_model.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@classmethod
def get_content_data(cls, content: str, parse_mode: str = "json", parse_func: Optional[Callable] = None, **kwargs) -> dict:
    """Parses LLM-generated content into a dictionary.

    This method takes content from an LLM response and converts it to a structured
    dictionary based on the specified parsing mode.

    Args:
        content: The content to parse.
        parse_mode: The mode to parse the content. Must be one of:
            - 'str': Assigns the raw text content to all attributes of the parser. 
            - 'json': Extracts and parses JSON objects from LLM output. It will return a dictionary parsed from the first valid JSON string.
            - 'xml': Parses content using XML tags. It will return a dictionary parsed from the XML tags.
            - 'title': Parses content with Markdown-style headings.
            - 'custom': Uses custom parsing logic. Requires providing `parse_func` parameter as a custom parsing function.
        parse_func: The function to parse the content, only valid when parse_mode is 'custom'.
        **kwargs (Any): Additional arguments passed to the parsing function.

    Returns:
        The parsed content as a dictionary.

    Raises:
        ValueError: If parse_mode is invalid or if parse_func is not provided when parse_mode is 'custom'.
    """
    attrs = cls.get_attrs()
    if len(attrs) <= 0:
        return {} 

    if parse_mode == "str":
        parse_func = cls._parse_str_content
    elif parse_mode == "json":
        parse_func = cls._parse_json_content
    elif parse_mode == "xml":
        parse_func = cls._parse_xml_content
    elif parse_mode == "title":
        parse_func = cls._parse_title_content
    elif parse_mode == "custom":
        if parse_func is None:
            raise ValueError("`parse_func` must be provided when `parse_mode` is 'custom'.")
        # obtain the function inputs
        signature = inspect.signature(parse_func)
        if "content" not in signature.parameters:
            raise ValueError("`parse_func` must have an input argument `content`.")

        func_args = {}
        func_args["content"] = content
        for param_name, param in signature.parameters.items():
            if param_name == "content":
                continue  # Already set
            if param_name in kwargs:
                func_args[param_name] = kwargs[param_name]
        data = parse_func(**func_args)
        if not isinstance(data, dict):
            raise ValueError(f"The output of `parse_func` must be a dictionary, but found {type(data)}.")
        return data
    else:
        raise ValueError(f"Invalid value '{parse_mode}' detected for `parse_mode`. Available choices: {PARSER_VALID_MODE}")
    data = parse_func(content=content, **kwargs)
    return data

parse classmethod 

parse(content: str, parse_mode: str = 'json', parse_func: Optional[Callable] = None, **kwargs) -> LLMOutputParser

Parses LLM-generated text into a structured parser instance.

This is the main method for creating parser instances from LLM output.

Parameters:

Name Type Description Default
content str

The text generated by the LLM.

 required
parse_mode str

The mode to parse the content, must be one of: - 'str': Assigns the raw text content to all attributes of the parser. - 'json': Extracts and parses JSON objects from LLM output. Uses the first valid JSON string to create an instance of LLMOutputParser. - 'xml': Parses content using XML tags. Uses the XML tags to create an instance of LLMOutputParser. - 'title': Parses content with Markdown-style headings. Uses the Markdown-style headings to create an instance of LLMOutputParser. The default title format is "## {title}", you can change it by providing title_format parameter, which should be a string that contains {title} placeholder. - 'custom': Uses custom parsing logic. Requires providing parse_func parameter as a custom parsing function. The parse_func must have a parameter named content and return a dictionary where the keys are the attribute names and the values are the parsed data.

'json'
parse_func Optional[Callable]

The function to parse the content, only valid when parse_mode is 'custom'.

 None
**kwargs Any

Additional arguments passed to parsing functions, such as: - title_format for parse_mode="title".

 {}

Returns:

Type Description
LLMOutputParser

An instance of LLMOutputParser containing the parsed data.

Raises:

Type Description
ValueError

If parse_mode is invalid or if content is not a string.
Source code in evoagentx/models/base_model.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
@classmethod
def parse(cls, content: str, parse_mode: str = "json", parse_func: Optional[Callable] = None, **kwargs) -> "LLMOutputParser":
    """Parses LLM-generated text into a structured parser instance.

    This is the main method for creating parser instances from LLM output.

    Args:
        content: The text generated by the LLM.
        parse_mode: The mode to parse the content, must be one of:
            - 'str': Assigns the raw text content to all attributes of the parser. 
            - 'json': Extracts and parses JSON objects from LLM output. Uses the first valid JSON string to create an instance of LLMOutputParser.
            - 'xml': Parses content using XML tags. Uses the XML tags to create an instance of LLMOutputParser.
            - 'title': Parses content with Markdown-style headings. Uses the Markdown-style headings to create an instance of LLMOutputParser. The default title format is "## {title}", you can change it by providing `title_format` parameter, which should be a string that contains `{title}` placeholder. 
            - 'custom': Uses custom parsing logic. Requires providing `parse_func` parameter as a custom parsing function. The `parse_func` must have a parameter named `content` and return a dictionary where the keys are the attribute names and the values are the parsed data. 
        parse_func: The function to parse the content, only valid when `parse_mode` is 'custom'.
        **kwargs (Any): Additional arguments passed to parsing functions, such as:
            - `title_format` for `parse_mode="title"`.

    Returns:
        An instance of LLMOutputParser containing the parsed data.

    Raises:
        ValueError: If parse_mode is invalid or if content is not a string.
    """
    if parse_mode not in PARSER_VALID_MODE:
        raise ValueError(f"'{parse_mode}' is an invalid value for `parse_mode`. Available choices: {PARSER_VALID_MODE}.")
    if not isinstance(content, str):
        raise ValueError(f"The input to {cls.__name__}.parse should be a str, but found {type(content)}.")
    data = cls.get_content_data(content=content, parse_mode=parse_mode, parse_func=parse_func, **kwargs)
    data.update({"content": content})
    parser = cls.from_dict(data, **kwargs)
    # parser.content = content
    return parser

__str__ 

__str__() -> str

Returns a string representation of the parser.

Source code in evoagentx/models/base_model.py 
311
312
313
314
315
def __str__(self) -> str:
    """
    Returns a string representation of the parser.
    """
    return self.to_str()

to_str 

to_str(**kwargs) -> str

Converts the parser to a string.

Source code in evoagentx/models/base_model.py 
317
318
319
320
321
def to_str(self, **kwargs) -> str:
    """
    Converts the parser to a string.
    """
    return self.content

get_structured_data 

get_structured_data() -> dict

Extracts structured data from the parser.

Returns:

Type Description
dict

A dictionary containing only the defined attributes and their values,
dict

excluding metadata like class_name.
Source code in evoagentx/models/base_model.py 
323
324
325
326
327
328
329
330
331
332
333
334
def get_structured_data(self) -> dict:
    """Extracts structured data from the parser.

    Returns:
        A dictionary containing only the defined attributes and their values,
        excluding metadata like class_name.
    """
    attrs = type(self).get_attrs()
    data = self.to_dict(ignore=["class_name"])
    # structured_data = {attr: data[attr] for attr in attrs}
    structured_data = {key: value for key, value in data.items() if key in attrs}
    return structured_data

BaseConfig 

BaseConfig(**kwargs)

Bases: BaseModule

Base configuration class that serves as parent for all configuration classes.

A config should inherit BaseConfig and specify the attributes and their types. Otherwise this will be an empty config.

Source code in evoagentx/core/module.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def __init__(self, **kwargs):
    """
    Initializes a BaseModule instance.

    Args:
        **kwargs (Any): Keyword arguments used to initialize the instance

    Raises:
        ValidationError: When parameter validation fails
        Exception: When other errors occur during initialization
    """

    try:
        for field_name, _ in type(self).model_fields.items():
            field_value = kwargs.get(field_name, None)
            if field_value:
                kwargs[field_name] = self._process_data(field_value)
            # if field_value and isinstance(field_value, dict) and "class_name" in field_value:
            #     class_name = field_value.get("class_name")
            #     sub_cls = MODULE_REGISTRY.get_module(cls_name=class_name)
            #     kwargs[field_name] = sub_cls._create_instance(field_value)
        super().__init__(**kwargs) 
        self.init_module()
    except (ValidationError, Exception) as e:
        exception_handler = callback_manager.get_callback("exception_buffer")
        if exception_handler is None:
            error_message = get_base_module_init_error_message(
                cls=self.__class__, 
                data=kwargs, 
                errors=e
            )
            logger.error(error_message)
            raise
        else:
            exception_handler.add(e)

save 

save(path: str, **kwargs) -> str

Save configuration to the specified path.

Parameters:

Name Type Description Default
path str

The file path to save the configuration

 required
**kwargs Any

Additional keyword arguments passed to save_module method

 {}

Returns:

Name Type Description
str str

The path where the file was saved
Source code in evoagentx/core/base_config.py 
13
14
15
16
17
18
19
20
21
22
23
24
def save(self, path: str, **kwargs)-> str:

    """Save configuration to the specified path.

    Args:
        path: The file path to save the configuration
        **kwargs (Any): Additional keyword arguments passed to save_module method

    Returns:
        str: The path where the file was saved
    """
    return super().save_module(path, **kwargs)

get_config_params 

get_config_params() -> List[str]

Get a list of configuration parameters.

Returns:

Type Description
List[str]

List[str]: List of configuration parameter names, excluding 'class_name'
Source code in evoagentx/core/base_config.py 
26
27
28
29
30
31
32
33
34
def get_config_params(self) -> List[str]:
    """Get a list of configuration parameters.

    Returns:
        List[str]: List of configuration parameter names, excluding 'class_name'
    """
    config_params = list(type(self).model_fields.keys())
    config_params.remove("class_name")
    return config_params

get_set_params 

get_set_params(ignore: List[str] = []) -> dict

Get a dictionary of explicitly set parameters.

Parameters:

Name Type Description Default
ignore List[str]

List of parameter names to ignore

 []

Returns:

Name Type Description
dict dict

Dictionary of explicitly set parameters, excluding 'class_name' and ignored parameters
Source code in evoagentx/core/base_config.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def get_set_params(self, ignore: List[str] = []) -> dict:
    """Get a dictionary of explicitly set parameters.

    Args:
        ignore: List of parameter names to ignore

    Returns:
        dict: Dictionary of explicitly set parameters, excluding 'class_name' and ignored parameters
    """
    explicitly_set_fields = {field: getattr(self, field) for field in self.model_fields_set}
    if self.kwargs:
        explicitly_set_fields.update(self.kwargs)
    for field in ignore:
        explicitly_set_fields.pop(field, None)
    explicitly_set_fields.pop("class_name", None)
    return explicitly_set_fields

LiteLLM 

LiteLLM(config: LLMConfig, **kwargs)

Bases: OpenAILLM

Source code in evoagentx/models/base_model.py 
349
350
351
352
353
354
355
356
357
358
def __init__(self, config: LLMConfig, **kwargs):
    """Initializes the LLM with configuration.

    Args:
        config: Configuration object for the LLM.
        **kwargs (Any): Additional keyword arguments.
    """
    self.config = config
    self.kwargs = kwargs
    self.init_model()

init_model 

init_model()

Initialize the model based on the configuration.

Source code in evoagentx/models/litellm_model.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def init_model(self):
    """
    Initialize the model based on the configuration.
    """
    # Check if llm_type is correct
    if self.config.llm_type != "LiteLLM":
        raise ValueError("llm_type must be 'LiteLLM'")

    # Set model and extract the company name
    self.model = self.config.model
    # company = self.model.split("/")[0] if "/" in self.model else "openai"
    company = infer_litellm_company_from_model(self.model)

    # Set environment variables based on the company
    if company == "openai":
        if not self.config.openai_key:
            raise ValueError("OpenAI API key is required for OpenAI models. You should set `openai_key` in LiteLLMConfig")
        os.environ["OPENAI_API_KEY"] = self.config.openai_key
    elif company == "deepseek":
        if not self.config.deepseek_key:
            raise ValueError("DeepSeek API key is required for DeepSeek models. You should set `deepseek_key` in LiteLLMConfig")
        os.environ["DEEPSEEK_API_KEY"] = self.config.deepseek_key
    elif company == "anthropic":
        if not self.config.anthropic_key:
            raise ValueError("Anthropic API key is required for Anthropic models. You should set `anthropic_key` in LiteLLMConfig")
        os.environ["ANTHROPIC_API_KEY"] = self.config.anthropic_key
    elif company == "gemini":
        if not self.config.gemini_key:
            raise ValueError("Gemini API key is required for Gemini models. You should set `gemini_key` in LiteLLMConfig")
        os.environ["GEMINI_API_KEY"] = self.config.gemini_key 
    elif company == "meta_llama":
        if not self.config.meta_llama_key:
            raise ValueError("Meta Llama API key is required for Meta Llama models. You should set `meta_llama_key` in LiteLLMConfig")
        os.environ["LLAMA_API_KEY"] = self.config.meta_llama_key
    elif company == "openrouter":
        if not self.config.openrouter_key:
            raise ValueError("OpenRouter API key is required for OpenRouter models. You should set `openrouter_key` in LiteLLMConfig. You can also set `openrouter_base` in LiteLLMConfig to use a custom base URL [optional]")
        os.environ["OPENROUTER_API_KEY"] = self.config.openrouter_key
        os.environ["OPENROUTER_API_BASE"] = self.config.openrouter_base # [optional]
    elif company == "perplexity":
        if not self.config.perplexity_key:
            raise ValueError("Perplexity API key is required for Perplexity models. You should set `perplexity_key` in LiteLLMConfig")
        os.environ["PERPLEXITYAI_API_KEY"] = self.config.perplexity_key
    elif company == "groq":
        if not self.config.groq_key:
            raise ValueError("Groq API key is required for Groq models. You should set `groq_key` in LiteLLMConfig")
        os.environ["GROQ_API_KEY"] = self.config.groq_key
    else:
        raise ValueError(f"Unsupported company: {company}")

    self._default_ignore_fields = [
        "llm_type", "output_response", "openai_key", "deepseek_key", "anthropic_key", 
        "gemini_key", "meta_llama_key", "openrouter_key", "openrouter_base", "perplexity_key", 
        "groq_key"
    ] # parameters in OpenAILLMConfig that are not OpenAI models' input parameters

single_generate 

single_generate(messages: List[dict], **kwargs) -> str

Generate a single response using the completion function.

Parameters:

Name Type Description Default
messages List[dict]

A list of dictionaries representing the conversation history.

 required
**kwargs Any

Additional parameters to be passed to the completion function.

 {}

Returns:

Name Type Description
str str

A string containing the model's response.
Source code in evoagentx/models/litellm_model.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(5))
def single_generate(self, messages: List[dict], **kwargs) -> str:

    """
    Generate a single response using the completion function.

    Args: 
        messages (List[dict]): A list of dictionaries representing the conversation history.
        **kwargs (Any): Additional parameters to be passed to the `completion` function.

    Returns: 
        str: A string containing the model's response.
    """
    stream = kwargs["stream"] if "stream" in kwargs else self.config.stream
    output_response = kwargs["output_response"] if "output_response" in kwargs else self.config.output_response

    try:
        completion_params = self.get_completion_params(**kwargs)
        response = completion(messages=messages, **completion_params)
        if stream:
            output = self.get_stream_output(response, output_response=output_response)
            cost = self._stream_cost(messages=messages, output=output)
        else:
            output: str = self.get_completion_output(response=response, output_response=output_response)
            cost = self._completion_cost(response=response)
        self._update_cost(cost=cost)
    except Exception as e:
        raise RuntimeError(f"Error during single_generate: {str(e)}")

    return output

batch_generate 

batch_generate(batch_messages: List[List[dict]], **kwargs) -> List[str]

Generate responses for a batch of messages.

Parameters:

Name Type Description Default
batch_messages List[List[dict]]

A list of message lists, where each sublist represents a conversation.

 required
**kwargs Any

Additional parameters to be passed to the completion function.

 {}

Returns:

Type Description
List[str]

List[str]: A list of responses for each conversation.
Source code in evoagentx/models/litellm_model.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def batch_generate(self, batch_messages: List[List[dict]], **kwargs) -> List[str]:
    """
    Generate responses for a batch of messages.

    Args: 
        batch_messages (List[List[dict]]): A list of message lists, where each sublist represents a conversation.
        **kwargs (Any): Additional parameters to be passed to the `completion` function.

    Returns: 
        List[str]: A list of responses for each conversation.
    """
    results = []
    for messages in batch_messages:
        response = self.single_generate(messages, **kwargs)
        results.append(response)
    return results

single_generate_async async 

single_generate_async(messages: List[dict], **kwargs) -> str

Generate a single response using the async completion function.

Parameters:

Name Type Description Default
messages List[dict]

A list of dictionaries representing the conversation history.

 required
**kwargs Any

Additional parameters to be passed to the completion function.

 {}

Returns:

Name Type Description
str str

A string containing the model's response.
Source code in evoagentx/models/litellm_model.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
async def single_generate_async(self, messages: List[dict], **kwargs) -> str:
    """
    Generate a single response using the async completion function.

    Args: 
        messages (List[dict]): A list of dictionaries representing the conversation history.
        **kwargs (Any): Additional parameters to be passed to the `completion` function.

    Returns: 
        str: A string containing the model's response.
    """
    stream = kwargs["stream"] if "stream" in kwargs else self.config.stream
    output_response = kwargs["output_response"] if "output_response" in kwargs else self.config.output_response

    try:
        completion_params = self.get_completion_params(**kwargs)
        response = await acompletion(messages=messages, **completion_params)
        if stream:
            if hasattr(response, "__aiter__"):
                output = await self.get_stream_output_async(response, output_response=output_response)
            else:
                output = self.get_stream_output(response, output_response=output_response)
            cost = self._stream_cost(messages=messages, output=output)
        else:
            output: str = self.get_completion_output(response=response, output_response=output_response)
            cost = self._completion_cost(response=response)
        self._update_cost(cost=cost)
    except Exception as e:
        raise RuntimeError(f"Error during single_generate_async: {str(e)}")

    return output

completion_cost 

completion_cost(completion_response=None, prompt='', messages: List = [], completion='', total_time=0.0, call_type='completion', size=None, quality=None, n=None) -> float

Calculate the cost of a given completion or other supported tasks.

Parameters:

Name Type Description Default
completion_response dict

The response received from a LiteLLM completion request.

 None
prompt str

Input prompt text.

 ''
messages list

Conversation history.

 []
completion str

Output text from the LLM.

 ''
total_time float

Total time used for request.

 0.0
call_type str

Type of request (e.g., "completion", "image_generation").

 'completion'
size str

Image size for image generation.

 None
quality str

Image quality for image generation.

 None
n int

Number of generated images.

 None

Returns:

Name Type Description
float float

The cost in USD.
Source code in evoagentx/models/litellm_model.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def completion_cost(
    self,
    completion_response=None,
    prompt="",
    messages: List = [],
    completion="",
    total_time=0.0,
    call_type="completion",
    size=None,
    quality=None,
    n=None,
) -> float:
    """
    Calculate the cost of a given completion or other supported tasks.

    Args:
        completion_response (dict): The response received from a LiteLLM completion request.
        prompt (str): Input prompt text.
        messages (list): Conversation history.
        completion (str): Output text from the LLM.
        total_time (float): Total time used for request.
        call_type (str): Type of request (e.g., "completion", "image_generation").
        size (str): Image size for image generation.
        quality (str): Image quality for image generation.
        n (int): Number of generated images.

    Returns:
        float: The cost in USD.
    """
    try:
        # Default parameters
        prompt_tokens = 0
        completion_tokens = 0
        model = self.model  # Use the class model by default

        # Handle completion response
        if completion_response:
            prompt_tokens = completion_response.get("usage", {}).get("prompt_tokens", 0)
            completion_tokens = completion_response.get("usage", {}).get("completion_tokens", 0)
            model = completion_response.get("model", model)
            size = completion_response.get("_hidden_params", {}).get("optional_params", {}).get("size", size)
            quality = completion_response.get("_hidden_params", {}).get("optional_params", {}).get("quality", quality)
            n = completion_response.get("_hidden_params", {}).get("optional_params", {}).get("n", n)

        # Handle manual token counting
        else:
            if messages:
                prompt_tokens = token_counter(model=model, messages=messages)
            elif prompt:
                prompt_tokens = token_counter(model=model, text=prompt)
            completion_tokens = token_counter(model=model, text=completion)

        # Ensure model is valid
        if not model:
            raise ValueError("Model is not defined for cost calculation.")

        # Image generation cost calculation
        if call_type in ["image_generation", "aimage_generation"]:
            if size and "x" in size and "-x-" not in size:
                size = size.replace("x", "-x-")
            height, width = map(int, size.split("-x-"))
            return (
                litellm.model_cost[f"{size}/{model}"]["input_cost_per_pixel"]
                * height * width * (n or 1)
            )

        # Regular completion cost calculation
        prompt_cost, completion_cost = cost_per_token(
            model=model,
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            response_time_ms=total_time,
        )
        return prompt_cost + completion_cost
    except Exception as e:
        print(f"Error calculating cost: {e}")
        return 0.0

BaseLLM 

BaseLLM(config: LLMConfig, **kwargs)

Bases: ABC

Abstract base class for Large Language Model implementations.

This class defines the interface that all LLM implementations must follow, providing methods for generating text, formatting messages, and parsing output.

Attributes:

Name Type Description
config

Configuration for the LLM.
kwargs

Additional keyword arguments provided during initialization.

Initializes the LLM with configuration.

Parameters:

Name Type Description Default
config LLMConfig

Configuration object for the LLM.

 required
**kwargs Any

Additional keyword arguments.

 {}
Source code in evoagentx/models/base_model.py 
349
350
351
352
353
354
355
356
357
358
def __init__(self, config: LLMConfig, **kwargs):
    """Initializes the LLM with configuration.

    Args:
        config: Configuration object for the LLM.
        **kwargs (Any): Additional keyword arguments.
    """
    self.config = config
    self.kwargs = kwargs
    self.init_model()

init_model abstractmethod 

init_model()

Initializes the underlying model.

This method should be implemented by subclasses to set up the actual LLM.

Source code in evoagentx/models/base_model.py 
360
361
362
363
364
365
366
@abstractmethod
def init_model(self):
    """Initializes the underlying model.

    This method should be implemented by subclasses to set up the actual LLM.
    """
    pass

__deepcopy__ 

__deepcopy__(memo) -> BaseLLM

Handles deep copying of the LLM instance.

Returns the same instance when deepcopy is called, as LLM instances often cannot be meaningfully deep-copied.

Parameters:

Name Type Description Default
memo Dict[int, Any]

Memo dictionary used by the deepcopy process.

 required

Returns:

Type Description
BaseLLM

The same LLM instance.
Source code in evoagentx/models/base_model.py 
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
def __deepcopy__(self, memo) -> "BaseLLM":
    """Handles deep copying of the LLM instance.

    Returns the same instance when deepcopy is called, as LLM instances
    often cannot be meaningfully deep-copied.

    Args:
        memo (Dict[int, Any]): Memo dictionary used by the deepcopy process.

    Returns:
        The same LLM instance.
    """
    # return the same instance when deepcopy
    memo[id(self)] = self
    return self

formulate_messages abstractmethod 

formulate_messages(prompts: List[str], system_messages: Optional[List[str]] = None) -> List[List[dict]]

Converts input prompts into the chat format compatible with different LLMs.

Parameters:

Name Type Description Default
prompts List[str]

A list of user prompts that need to be converted.

 required
system_messages Optional[List[str]]

An optional list of system messages that provide instructions or context to the model.

 None

Returns:

Type Description
List[List[dict]]

A list of message lists, where each inner list contains messages in the chat format required by LLMs.
Source code in evoagentx/models/base_model.py 
384
385
386
387
388
389
390
391
392
393
394
395
@abstractmethod
def formulate_messages(self, prompts: List[str], system_messages: Optional[List[str]] = None) -> List[List[dict]]:
    """Converts input prompts into the chat format compatible with different LLMs.

    Args:
        prompts: A list of user prompts that need to be converted.
        system_messages: An optional list of system messages that provide instructions or context to the model.

    Returns:
        A list of message lists, where each inner list contains messages in the chat format required by LLMs. 
    """
    pass

single_generate abstractmethod 

single_generate(messages: List[dict], **kwargs) -> str

Generates LLM output for a single set of messages.

Parameters:

Name Type Description Default
messages List[dict]

The input messages to the LLM in chat format.

 required
**kwargs Any

Additional keyword arguments for generation settings.

 {}

Returns:

Type Description
str

The generated output text from the LLM.
Source code in evoagentx/models/base_model.py 
397
398
399
400
401
402
403
404
405
406
407
408
@abstractmethod
def single_generate(self, messages: List[dict], **kwargs) -> str:
    """Generates LLM output for a single set of messages.

    Args:
        messages: The input messages to the LLM in chat format.
        **kwargs (Any): Additional keyword arguments for generation settings.

    Returns:
        The generated output text from the LLM.
    """
    pass

batch_generate abstractmethod 

batch_generate(batch_messages: List[List[dict]], **kwargs) -> List[str]

Generates outputs for a batch of message sets.

Parameters:

Name Type Description Default
batch_messages List[List[dict]]

A list of message lists, where each inner list contains messages for a single generation.

 required
**kwargs Any

Additional keyword arguments for generation settings.

 {}

Returns:

Type Description
List[str]

A list of generated outputs from the LLM, one for each input message set.
Source code in evoagentx/models/base_model.py 
410
411
412
413
414
415
416
417
418
419
420
421
@abstractmethod
def batch_generate(self, batch_messages: List[List[dict]], **kwargs) -> List[str]:
    """Generates outputs for a batch of message sets.

    Args: 
        batch_messages: A list of message lists, where each inner list contains messages for a single generation.
        **kwargs (Any): Additional keyword arguments for generation settings.

    Returns:
        A list of generated outputs from the LLM, one for each input message set.
    """
    pass

single_generate_async async 

single_generate_async(messages: List[dict], **kwargs) -> str

Asynchronously generates LLM output for a single set of messages.

This default implementation wraps the synchronous method in an async executor. Subclasses should override this for true async implementation if supported.

Parameters:

Name Type Description Default
messages List[dict]

The input messages to the LLM in chat format.

 required
**kwargs Any

Additional keyword arguments for generation settings.

 {}

Returns:

Type Description
str

The generated output text from the LLM.
Source code in evoagentx/models/base_model.py 
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
async def single_generate_async(self, messages: List[dict], **kwargs) -> str:
    """Asynchronously generates LLM output for a single set of messages.

    This default implementation wraps the synchronous method in an async executor.
    Subclasses should override this for true async implementation if supported.

    Args:
        messages: The input messages to the LLM in chat format.
        **kwargs (Any): Additional keyword arguments for generation settings.

    Returns:
        The generated output text from the LLM.
    """
    # Default implementation for backward compatibility
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, self.single_generate, messages, **kwargs)
    return result

batch_generate_async async 

batch_generate_async(batch_messages: List[List[dict]], **kwargs) -> List[str]

Asynchronously generates outputs for a batch of message sets.

This default implementation runs each generation as a separate async task. Subclasses should override this for more efficient async batching if supported.

Parameters:

Name Type Description Default
batch_messages List[List[dict]]

A list of message lists, where each inner list contains messages for a single generation.

 required
**kwargs Any

Additional keyword arguments for generation settings.

 {}

Returns:

Type Description
List[str]

A list of generated outputs from the LLM, one for each input message set.
Source code in evoagentx/models/base_model.py 
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
async def batch_generate_async(self, batch_messages: List[List[dict]], **kwargs) -> List[str]:
    """Asynchronously generates outputs for a batch of message sets.

    This default implementation runs each generation as a separate async task.
    Subclasses should override this for more efficient async batching if supported.

    Args: 
        batch_messages: A list of message lists, where each inner list contains messages for a single generation.
        **kwargs (Any): Additional keyword arguments for generation settings.

    Returns:
        A list of generated outputs from the LLM, one for each input message set.
    """
    # Default implementation for backward compatibility
    tasks = [self.single_generate_async(messages, **kwargs) for messages in batch_messages]
    return await asyncio.gather(*tasks)

parse_generated_text 

parse_generated_text(text: str, parser: Optional[Type[LLMOutputParser]] = None, parse_mode: Optional[str] = 'json', parse_func: Optional[Callable] = None, **kwargs) -> LLMOutputParser

Parses generated text into a structured output using a parser.

Parameters:

Name Type Description Default
text str

The text generated by the LLM.

 required
parser Optional[Type[LLMOutputParser]]

An LLMOutputParser class to use for parsing. If None, the default LLMOutputParser is used.

 None
parse_mode Optional[str]

The mode to use for parsing, must be the parse_mode supported by the parser.

'json'
**kwargs Any

Additional arguments passed to the parser.

 {}

Returns:

Type Description
LLMOutputParser

An LLMOutputParser instance containing the parsed data.
Source code in evoagentx/models/base_model.py 
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
def parse_generated_text(self, text: str, parser: Optional[Type[LLMOutputParser]]=None, parse_mode: Optional[str] = "json", parse_func: Optional[Callable] = None, **kwargs) -> LLMOutputParser:
    """Parses generated text into a structured output using a parser.

    Args: 
        text: The text generated by the LLM.
        parser: An LLMOutputParser class to use for parsing. If None, the default LLMOutputParser is used.
        parse_mode: The mode to use for parsing, must be the `parse_mode` supported by the `parser`. 
        **kwargs (Any): Additional arguments passed to the parser.

    Returns:
        An LLMOutputParser instance containing the parsed data.
    """
    if not parser:
        parser = LLMOutputParser
    return parser.parse(text, parse_mode=parse_mode, parse_func=parse_func)

parse_generated_texts 

parse_generated_texts(texts: List[str], parser: Optional[Type[LLMOutputParser]] = None, parse_mode: Optional[str] = 'json', parse_func: Optional[Callable] = None, **kwargs) -> List[LLMOutputParser]

Parses multiple generated texts into structured outputs.

Parameters:

Name Type Description Default
texts List[str]

A list of texts generated by the LLM.

 required
parser Optional[Type[LLMOutputParser]]

An LLMOutputParser class to use for parsing.

 None
parse_mode Optional[str]

The mode to use for parsing, must be the parse_mode supported by the parser.

'json'
**kwargs Any

Additional arguments passed to the parser.

 {}

Returns:

Type Description
List[LLMOutputParser]

A list of LLMOutputParser instances containing the parsed data.
Source code in evoagentx/models/base_model.py 
474
475
476
477
478
479
480
481
482
483
484
485
486
487
def parse_generated_texts(self, texts: List[str], parser: Optional[Type[LLMOutputParser]]=None, parse_mode: Optional[str] = "json", parse_func: Optional[Callable] = None, **kwargs) -> List[LLMOutputParser]:
    """Parses multiple generated texts into structured outputs.

    Args:
        texts: A list of texts generated by the LLM.
        parser: An LLMOutputParser class to use for parsing.
        parse_mode: The mode to use for parsing, must be the `parse_mode` supported by the `parser`. 
        **kwargs (Any): Additional arguments passed to the parser.

    Returns:
        A list of LLMOutputParser instances containing the parsed data.
    """
    parsed_results = [self.parse_generated_text(text=text, parser=parser, parse_mode=parse_mode, parse_func=parse_func, **kwargs) for text in texts]
    return parsed_results

generate 

generate(prompt: Optional[Union[str, List[str]]] = None, system_message: Optional[Union[str, List[str]]] = None, messages: Optional[Union[List[dict], List[List[dict]]]] = None, parser: Optional[Type[LLMOutputParser]] = None, parse_mode: Optional[str] = 'json', parse_func: Optional[Callable] = None, **kwargs) -> Union[LLMOutputParser, List[LLMOutputParser]]

Generates LLM output(s) and parses the result(s).

This is the main method for generating text with the LLM. It handles both single and batch generation, and automatically parses the outputs.

Parameters:

Name Type Description Default
prompt Optional[Union[str, List[str]]]

Input prompt(s) to the LLM.

 None
system_message Optional[Union[str, List[str]]]

System message(s) for the LLM.

 None
messages Optional[Union[List[dict], List[List[dict]]]]

Chat message(s) for the LLM, already in the required format (either prompt or messages must be provided).

 None
parser Optional[Type[LLMOutputParser]]

Parser class to use for processing the output.

 None
parse_mode Optional[str]

The mode to use for parsing, must be the parse_mode supported by the parser.

'json'
**kwargs Any

Additional generation configuration parameters.

 {}

Returns:

Type Description
Union[LLMOutputParser, List[LLMOutputParser]]

For single generation: An LLMOutputParser instance.
Union[LLMOutputParser, List[LLMOutputParser]]

For batch generation: A list of LLMOutputParser instances.

Raises:

Type Description
ValueError

If the input format is invalid.
Note

Either prompt or messages must be provided. If both or neither is provided, an error will be raised.

Source code in evoagentx/models/base_model.py 
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
def generate(
    self,
    prompt: Optional[Union[str, List[str]]] = None,
    system_message: Optional[Union[str, List[str]]] = None,
    messages: Optional[Union[List[dict],List[List[dict]]]] = None,
    parser: Optional[Type[LLMOutputParser]] = None,
    parse_mode: Optional[str] = "json", 
    parse_func: Optional[Callable] = None,
    **kwargs
) -> Union[LLMOutputParser, List[LLMOutputParser]]:
    """Generates LLM output(s) and parses the result(s).

    This is the main method for generating text with the LLM. It handles both
    single and batch generation, and automatically parses the outputs.

    Args:
        prompt: Input prompt(s) to the LLM.
        system_message: System message(s) for the LLM.
        messages: Chat message(s) for the LLM, already in the required format (either `prompt` or `messages` must be provided).
        parser: Parser class to use for processing the output.
        parse_mode: The mode to use for parsing, must be the `parse_mode` supported by the `parser`. 
        **kwargs (Any): Additional generation configuration parameters.

    Returns:
        For single generation: An LLMOutputParser instance.
        For batch generation: A list of LLMOutputParser instances.

    Raises:
        ValueError: If the input format is invalid.

    Note:
        Either prompt or messages must be provided. If both or neither is provided,
        an error will be raised.
    """
    prepared_messages, single_generate = self._prepare_messages(prompt, system_message, messages)
    if not prepared_messages:  # Handle empty messages case
        return []

    generated_texts = self.batch_generate(batch_messages=prepared_messages, **kwargs)
    parsed_outputs = self.parse_generated_texts(texts=generated_texts, parser=parser, parse_mode=parse_mode, parse_func=parse_func, **kwargs)
    return parsed_outputs[0] if single_generate else parsed_outputs

async_generate async 

async_generate(prompt: Optional[Union[str, List[str]]] = None, system_message: Optional[Union[str, List[str]]] = None, messages: Optional[Union[List[dict], List[List[dict]]]] = None, parser: Optional[Type[LLMOutputParser]] = None, parse_mode: Optional[str] = 'json', parse_func: Optional[Callable] = None, **kwargs) -> Union[LLMOutputParser, List[LLMOutputParser]]

Asynchronously generates LLM output(s) and parses the result(s).

This is the async version of the generate method. It works identically but performs the generation asynchronously.

Source code in evoagentx/models/base_model.py 
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
async def async_generate(
    self,
    prompt: Optional[Union[str, List[str]]] = None,
    system_message: Optional[Union[str, List[str]]] = None,
    messages: Optional[Union[List[dict],List[List[dict]]]] = None,
    parser: Optional[Type[LLMOutputParser]] = None,
    parse_mode: Optional[str] = "json", 
    parse_func: Optional[Callable] = None,
    **kwargs
) -> Union[LLMOutputParser, List[LLMOutputParser]]:
    """Asynchronously generates LLM output(s) and parses the result(s).

    This is the async version of the generate method. It works identically but
    performs the generation asynchronously.
    """
    prepared_messages, single_generate = self._prepare_messages(prompt, system_message, messages)
    if not prepared_messages:  # Handle empty messages case
        return []

    generated_texts = await self.batch_generate_async(batch_messages=prepared_messages, **kwargs)
    parsed_outputs = self.parse_generated_texts(texts=generated_texts, parser=parser, parse_mode=parse_mode, parse_func=parse_func, **kwargs)
    return parsed_outputs[0] if single_generate else parsed_outputs

OpenAILLM 

OpenAILLM(config: LLMConfig, **kwargs)

Bases: BaseLLM

Source code in evoagentx/models/base_model.py 
349
350
351
352
353
354
355
356
357
358
def __init__(self, config: LLMConfig, **kwargs):
    """Initializes the LLM with configuration.

    Args:
        config: Configuration object for the LLM.
        **kwargs (Any): Additional keyword arguments.
    """
    self.config = config
    self.kwargs = kwargs
    self.init_model()

get_stream_output 

get_stream_output(response: Stream, output_response: bool = True) -> str

Process stream response and return the complete output.

Parameters:

Name Type Description Default
response Stream

The stream response from OpenAI

 required
output_response bool

Whether to print the response in real-time

 True

Returns:

Name Type Description
str str

The complete output text
Source code in evoagentx/models/openai_model.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def get_stream_output(self, response: Stream, output_response: bool=True) -> str:
    """
    Process stream response and return the complete output.

    Args:
        response: The stream response from OpenAI
        output_response: Whether to print the response in real-time

    Returns:
        str: The complete output text
    """
    output = ""
    for chunk in response:
        content = chunk.choices[0].delta.content
        if content:
            if output_response:
                print(content, end="", flush=True)
            output += content
    if output_response:
        print("")
    return output

get_stream_output_async async 

get_stream_output_async(response, output_response: bool = False) -> str

Process async stream response and return the complete output.

Parameters:

Name Type Description Default
response AsyncIterator[ChatCompletionChunk]

The async stream response from OpenAI

 required
output_response bool

Whether to print the response in real-time

 False

Returns:

Name Type Description
str str

The complete output text
Source code in evoagentx/models/openai_model.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
async def get_stream_output_async(self, response, output_response: bool = False) -> str:
    """
    Process async stream response and return the complete output.

    Args:
        response (AsyncIterator[ChatCompletionChunk]): The async stream response from OpenAI
        output_response (bool): Whether to print the response in real-time


    Returns:
        str: The complete output text
    """
    output = ""
    async for chunk in response:
        content = chunk.choices[0].delta.content
        if content:
            if output_response:
                print(content, end="", flush=True)
            output += content
    if output_response:
        print("")
    return output

atomic_method 

atomic_method(func)

threading safe decorator for class methods. If there are self._lock in the instance, it will use the lock. Otherwise, use nullcontext for execution.

Source code in evoagentx/core/decorators.py 
23
24
25
26
27
28
29
30
31
32
33
def atomic_method(func):
    """
    threading safe decorator for class methods. 
    If there are self._lock in the instance, it will use the lock. Otherwise, use nullcontext for execution.
    """
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        context = getattr(self, "_lock", nullcontext())
        with context:
            return func(self, *args, **kwargs)
    return wrapper