跳转至

🎯 动作接口

evoagentx.actions

Action 

Action(**kwargs)

Bases: BaseModule

Base class for all actions in the EvoAgentX framework.

Actions represent discrete operations that can be performed by agents. They define inputs, outputs, and execution behavior, and can optionally use tools to accomplish their tasks.

Attributes:

Name Type Description
name str

Unique identifier for the action.
description str

Human-readable description of what the action does.
prompt Optional[str]

Optional prompt template for this action.
tools Optional[List[Tool]]

Optional list of tools that can be used by this action.
inputs_format Optional[Type[ActionInput]]

Optional class defining the expected input structure.
outputs_format Optional[Type[Parser]]

Optional class defining the expected output structure.
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)

init_module 

init_module()

Initialize the action module.

This method is called after the action is instantiated. Subclasses can override this to perform custom initialization.

Source code in evoagentx/actions/action.py 
130
131
132
133
134
135
136
def init_module(self):
    """Initialize the action module.

    This method is called after the action is instantiated.
    Subclasses can override this to perform custom initialization.
    """
    pass

to_dict 

to_dict(exclude_none: bool = True, ignore: List[str] = [], **kwargs) -> dict

Convert the action to a dictionary for saving.

Source code in evoagentx/actions/action.py 
138
139
140
141
142
143
144
145
146
147
148
def to_dict(self, exclude_none: bool = True, ignore: List[str] = [], **kwargs) -> dict:
    """
    Convert the action to a dictionary for saving.  
    """
    data = super().to_dict(exclude_none=exclude_none, ignore=ignore, **kwargs)
    if self.inputs_format:
        data["inputs_format"] = self.inputs_format.__name__ 
    if self.outputs_format:
        data["outputs_format"] = self.outputs_format.__name__ 
    # TODO: customize serialization for the tools 
    return data

execute 

execute(llm: Optional[BaseLLM] = None, inputs: Optional[dict] = None, sys_msg: Optional[str] = None, return_prompt: bool = False, **kwargs) -> Optional[Union[Parser, Tuple[Parser, str]]]

Execute the action to produce a result.

This is the main entry point for executing an action. Subclasses must implement this method to define the action's behavior.

Parameters:

Name Type Description Default
llm Optional[BaseLLM]

The LLM used to execute the action.

 None
inputs Optional[dict]

Input data for the action execution. The input data should be a dictionary that matches the input format of the provided prompt. For example, if the prompt contains a variable {input_var}, the inputs dictionary should have a key input_var, otherwise the variable will be set to empty string.

None
sys_msg Optional[str]

Optional system message for the LLM.

 None
return_prompt bool

Whether to return the complete prompt passed to the LLM.

 False
**kwargs Any

Additional keyword arguments for the execution.

 {}

Returns:

Type Description
Optional[Union[Parser, Tuple[Parser, str]]]

If return_prompt is False, the method returns a Parser object containing the structured result of the action.
Optional[Union[Parser, Tuple[Parser, str]]]

If return_prompt is True, the method returns a tuple containing the Parser object and the complete prompt passed to the LLM.
Source code in evoagentx/actions/action.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def execute(self, llm: Optional[BaseLLM] = None, inputs: Optional[dict] = None, sys_msg: Optional[str]=None, return_prompt: bool = False, **kwargs) -> Optional[Union[Parser, Tuple[Parser, str]]]:
    """Execute the action to produce a result.

    This is the main entry point for executing an action. Subclasses must
    implement this method to define the action's behavior.

    Args:
        llm (Optional[BaseLLM]): The LLM used to execute the action.
        inputs (Optional[dict]): Input data for the action execution. The input data should be a dictionary that matches the input format of the provided prompt. 
            For example, if the prompt contains a variable `{input_var}`, the `inputs` dictionary should have a key `input_var`, otherwise the variable will be set to empty string. 
        sys_msg (Optional[str]): Optional system message for the LLM.
        return_prompt (bool): Whether to return the complete prompt passed to the LLM.
        **kwargs (Any): Additional keyword arguments for the execution.

    Returns:
        If `return_prompt` is False, the method returns a Parser object containing the structured result of the action.
        If `return_prompt` is True, the method returns a tuple containing the Parser object and the complete prompt passed to the LLM.
    """
    raise NotImplementedError(f"`execute` function of {type(self).__name__} is not implemented!")

async_execute async 

async_execute(llm: Optional[BaseLLM] = None, inputs: Optional[dict] = None, sys_msg: Optional[str] = None, return_prompt: bool = False, **kwargs) -> Optional[Union[Parser, Tuple[Parser, str]]]

Asynchronous execution of the action.

This method is the asynchronous counterpart of the execute method. It allows the action to be executed asynchronously using an LLM.

