A code class can be executed in a `CodeAgent` in a workflow
| 8 | |
| 9 | |
| 10 | class CodeAgent(Agent): |
| 11 | """A code class can be executed in a `CodeAgent` in a workflow""" |
| 12 | |
| 13 | AGENT_NAME = 'CodeAgent' |
| 14 | |
| 15 | def __init__(self, |
| 16 | config: DictConfig, |
| 17 | tag: str, |
| 18 | trust_remote_code: bool = False, |
| 19 | **kwargs): |
| 20 | super().__init__(config, tag, trust_remote_code, **kwargs) |
| 21 | self.load_cache = kwargs.get('load_cache', False) |
| 22 | |
| 23 | async def run(self, inputs: Union[str, List[Message]], |
| 24 | **kwargs) -> List[Message]: |
| 25 | """Run the external code. Default implementation here does nothing. |
| 26 | |
| 27 | Args: |
| 28 | inputs(`Union[str, List[Message]]`): The inputs can be a prompt string, |
| 29 | or a list of messages from the previous agent |
| 30 | |
| 31 | Returns: |
| 32 | The messages to output to the next agent |
| 33 | """ |
| 34 | _config = None |
| 35 | _messages = None |
| 36 | if self.load_cache: |
| 37 | _config, _messages = self.read_history(inputs) |
| 38 | if _config is not None and _messages is not None: |
| 39 | self.config = _config |
| 40 | return _messages |
| 41 | messages = await self.execute_code(inputs, **kwargs) |
| 42 | self.save_history(messages, **kwargs) |
| 43 | return messages |
| 44 | |
| 45 | async def execute_code(self, inputs: Union[str, List[Message]], |
| 46 | **kwargs) -> List[Message]: |
| 47 | return inputs |