(self, message: str)
| 208 | 交替出现.""" |
| 209 | |
| 210 | def chat(self, message: str) -> AgentReturn: |
| 211 | for hist in self._session_history: |
| 212 | if hist['role'] == 'system': |
| 213 | hist['role'] = self.system_role |
| 214 | self._inner_history = [] |
| 215 | # append the user message for session history |
| 216 | self._session_history.append(dict(role='user', content=message)) |
| 217 | agent_return = AgentReturn() |
| 218 | force_stop = False |
| 219 | default_response = '对不起，我无法回答你的问题' |
| 220 | for turn in range(self.max_turn): |
| 221 | prompt = self._protocol.format( |
| 222 | chat_history=self.session_history, |
| 223 | inner_step=self._inner_history, |
| 224 | action_executor=self._action_executor, |
| 225 | force_stop=force_stop) |
| 226 | prompt = self.merge_role(prompt) |
| 227 | response = self._llm.generate_from_template(prompt, 512) |
| 228 | self._inner_history.append(dict(role='assistant', |
| 229 | content=response)) |
| 230 | thought, action, action_input = self._protocol.parse( |
| 231 | response, self._action_executor) |
| 232 | action_return: ActionReturn = self._action_executor( |
| 233 | action, action_input) |
| 234 | action_return.thought = thought |
| 235 | agent_return.actions.append(action_return) |
| 236 | if action_return.state == ActionStatusCode.SUCCESS: |
| 237 | # if success, stash model response and system response |
| 238 | self._session_history.append( |
| 239 | dict(role='assistant', content=response)) |
| 240 | self._session_history.append( |
| 241 | dict( |
| 242 | role=self.system_role, |
| 243 | content=self._protocol.format_response(action_return))) |
| 244 | agent_return.response = action_return.result['text'] |
| 245 | return agent_return |
| 246 | elif action_return.type == self._action_executor.invalid_action.name: # noqa |
| 247 | action_return.errmsg = 'The action is invalid, please check the action name.' # noqa |
| 248 | self._inner_history.append( |
| 249 | dict(role=self.system_role, |
| 250 | content=self._protocol.format_response(action_return))) |
| 251 | if turn == self.max_turn - 1: |
| 252 | force_stop = True |
| 253 | agent_return.response = default_response |
| 254 | self._session_history.append( |
| 255 | dict(role='assistant', content=agent_return.response)) |
| 256 | return agent_return |
| 257 | |
| 258 | def merge_role(self, inputs): |
| 259 | messages = [] |
nothing calls this directly
no test coverage detected