Source code in evoagentx/actions/action.py 
182
183
184
185
186
187
188
189
async def async_execute(self, llm: Optional[BaseLLM] = None, inputs: Optional[dict] = None, sys_msg: Optional[str]=None, return_prompt: bool = False, **kwargs) -> Optional[Union[Parser, Tuple[Parser, str]]]:
    """
    Asynchronous execution of the action.

    This method is the asynchronous counterpart of the `execute` method.
    It allows the action to be executed asynchronously using an LLM.
    """
    raise NotImplementedError(f"`async_execute` function of {type(self).__name__} is not implemented!")

ActionInput 

ActionInput(**kwargs)

Bases: LLMOutputParser

Input specification and parsing for actions.

This class defines the input requirements for actions and provides methods to generate structured input specifications. It inherits from LLMOutputParser to allow parsing of LLM outputs into structured inputs for actions.

Notes

Parameters in ActionInput should be defined in Pydantic Field format. For optional variables, use format: var: Optional[int] = Field(default=None, description="xxx") Remember to add default=None for optional parameters.

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_input_specification classmethod 

get_input_specification(ignore_fields: List[str] = []) -> str

Generate a JSON specification of the input requirements.

Examines the class fields and produces a structured specification of the input parameters, including their types, descriptions, and whether they are required.

Parameters:

Name Type Description Default
ignore_fields List[str]

List of field names to exclude from the specification.

 []

Returns:

Type Description
str

A JSON string containing the input specification, or an empty string
str

if no fields are defined or all are ignored.
Source code in evoagentx/actions/action.py 
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
@classmethod
def get_input_specification(cls, ignore_fields: List[str] = []) -> str:
    """Generate a JSON specification of the input requirements.

    Examines the class fields and produces a structured specification of
    the input parameters, including their types, descriptions, and whether
    they are required.

    Args:
        ignore_fields (List[str]): List of field names to exclude from the specification.

    Returns:
        A JSON string containing the input specification, or an empty string
        if no fields are defined or all are ignored.
    """
    fields_info = {}
    attrs = cls.get_attrs()
    for field_name, field_info in cls.model_fields.items():
        if field_name in ignore_fields:
            continue
        if field_name not in attrs:
            continue
        field_type = get_type_name(field_info.annotation)
        field_desc = field_info.description if field_info.description is not None else None
        # field_required = field_info.is_required()
        field_default = str(field_info.default) if field_info.default is not PydanticUndefined else None
        field_required = True if field_default is None else False
        description = field_type + ", "
        if field_desc is not None:
            description += (field_desc.strip() + ", ") 
        description += ("required" if field_required else "optional")
        if field_default is not None:
            description += (", Default value: " + field_default)
        fields_info[field_name] = description

    if len(fields_info) == 0:
        return "" 
    fields_info_str = json.dumps(fields_info, indent=4)
    return fields_info_str

get_required_input_names classmethod 

get_required_input_names() -> List[str]

Get a list of all required input parameter names.

Returns:

Type Description
List[str]

List[str]: Names of all parameters that are required (don't have default values).
Source code in evoagentx/actions/action.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@classmethod
def get_required_input_names(cls) -> List[str]:
    """Get a list of all required input parameter names.

    Returns:
        List[str]: Names of all parameters that are required (don't have default values).
    """
    required_fields = []
    attrs = cls.get_attrs()
    for field_name, field_info in cls.model_fields.items():
        if field_name not in attrs:
            continue
        field_default = field_info.default
        # A field is required if it doesn't have a default value
        if field_default is PydanticUndefined:
            required_fields.append(field_name)
    return required_fields

ActionOutput 

ActionOutput(**kwargs)

Bases: LLMOutputParser

Output representation for actions.

This class handles the structured output of actions, providing methods to convert the output to structured data. It inherits from LLMOutputParser to support parsing of LLM outputs into structured action results.

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)

to_str 

to_str() -> str

Convert the output to a formatted JSON string.

Returns:

Type Description
str

A pretty-printed JSON string representation of the structured data.
Source code in evoagentx/actions/action.py 
 98
 99
100
101
102
103
104
def to_str(self) -> str:
    """Convert the output to a formatted JSON string.

    Returns:
        A pretty-printed JSON string representation of the structured data.
    """
    return json.dumps(self.get_structured_data(), indent=4)

CodeExtraction 

CodeExtraction(**kwargs)

Bases: Action

An action that extracts and organizes code blocks from text.

This action uses an LLM to analyze text containing code blocks, extract them, suggest appropriate filenames, and save them to a specified directory. It can also identify which file is likely the main entry point based on heuristics.

Attributes:

Name Type Description
name str

