in maga_transformer/openai/renderers/qwen_agent/llm/function_calling.py [0:0]
def _postprocess_fncall_messages(self, messages: List[Message]) -> List[Message]:
"""
If the model calls function by built-in function call template,
convert and display it in function_call format.
"""
# Remove ': ' brought by continued generation of function calling
last_msg = messages[-1].content
for i in range(len(last_msg)):
item_type, item_text = last_msg[i].get_type_and_value()
if item_type == 'text':
if item_text.startswith(': '):
last_msg[i].text = item_text[2:]
elif item_text.startswith(':'):
last_msg[i].text = item_text[1:]
break
new_messages = []
for msg in messages:
role, content = msg.role, msg.content
assert isinstance(content, list)
if role in (SYSTEM, USER):
new_messages.append(Message(role=role, content=content))
continue
new_content = []
for item in content:
item_type, item_text = item.get_type_and_value()
if item_type != 'text': # multimodal
new_content.append(item)
continue
for stop_word in [FN_RESULT, FN_EXIT]:
assert stop_word in FN_STOP_WORDS
assert stop_word not in item_text, 'Something wrong, stop words are expected to be excluded.'
i = item_text.find(f'{FN_NAME}:')
# If no function call:
if i < 0:
show_text = remove_incomplete_special_tokens(item_text)
if show_text:
new_content.append(ContentItem(text=show_text))
continue
# If it says something before function call:
if i > 0:
answer = item_text[:i].lstrip('\n').rstrip()
if answer.endswith('\n'):
answer = answer[:-1]
show_text = remove_incomplete_special_tokens(answer)
if show_text:
new_content.append(ContentItem(text=show_text))
if new_content:
new_messages.append(Message(
role=role,
content=new_content,
)) # split thought and function call
new_content = []
item_text = item_text[i:]
# If has function call:
for part in item_text.split(f'{FN_NAME}:'):
if not part:
continue
if part.endswith('\n'):
part = part[:-1]
arg_sep = f'\n{FN_ARGS}:'
i = part.find(arg_sep)
if i < 0:
fn_name = part.strip()
list_of_fn_args = ['']
else:
fn_name = part[:i].strip()
list_of_fn_args = [_.strip() for _ in part[i + len(arg_sep):].split(arg_sep)]
fn_name = remove_incomplete_special_tokens(fn_name)
for fn_args in list_of_fn_args:
fn_args = remove_incomplete_special_tokens(fn_args)
fn_args = remove_trailing_comment_of_fn_args(fn_args)
new_messages.append(
Message(
role=ASSISTANT,
content=[],
function_call=FunctionCall(
name=fn_name,
arguments=fn_args,
),
))
# Break here and discard the text after function call
return new_messages
if new_content:
new_messages.append(Message(role=role, content=new_content))
return new_messages