import os
from typing import Dict, List
from qwen_agent.llm.schema import ASSISTANT, CONTENT, FUNCTION, NAME, ROLE, SYSTEM, USER
TOOL_CALL = '''
Start calling tool "{tool_name}"...
{tool_input}
'''
TOOL_OUTPUT = '''
Finished tool calling.
{tool_output}
'''
def get_avatar_image(name: str = 'user') -> str:
if name == 'user':
return os.path.join(os.path.dirname(__file__), 'assets/user.jpeg')
return os.path.join(os.path.dirname(__file__), 'assets/logo.jpeg')
def convert_history_to_chatbot(messages):
if not messages:
return None
chatbot_history = [[None, None]]
for message in messages:
if message.keys() != {'role', 'content'}:
raise ValueError('Each message must be a dict containing only "role" and "content".')
if message['role'] == USER:
chatbot_history[-1][0] = message['content']
elif message['role'] == ASSISTANT:
chatbot_history[-1][1] = message['content']
chatbot_history.append([None, None])
else:
raise ValueError(f'Message role must be {USER} or {ASSISTANT}.')
return chatbot_history
def convert_fncall_to_text(messages: List[Dict]) -> List[Dict]:
new_messages = []
for msg in messages:
role, content, name = msg[ROLE], msg[CONTENT], msg.get(NAME, None)
content = (content or '').lstrip('\n').rstrip()
# if role is system or user, just append the message
if role in (SYSTEM, USER):
new_messages.append({ROLE: role, CONTENT: content, NAME: name})
# if role is assistant, append the message and add function call details
elif role == ASSISTANT:
fn_call = msg.get(f'{FUNCTION}_call', {})
if fn_call:
f_name = fn_call['name']
f_args = fn_call['arguments']
content += TOOL_CALL.format(tool_name=f_name, tool_input=f_args)
if len(new_messages) > 0 and new_messages[-1][ROLE] == ASSISTANT and new_messages[-1][NAME] == name:
new_messages[-1][CONTENT] += content
else:
new_messages.append({ROLE: role, CONTENT: content, NAME: name})
# if role is function, append the message and add function result and exit details
elif role == FUNCTION:
assert new_messages[-1][ROLE] == ASSISTANT
new_messages[-1][CONTENT] += TOOL_OUTPUT.format(tool_output=content)
# if role is not system, user, assistant or function, raise TypeError
else:
raise TypeError
return new_messages