utils.py 2.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879
  1. import os
  2. from typing import Dict, List
  3. from qwen_agent.llm.schema import ASSISTANT, CONTENT, FUNCTION, NAME, ROLE, SYSTEM, USER
  4. TOOL_CALL = '''
  5. <details>
  6. <summary>Start calling tool "{tool_name}"...</summary>
  7. {tool_input}
  8. </details>
  9. '''
  10. TOOL_OUTPUT = '''
  11. <details>
  12. <summary>Finished tool calling.</summary>
  13. {tool_output}
  14. </details>
  15. '''
  16. def get_avatar_image(name: str = 'user') -> str:
  17. if name == 'user':
  18. return os.path.join(os.path.dirname(__file__), 'assets/user.jpeg')
  19. return os.path.join(os.path.dirname(__file__), 'assets/logo.jpeg')
  20. def convert_history_to_chatbot(messages):
  21. if not messages:
  22. return None
  23. chatbot_history = [[None, None]]
  24. for message in messages:
  25. if message.keys() != {'role', 'content'}:
  26. raise ValueError('Each message must be a dict containing only "role" and "content".')
  27. if message['role'] == USER:
  28. chatbot_history[-1][0] = message['content']
  29. elif message['role'] == ASSISTANT:
  30. chatbot_history[-1][1] = message['content']
  31. chatbot_history.append([None, None])
  32. else:
  33. raise ValueError(f'Message role must be {USER} or {ASSISTANT}.')
  34. return chatbot_history
  35. def convert_fncall_to_text(messages: List[Dict]) -> List[Dict]:
  36. new_messages = []
  37. for msg in messages:
  38. role, content, name = msg[ROLE], msg[CONTENT], msg.get(NAME, None)
  39. content = (content or '').lstrip('\n').rstrip()
  40. # if role is system or user, just append the message
  41. if role in (SYSTEM, USER):
  42. new_messages.append({ROLE: role, CONTENT: content, NAME: name})
  43. # if role is assistant, append the message and add function call details
  44. elif role == ASSISTANT:
  45. fn_call = msg.get(f'{FUNCTION}_call', {})
  46. if fn_call:
  47. f_name = fn_call['name']
  48. f_args = fn_call['arguments']
  49. content += TOOL_CALL.format(tool_name=f_name, tool_input=f_args)
  50. if len(new_messages) > 0 and new_messages[-1][ROLE] == ASSISTANT and new_messages[-1][NAME] == name:
  51. new_messages[-1][CONTENT] += content
  52. else:
  53. new_messages.append({ROLE: role, CONTENT: content, NAME: name})
  54. # if role is function, append the message and add function result and exit details
  55. elif role == FUNCTION:
  56. assert new_messages[-1][ROLE] == ASSISTANT
  57. new_messages[-1][CONTENT] += TOOL_OUTPUT.format(tool_output=content)
  58. # if role is not system, user, assistant or function, raise TypeError
  59. else:
  60. raise TypeError
  61. return new_messages