The name of the action.
description str

A description of what the action does.
prompt Optional[str]

The prompt template used by the action.
inputs_format Optional[Type[ActionInput]]

The expected format of inputs to this action.
outputs_format Optional[Type[Parser]]

The format of the action's output.
Source code in evoagentx/actions/code_extraction.py 
60
61
62
63
64
65
66
67
68
69
def __init__(self, **kwargs):

    name = kwargs.pop("name") if "name" in kwargs else CODE_EXTRACTION["name"]
    description = kwargs.pop("description") if "description" in kwargs else CODE_EXTRACTION["description"]
    prompt = kwargs.pop("prompt") if "prompt" in kwargs else CODE_EXTRACTION["prompt"]
    # inputs_format = kwargs.pop("inputs_format") if "inputs_format" in kwargs else CodeExtractionInput
    # outputs_format = kwargs.pop("outputs_format") if "outputs_format" in kwargs else CodeExtractionOutput
    inputs_format = kwargs.pop("inputs_format", None) or CodeExtractionInput
    outputs_format = kwargs.pop("outputs_format", None) or CodeExtractionOutput
    super().__init__(name=name, description=description, prompt=prompt, inputs_format=inputs_format, outputs_format=outputs_format, **kwargs)

identify_main_file 

identify_main_file(saved_files: Dict[str, str]) -> Optional[str]

Identify the main file from the saved files based on content and file type.

This method uses a combination of common filename conventions and content analysis to determine which file is likely the main entry point of a project.

Parameters:

Name Type Description Default
saved_files Dict[str, str]

Dictionary mapping filenames to their full paths

 required

Returns:

Type Description
Optional[str]

Path to the main file if found, None otherwise
Source code in evoagentx/actions/code_extraction.py 
 71
 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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def identify_main_file(self, saved_files: Dict[str, str]) -> Optional[str]:
    """Identify the main file from the saved files based on content and file type.

    This method uses a combination of common filename conventions and content
    analysis to determine which file is likely the main entry point of a project.

    Args:
        saved_files: Dictionary mapping filenames to their full paths

    Returns:
        Path to the main file if found, None otherwise

    """
    # Priority lookup for common main files by language
    main_file_priorities = [
        # HTML files
        "index.html",
        # Python files
        "main.py", 
        "app.py",
        # JavaScript files
        "index.js",
        "main.js",
        "app.js",
        # Java files
        "Main.java",
        # C/C++ files
        "main.cpp", 
        "main.c",
        # Go files
        "main.go",
        # Other common entry points
        "index.php",
        "Program.cs"
    ]

    # First check priority list
    for main_file in main_file_priorities:
        if main_file in saved_files:
            return saved_files[main_file]

    # If no priority file found, use heuristics based on file extensions

    # If we have HTML files, use the first one
    html_files = {k: v for k, v in saved_files.items() if k.endswith('.html')}
    if html_files:
        return next(iter(html_files.values()))

    # Check for Python files with "__main__" section
    py_files = {k: v for k, v in saved_files.items() if k.endswith('.py')}
    if py_files:
        for filename, path in py_files.items():
            with open(path, 'r', encoding='utf-8') as f:
                content = f.read()
                if "if __name__ == '__main__'" in content or 'if __name__ == "__main__"' in content:
                    return path
        # If no main found, return the first Python file
        if py_files:
            return next(iter(py_files.values()))

    # If we have Java files, look for one with a main method
    java_files = {k: v for k, v in saved_files.items() if k.endswith('.java')}
    if java_files:
        for filename, path in java_files.items():
            with open(path, 'r', encoding='utf-8') as f:
                content = f.read()
                if "public static void main" in content:
                    return path
        # If no main found, return the first Java file
        if java_files:
            return next(iter(java_files.values()))

    # For JavaScript applications
    js_files = {k: v for k, v in saved_files.items() if k.endswith('.js')}
    if js_files:
        return next(iter(js_files.values()))

    # If all else fails, return the first file
    if saved_files:
        return next(iter(saved_files.values()))

    # No files found
    return None

save_code_blocks 

save_code_blocks(code_blocks: List[Dict], target_directory: str) -> Dict[str, str]

Save code blocks to files in the target directory.

Creates the target directory if it doesn't exist and saves each code block to a file with an appropriate name, handling filename conflicts.

Parameters:

Name Type Description Default
code_blocks List[Dict]

List of dictionaries containing code block information

 required
target_directory str

Directory path where files should be saved

 required

Returns:

Type Description
Dict[str, str]

