| | from copy import deepcopy |
| |
|
| | from lagent.schema import ActionReturn, ActionStatusCode, FunctionCall |
| | from .hook import Hook |
| |
|
| |
|
| | class ActionPreprocessor(Hook): |
| | """The ActionPreprocessor is a hook that preprocesses the action message |
| | and postprocesses the action return message. |
| | |
| | """ |
| |
|
| | def before_action(self, executor, message, session_id): |
| | assert isinstance(message.formatted, FunctionCall) or ( |
| | isinstance(message.formatted, dict) and 'name' in message.content |
| | and 'parameters' in message.formatted) or ( |
| | 'action' in message.formatted |
| | and 'parameters' in message.formatted['action'] |
| | and 'name' in message.formatted['action']) |
| | if isinstance(message.formatted, dict): |
| | name = message.formatted.get('name', |
| | message.formatted['action']['name']) |
| | parameters = message.formatted.get( |
| | 'parameters', message.formatted['action']['parameters']) |
| | else: |
| | name = message.formatted.name |
| | parameters = message.formatted.parameters |
| | message.content = dict(name=name, parameters=parameters) |
| | return message |
| |
|
| | def after_action(self, executor, message, session_id): |
| | action_return = message.content |
| | if isinstance(action_return, ActionReturn): |
| | if action_return.state == ActionStatusCode.SUCCESS: |
| | response = action_return.format_result() |
| | else: |
| | response = action_return.errmsg |
| | else: |
| | response = action_return |
| | message.content = response |
| | return message |
| |
|
| |
|
| | class InternLMActionProcessor(ActionPreprocessor): |
| |
|
| | def __init__(self, code_parameter: str = 'command'): |
| | self.code_parameter = code_parameter |
| |
|
| | def before_action(self, executor, message, session_id): |
| | message = deepcopy(message) |
| | assert isinstance(message.formatted, dict) and set( |
| | message.formatted).issuperset( |
| | {'tool_type', 'thought', 'action', 'status'}) |
| | if isinstance(message.formatted['action'], str): |
| | |
| | action_name = next(iter(executor.actions)) |
| | parameters = {self.code_parameter: message.formatted['action']} |
| | if action_name in ['AsyncIPythonInterpreter']: |
| | parameters['session_id'] = session_id |
| | message.formatted['action'] = dict( |
| | name=action_name, parameters=parameters) |
| | return super().before_action(executor, message, session_id) |
| |
|