Dictionary mapping filenames to their full paths
Source code in evoagentx/actions/code_extraction.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
def save_code_blocks(self, code_blocks: List[Dict], target_directory: str) -> Dict[str, str]:
    """Save code blocks to files in the target directory.

    Creates the target directory if it doesn't exist and saves each code block
    to a file with an appropriate name, handling filename conflicts.

    Args:
        code_blocks: List of dictionaries containing code block information
        target_directory: Directory path where files should be saved

    Returns:
        Dictionary mapping filenames to their full paths
    """
    os.makedirs(target_directory, exist_ok=True)
    saved_files = {}

    for block in code_blocks:
        filename = block.get("filename", "unknown.txt")
        content = block.get("content", "")

        # Skip empty blocks
        if not content.strip():
            continue

        # Handle filename conflicts
        base_filename = filename
        counter = 1
        while filename in saved_files:
            name_parts = base_filename.split('.')
            if len(name_parts) > 1:
                filename = f"{'.'.join(name_parts[:-1])}_{counter}.{name_parts[-1]}"
            else:
                filename = f"{base_filename}_{counter}"
            counter += 1

        # Save to file
        file_path = os.path.join(target_directory, filename)
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(content)

        # Add to map
        saved_files[filename] = file_path

    return saved_files

execute 

execute(llm: Optional[BaseLLM] = None, inputs: Optional[dict] = None, sys_msg: Optional[str] = None, return_prompt: bool = False, **kwargs) -> CodeExtractionOutput

Execute the CodeExtraction action.

Extracts code blocks from the provided text using the specified LLM, saves them to the target directory, and identifies the main file.

Parameters:

Name Type Description Default
llm Optional[BaseLLM]

The LLM to use for code extraction

 None
inputs Optional[dict]

Dictionary containing: - code_string: The string with code blocks to extract - target_directory: Where to save the files - project_name: Optional project folder name

 None
sys_msg Optional[str]

Optional system message override for the LLM

 None
return_prompt bool

Whether to return the prompt along with the result

 False
**kwargs Any

Additional keyword arguments

 {}

Returns:

Type Description
CodeExtractionOutput

CodeExtractionOutput with extracted file information
Source code in evoagentx/actions/code_extraction.py 
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
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def execute(self, llm: Optional[BaseLLM] = None, inputs: Optional[dict] = None, sys_msg: Optional[str]=None, return_prompt: bool = False, **kwargs) -> CodeExtractionOutput:
    """Execute the CodeExtraction action.

    Extracts code blocks from the provided text using the specified LLM,
    saves them to the target directory, and identifies the main file.

    Args:
        llm: The LLM to use for code extraction
        inputs: Dictionary containing:
            - code_string: The string with code blocks to extract
            - target_directory: Where to save the files
            - project_name: Optional project folder name
        sys_msg: Optional system message override for the LLM
        return_prompt: Whether to return the prompt along with the result
        **kwargs (Any): Additional keyword arguments

    Returns:
        CodeExtractionOutput with extracted file information
    """
    if not llm:
        error_msg = "CodeExtraction action requires an LLM."
        return CodeExtractionOutput(extracted_files={}, error=error_msg)

    if not inputs:
        error_msg = "CodeExtraction action received invalid `inputs`: None or empty."
        return CodeExtractionOutput(extracted_files={}, error=error_msg)

    code_string = inputs.get("code_string", "")
    target_directory = inputs.get("target_directory", "")
    project_name = inputs.get("project_name", None)

    if not code_string:
        error_msg = "No code string provided."
        return CodeExtractionOutput(extracted_files={}, error=error_msg)

    if not target_directory:
        error_msg = "No target directory provided."
        return CodeExtractionOutput(extracted_files={}, error=error_msg)

    # Create project folder if name is provided
    if project_name:
        project_dir = os.path.join(target_directory, project_name)
    else:
        project_dir = target_directory

    try:
        # Use LLM to extract code blocks and suggest filenames
        prompt_params = {"code_string": code_string}
        system_message = CODE_EXTRACTION["system_prompt"] if sys_msg is None else sys_msg

        llm_response: CodeBlockList = llm.generate(
            prompt=self.prompt.format(**prompt_params),
            system_message=system_message,
            parser=CodeBlockList,
            parse_mode="json"
        )
        code_blocks = llm_response.get_structured_data().get("code_blocks", [])

        # Save code blocks to files
        saved_files = self.save_code_blocks(code_blocks, project_dir)

        # Identify main file
        main_file = self.identify_main_file(saved_files)

        result = CodeExtractionOutput(
            extracted_files=saved_files,
            main_file=main_file
        )

        if return_prompt:
            return result, self.prompt.format(**prompt_params)

        return result

    except Exception as e:
        error_msg = f"Error extracting code: {str(e)}"
        return CodeExtractionOutput(extracted_files={}, error=error_msg)