针对Qwen-Agent框架的源码阅读与解析：FnCallAgent与ReActChat篇

在《针对Qwen-Agent框架的Function Call及ReAct的源码阅读与解析：Agent基类篇》中，我们已经了解了Agent基类的大体实现。这里我们就再详细学习一下FnCallAgent类和ReActChat的实现思路，从而对Agent的两条主流技术路径有更深刻的了解。同时，在前面的文章《基于Qwen-Agent框架的Function Call及ReAct方式调用自定义工具》中我们尝试了中英文提示词和并行调用工具，这里我们将追溯源码，看看它们的深层次实现以及失效的原因。

---

🎉进入大模型应用与实战专栏 | 🚀查看更多专栏内容

---

FnCallAgent类

在Agent类的基础上，qwen_agent/agents/fncall_agent.py下的FnCallAgent类又重写了三个函数，接下来我们就分别来看一下它们的源码实现，并做对应的分析。

初始化函数

    def __init__(self,function_list: Optional[List[Union[str, Dict, BaseTool]]] = None,llm: Optional[Union[Dict, BaseChatModel]] = None,system_message: Optional[str] = DEFAULT_SYSTEM_MESSAGE,name: Optional[str] = None,description: Optional[str] = None,files: Optional[List[str]] = None,**kwargs):"""Initialization the agent.Args:function_list: One list of tool name, tool configuration or Tool object,such as 'code_interpreter', {'name': 'code_interpreter', 'timeout': 10}, or CodeInterpreter().llm: The LLM model configuration or LLM model object.Set the configuration as {'model': '', 'api_key': '', 'model_server': ''}.system_message: The specified system message for LLM chat.name: The name of this agent.description: The description of this agent, which will be used for multi_agent.files: A file url list. The initialized files for the agent."""super().__init__(function_list=function_list,llm=llm,system_message=system_message,name=name,description=description)if not hasattr(self, 'mem'):# Default to use Memory to manage filesself.mem = Memory(llm=self.llm, files=files, **kwargs)

其实很简单，就是添加了Memory，用来完成文件相关的操作。

执行函数

    def _run(self, messages: List[Message], lang: Literal['en', 'zh'] = 'en', **kwargs) -> Iterator[List[Message]]:messages = copy.deepcopy(messages)num_llm_calls_available = MAX_LLM_CALL_PER_RUNresponse = []while True and num_llm_calls_available > 0:num_llm_calls_available -= 1extra_generate_cfg = {'lang': lang}if kwargs.get('seed') is not None:extra_generate_cfg['seed'] = kwargs['seed']output_stream = self._call_llm(messages=messages,functions=[func.function for func in self.function_map.values()],extra_generate_cfg=extra_generate_cfg)output: List[Message] = []for output in output_stream:if output:yield response + outputif output:response.extend(output)messages.extend(output)used_any_tool = Falsefor out in output:use_tool, tool_name, tool_args, _ = self._detect_tool(out)if use_tool:tool_result = self._call_tool(tool_name, tool_args, messages=messages, **kwargs)fn_msg = Message(role=FUNCTION,name=tool_name,content=tool_result,)messages.append(fn_msg)response.append(fn_msg)yield responseused_any_tool = Trueif not used_any_tool:breakyield response

这是一个大模型推理和工具执行的核心方法，主要功能和特点如下：

1. 输入与初始化

• 接收消息列表和语言参数
• 深拷贝输入消息
• 设置最大LLM调用次数限制（默认是8）

2. 推理流程

• 循环调用语言模型
• 通过_call_llm获取模型输出
• 逐步生成和扩展响应消息链

3. 工具调用机制

• 检测模型输出中是否需要调用工具
• 使用_call_tool执行工具
• 将工具调用结果作为函数消息追加到消息链

4. 特殊设计

• 支持设置随机数种子
• 可配置语言(中/英)
• 实时流式输出响应
• 通过yield返回中间结果

5. 终止条件

• LLM调用次数耗尽
• 没有工具被调用时退出

核心逻辑：不断调用LLM，检测并执行工具，直到达到最大调用次数或无需调用工具。

工具调用

在上面的执行过程中，需要调用_call_tool方法完成工具的调用，其实该方法在基类中已有定义，代码中的super()._call_tool对应基类的方法。这里之所以要重写，主要是为了对file_access标记进行判断：

• 如果工具需要文件访问，则从消息和系统文件中提取文件
• 否则直接调用原始工具调用方法

这是一种针对特定工具需求的定制处理方案。

    def _call_tool(self, tool_name: str, tool_args: Union[str, dict] = '{}', **kwargs) -> str:if tool_name not in self.function_map:return f'Tool {tool_name} does not exists.'# Temporary plan: Check if it is necessary to transfer files to the tool# Todo: This should be changed to parameter passing, and the file URL should be determined by the modelif self.function_map[tool_name].file_access:assert 'messages' in kwargsfiles = extract_files_from_messages(kwargs['messages'], include_images=True) + self.mem.system_filesreturn super()._call_tool(tool_name, tool_args, files=files, **kwargs)else:return super()._call_tool(tool_name, tool_args, **kwargs)

关于中英文提示词的分析

在上面的代码中，可以通过修改extra_generate_cfg = {'lang': lang}来设置模型推理时使用的语言，我们接下来就看看随着设置不同的语言类型，模型所采用的提示词会有哪些差异。为了更好地分析这一点，我们需要分别从模型初始化和模型推理这两个角度出发，查看并厘清FnCallAgent类的调用路径。

模型初始化路径

qwen_agent/agents/fncall_agent.py路径下FnCallAgent的 __init__函数→【执行】→super().__init__
qwen_agent/agent.py路径下Agent的 __init__函数→【执行】→self.llm = get_chat_model(llm)
qwen_agent/llm/__init__.py路径下的get_chat_model函数→【执行】→return LLM_REGISTRY[model_type](cfg)

这里还做了下面的判断，以openai的形式调用模型

	if 'model_server' in cfg:if cfg['model_server'].strip().startswith('http'):model_type = 'oai'return LLM_REGISTRY[model_type](cfg)

而LLM_REGISTRY是一个字典，用于记录不同类型模型的初始化函数，其内容如下所示

所以接下来的调用路径是这样的：
qwen_agent/llm/oai.py路径下TextChatAtOAI的 __init__函数→【执行】→super().__init__

又由于TextChatAtOAI继承自BaseFnCallModel，所以最终的初始化函数位于qwen_agent/llm/function_calling.py下的BaseFnCallModel，相关代码如下：

class BaseFnCallModel(BaseChatModel, ABC):def __init__(self, cfg: Optional[Dict] = None):super().__init__(cfg)fncall_prompt_type = self.generate_cfg.get('fncall_prompt_type', 'qwen')if fncall_prompt_type == 'qwen':from qwen_agent.llm.fncall_prompts.qwen_fncall_prompt import FN_STOP_WORDS, QwenFnCallPromptself.fncall_prompt = QwenFnCallPrompt()stop = self.generate_cfg.get('stop', [])self.generate_cfg['stop'] = stop + [x for x in FN_STOP_WORDS if x not in stop]else:raise NotImplementedError

这里的提示词self.fncall_prompt调用了QwenFnCallPrompt()，我们将在推理阶段分析里面的中英文提示设置。

模型推理路径

在推理阶段，首先进入qwen_agent/agents/fncall_agent.py路径下FnCallAgent的 _run函数→【执行下面代码】

output_stream = self._call_llm(messages=messages,functions=[func.function for func in self.function_map.values()],extra_generate_cfg=extra_generate_cfg)

接下来进入qwen_agent/agent.py路径下Agent的 _call_llm函数→【执行下面代码】

return self.llm.chat(messages=messages,functions=functions,stream=stream,extra_generate_cfg=merge_generate_cfgs(base_generate_cfg=self.extra_generate_cfg,new_generate_cfg=extra_generate_cfg,))

接下来进入qwen_agent/llm/base.py路径下BaseChatModel的 chat函数→【执行下面代码】

messages = self._preprocess_messages(messages, lang=lang, generate_cfg=generate_cfg, functions=functions)

最终进入qwen_agent/llm/function_calling.py下的BaseFnCallModel，执行_preprocess_messages

def _preprocess_messages(self,messages: List[Message],lang: Literal['en', 'zh'],generate_cfg: dict,functions: Optional[List[Dict]] = None,) -> List[Message]:messages = super()._preprocess_messages(messages, lang=lang, generate_cfg=generate_cfg)if (not functions) or (generate_cfg.get('function_choice', 'auto') == 'none'):messages = self._remove_fncall_messages(messages, lang=lang)else:validate_num_fncall_results(messages=messages,support_multimodal_input=self.support_multimodal_input,)messages = self.fncall_prompt.preprocess_fncall_messages(messages=messages,functions=functions,lang=lang,parallel_function_calls=generate_cfg.get('parallel_function_calls', False),function_choice=generate_cfg.get('function_choice', 'auto'),)return messages

其具体实现位于qwen_agent/llm/fncall_prompts/qwen_fncall_prompt.py下的QwenFnCallPrompt中的preprocess_fncall_messages函数

    def preprocess_fncall_messages(messages: List[Message],functions: List[dict],lang: Literal['en', 'zh'],parallel_function_calls: bool = True,function_choice: Union[Literal['auto'], str] = 'auto',) -> List[Message]:ori_messages = messages# Change function_call responses to plaintext responses:messages = []for msg in copy.deepcopy(ori_messages):role, content = msg.role, msg.contentif role in (SYSTEM, USER):messages.append(msg)elif role == ASSISTANT:content = (content or [])fn_call = msg.function_callif fn_call:f_name = fn_call.namef_args = fn_call.argumentsif f_args.startswith('```'):  # if code snippetf_args = '\n' + f_args  # for markdown renderingfunc_content = '\n' if messages[-1].role == ASSISTANT else ''func_content += f'{FN_NAME}: {f_name}'func_content += f'\n{FN_ARGS}: {f_args}'content.append(ContentItem(text=func_content))if messages[-1].role == ASSISTANT:messages[-1].content += contentelse:messages.append(Message(role=role, content=content))elif role == FUNCTION:assert messages[-1].role == ASSISTANTassert isinstance(content, list)assert all(isinstance(item, ContentItem) for item in content)if content:f_result = copy.deepcopy(content)else:f_result = [ContentItem(text='')]f_exit = f'\n{FN_EXIT}: 'last_text_content = messages[-1].content[-1].textif last_text_content.endswith(f_exit):messages[-1].content[-1].text = last_text_content[:-len(f_exit)]f_result = [ContentItem(text=f'\n{FN_RESULT}: ')] + f_result + [ContentItem(text=f_exit)]messages[-1].content += f_resultelse:raise TypeError# Add a system prompt for function calling:tool_desc_template = FN_CALL_TEMPLATE[lang + ('_parallel' if parallel_function_calls else '')]tool_descs = '\n\n'.join(get_function_description(function, lang=lang) for function in functions)tool_names = ','.join(function.get('name_for_model', function.get('name', '')) for function in functions)tool_system = tool_desc_template.format(tool_descs=tool_descs, tool_names=tool_names)if messages[0].role == SYSTEM:messages[0].content.append(ContentItem(text='\n\n' + tool_system))else:messages = [Message(role=SYSTEM, content=[ContentItem(text=tool_system)])] + messages# Remove ': ' for continued generation of function calling,# because ': ' may form a single token with its following words:if messages[-1].role == ASSISTANT:last_msg = messages[-1].contentfor i in range(len(last_msg) - 1, -1, -1):item_type, item_text = last_msg[i].get_type_and_value()if item_type == 'text':if item_text.endswith(f'{FN_EXIT}: '):last_msg[i].text = item_text[:-2]break# Add the function_choice prefix:if function_choice not in ('auto', 'none'):if messages[-1].role == ASSISTANT:last_msg = messages[-1]if last_msg.content:if extract_text_from_message(last_msg, add_upload_info=False).endswith(FN_EXIT):last_msg.content.append(ContentItem(text=': \n'))else:last_msg.content.append(ContentItem(text='\n'))messages = messages[:-1]else:last_msg = Message(role=ASSISTANT, content=[])last_msg.content.append(ContentItem(text=f'{FN_NAME}: {function_choice}'))messages = messages + [last_msg]return messages

我们可以看到，与提示词直接相关的是这句话：

tool_desc_template = FN_CALL_TEMPLATE[lang + ('_parallel' if parallel_function_calls else '')]

而FN_CALL_TEMPLATE定义了以下四种提示词，包括了中英文及它们各自的并行版本

FN_CALL_TEMPLATE = {'zh': FN_CALL_TEMPLATE_INFO_ZH + '\n\n' + FN_CALL_TEMPLATE_FMT_ZH,'en': FN_CALL_TEMPLATE_INFO_EN + '\n\n' + FN_CALL_TEMPLATE_FMT_EN,'zh_parallel': FN_CALL_TEMPLATE_INFO_ZH + '\n\n' + FN_CALL_TEMPLATE_FMT_PARA_ZH,'en_parallel': FN_CALL_TEMPLATE_INFO_EN + '\n\n' + FN_CALL_TEMPLATE_FMT_PARA_EN,
}

列表记录一下它们的具体内容

• 中文

# 工具## 你拥有如下工具：{tool_descs}## 你可以在回复中插入零次、一次或多次以下命令以调用工具：✿FUNCTION✿: 工具名称，必须是[{tool_names}]之一。
✿ARGS✿: 工具输入
✿RESULT✿: 工具结果
✿RETURN✿: 根据工具结果进行回复，需将图片用![](url)渲染出来

• 英文

# Tools## You have access to the following tools:{tool_descs}## When you need to call a tool, please insert the following command in your reply, which can be called zero or multiple times according to your needs:✿FUNCTION✿: The tool to use, should be one of [{tool_names}]
✿ARGS✿: The input of the tool
✿RESULT✿: Tool results
✿RETURN✿: Reply based on tool results. Images need to be rendered as ![](url)

• 中文并行

# 工具## 你拥有如下工具：{tool_descs}## 你可以在回复中插入以下命令以并行调用N个工具：✿FUNCTION✿: 工具1的名称，必须是[{tool_names}]之一
✿ARGS✿: 工具1的输入
✿FUNCTION✿: 工具2的名称
✿ARGS✿: 工具2的输入
...
✿FUNCTION✿: 工具N的名称
✿ARGS✿: 工具N的输入
✿RESULT✿: 工具1的结果
✿RESULT✿: 工具2的结果
...
✿RESULT✿: 工具N的结果
✿RETURN✿: 根据工具结果进行回复，需将图片用![](url)渲染出来

• 英文并行

# Tools## You have access to the following tools:{tool_descs}## Insert the following command in your reply when you need to call N tools in parallel:✿FUNCTION✿: The name of tool 1, should be one of [{tool_names}]
✿ARGS✿: The input of tool 1
✿FUNCTION✿: The name of tool 2
✿ARGS✿: The input of tool 2
...
✿FUNCTION✿: The name of tool N
✿ARGS✿: The input of tool N
✿RESULT✿: The result of tool 1
✿RESULT✿: The result of tool 2
...
✿RESULT✿: The result of tool N
✿RETURN✿: Reply based on tool results. Images need to be rendered as ![](url)

在确定_preprocess_messages的提示词之后，我们跳回去看一下qwen_agent/llm/base.py下chat函数的关键执行逻辑

1. 配置和语言处理

generate_cfg = merge_generate_cfgs(base_generate_cfg=self.generate_cfg, new_generate_cfg=extra_generate_cfg)
if 'seed' not in generate_cfg:generate_cfg['seed'] = random.randint(a=0, b=2**30)
if 'lang' in generate_cfg:lang: Literal['en', 'zh'] = generate_cfg.pop('lang')
else:lang: Literal['en', 'zh'] = 'zh' if has_chinese_messages(messages) else 'en'if messages[0].role != SYSTEM:messages = [Message(role=SYSTEM, content=DEFAULT_SYSTEM_MESSAGE)] + messages

这部分代码：

• 合并基本配置和额外配置
• 确保有一个随机种子（如果未提供）
• 确定语言（通过配置或自动检测中文）
• 如果第一条消息不是系统消息，添加默认系统消息

2. 输入截断处理

max_input_tokens = generate_cfg.pop('max_input_tokens', DEFAULT_MAX_INPUT_TOKENS)
if max_input_tokens > 0:messages = _truncate_input_messages_roughly(messages=messages,max_tokens=max_input_tokens,)

根据最大输入令牌限制截断消息，避免超过模型的输入限制。注释指出这是一个粗略的估计，尤其对函数调用和多模态内容的令牌计数不精确。通过查看相关代码实现，截断逻辑事实上采用了"保留最新（采用从最新到最旧的方式截断，优先保留最近的对话）、完整轮次（按完整轮次（用户+助手）进行保留，不会截断单个轮次中间，至少保留最新的一轮用户-助手交互）、系统消息必留（无论令牌限制如何，系统消息始终保留）"的策略，尽可能保证在令牌限制内保留最有价值的对话内容。

3. 函数调用模式处理

if functions:fncall_mode = True
else:fncall_mode = False
if 'function_choice' in generate_cfg:fn_choice = generate_cfg['function_choice']valid_fn_choices = [f.get('name', f.get('name_for_model', None)) for f in (functions or [])]valid_fn_choices = ['auto', 'none'] + [f for f in valid_fn_choices if f]if fn_choice not in valid_fn_choices:raise ValueError(f'The value of function_choice must be one of the following: {valid_fn_choices}. 'f'But function_choice="{fn_choice}" is received.')if fn_choice == 'none':fncall_mode = False

这段逻辑处理函数调用相关配置：

• 默认情况下，如果提供了functions参数则启用函数调用
• 验证function_choice参数是否有效
• 如果function_choice为’none’，则禁用函数调用

4. 预处理和多模态处理

messages = self._preprocess_messages(messages, lang=lang, generate_cfg=generate_cfg, functions=functions)
if not self.support_multimodal_input:messages = [format_as_text_message(msg, add_upload_info=False) for msg in messages]

对消息进行预处理，并处理不支持多模态输入的模型情况，将消息格式化为纯文本。

5. 模型服务调用逻辑

def _call_model_service():if fncall_mode:return self._chat_with_functions(messages=messages,functions=functions,stream=stream,delta_stream=delta_stream,generate_cfg=generate_cfg,lang=lang,)else:# TODO: Optimize code structureif messages[-1].role == ASSISTANT:assert not delta_stream, 'Continuation mode does not currently support `delta_stream`'return self._continue_assistant_response(messages, generate_cfg=generate_cfg, stream=stream)else:return self._chat(messages,stream=stream,delta_stream=delta_stream,generate_cfg=generate_cfg,)

这个内部函数根据不同情况调用不同的模型服务方法：

• 函数调用模式使用_chat_with_functions

• 如果最后一条消息是助手消息，使用_continue_assistant_response（续写模式）

  续写模式本质上是一种"巧妙的欺骗"：

  1. 它将用户的最新消息和助手的最新回复合并成一个新的"用户消息"
  2. 这让模型认为用户输入了一段已经包含部分回答的文本
  3. 模型会自然地继续这段文本，从而实现"续写"效果
  4. 从用户角度看，这就像是助手继续了之前的回复

  • 这种方法的优点是不需要模型专门支持续写功能，通过重新构造输入就能实现。但它也有局限性，例如对函数调用不支持，且可能在某些复杂场景下效果不佳。

  • 总结来说，续写模式通过将"已有对话"重新构造为"用户输入"，让模型自然地延续之前的回复，实现了简单而有效的续写功能。

• 否则使用普通的_chat方法

6. 重试和流式处理

if stream and delta_stream:# No retry for delta streamingoutput = _call_model_service()
elif stream and (not delta_stream):output = retry_model_service_iterator(_call_model_service, max_retries=self.max_retries)
else:output = retry_model_service(_call_model_service, max_retries=self.max_retries)

根据流式选项决定如何调用模型服务：

• delta_stream模式下不进行重试（已弃用）
• 普通流模式使用迭代器重试方法
• 非流模式使用普通重试方法

7. 后处理和返回结果

非流式情况：

if isinstance(output, list):assert not streamlogger.debug(f'LLM Output:\n{pformat([_.model_dump() for _ in output], indent=2)}')output = self._postprocess_messages(output, fncall_mode=fncall_mode, generate_cfg=generate_cfg)if not self.support_multimodal_output:output = _format_as_text_messages(messages=output)if self.cache:self.cache.set(cache_key, json_dumps_compact(output))return self._convert_messages_to_target_type(output, _return_message_type)

流式情况：

else:assert streamif delta_stream:# Hack: To avoid potential errors during the postprocessing of stop words when delta_stream=True.# Man, we should never have implemented the support for `delta_stream=True` in the first place!generate_cfg = copy.deepcopy(generate_cfg)  # copy to avoid conflicts with `_call_model_service`assert 'skip_stopword_postproc' not in generate_cfggenerate_cfg['skip_stopword_postproc'] = Trueoutput = self._postprocess_messages_iterator(output, fncall_mode=fncall_mode, generate_cfg=generate_cfg)def _format_and_cache() -> Iterator[List[Message]]:o = []for o in output:if o:if not self.support_multimodal_output:o = _format_as_text_messages(messages=o)yield oif o and (self.cache is not None):self.cache.set(cache_key, json_dumps_compact(o))return self._convert_messages_iterator_to_target_type(_format_and_cache(), _return_message_type)

这部分处理结果输出：

• 对输出进行后处理
• 处理不支持多模态输出的情况
• 根据需要缓存结果
• 将结果转换为原始输入类型（字典或Message对象）
• 流式输出使用迭代器，并在结束时缓存最终结果

总体流程图如下：

关于并行调用工具失效的分析

在上一章中，我们尝试通过在run函数中设置parallel_function_calls=True来实现工具的并行化调用，但并没有生效。经过分析资料和阅读源码，我发现真正的调用方法应该写成下面这种，即在extra_generate_cfg里面来设置

    def _run(self, messages: List[Message], lang: Literal['en', 'zh'] = 'en', **kwargs) -> Iterator[List[Message]]:messages = copy.deepcopy(messages)num_llm_calls_available = MAX_LLM_CALL_PER_RUNresponse = []while True and num_llm_calls_available > 0:num_llm_calls_available -= 1extra_generate_cfg = {'lang': lang, 'parallel_function_calls': True}...

但即使是这样设，依然没有预期的并行调用的效果，所以我们再往深处瞧一瞧。

跟分析中英文提示词一样，从调用模型服务时的self._chat_with_functions追根溯源到qwen_agent/llm/function_calling.py，里面有这样一个神奇的函数：

    def _chat_with_functions(self,messages: List[Message],functions: List[Dict],stream: bool,delta_stream: bool,generate_cfg: dict,lang: Literal['en', 'zh'],) -> Union[List[Message], Iterator[List[Message]]]:if delta_stream:raise NotImplementedError('Please use stream=True with delta_stream=False, because delta_stream=True'' is not implemented for function calling due to some technical reasons.')generate_cfg = copy.deepcopy(generate_cfg)for k in ['parallel_function_calls', 'function_choice']:if k in generate_cfg:del generate_cfg[k]return self._continue_assistant_response(messages, generate_cfg=generate_cfg, stream=stream)

在倒数第二行，它移除了配置字典中的并行函数调用parallel_function_calls和函数选择配置function_choice！！！所以你设了这两个参数也是白设！！！

所以并行函数调用唯一影响的地方只有提示词，而在我看来，似乎实际上并不会对调用并行有什么实际影响。

这个函数最后调用了_continue_assistant_response续写函数，有点奇怪，因为最后一句的角色不是assistant，等价于直接调用_chat函数。

ReActChat类

在FnCallAgent类的基础上，qwen_agent/agents/fncall_agent.py下的ReActChat类又重写了三个函数，接下来我们就分别来看一下它们的源码实现，并做对应的分析。

初始化函数

ReActChat 的初始化函数 __init__ 继承了 FnCallAgent 的初始化方法，并设置了特定的生成配置参数：

def __init__(self,function_list: Optional[List[Union[str, Dict, BaseTool]]] = None,llm: Optional[Union[Dict, BaseChatModel]] = None,system_message: Optional[str] = DEFAULT_SYSTEM_MESSAGE,name: Optional[str] = None,description: Optional[str] = None,files: Optional[List[str]] = None,**kwargs):super().__init__(function_list=function_list,llm=llm,system_message=system_message,name=name,description=description,files=files,**kwargs)self.extra_generate_cfg = merge_generate_cfgs(base_generate_cfg=self.extra_generate_cfg,new_generate_cfg={'stop': ['Observation:', 'Observation:\n']},)

这里的关键点是设置了 stop 序列为 ['Observation:', 'Observation:\n']，这是为了在生成过程中在观察结果之前停止生成，以便于后续处理中插入工具调用的结果。这种方式确保了 ReAct 格式（思考-行动-观察循环）的正确实现。

执行函数

_run 方法是实现 ReAct 推理过程的核心，接收消息列表并使用迭代器返回回复：

def _run(self, messages: List[Message], lang: Literal['en', 'zh'] = 'en', **kwargs) -> Iterator[List[Message]]:text_messages = self._prepend_react_prompt(messages, lang=lang)num_llm_calls_available = MAX_LLM_CALL_PER_RUNresponse: str = 'Thought: 'while num_llm_calls_available > 0:num_llm_calls_available -= 1# 显示流式响应output = []for output in self._call_llm(messages=text_messages):if output:yield [Message(role=ASSISTANT, content=response + output[-1].content)]# 累积当前响应if output:response += output[-1].contenthas_action, action, action_input, thought = self._detect_tool(output[-1].content)if not has_action:break# 添加工具调用结果observation = self._call_tool(action, action_input, messages=messages, **kwargs)observation = f'\nObservation: {observation}\nThought: 'response += observationyield [Message(role=ASSISTANT, content=response)]if (not text_messages[-1].content.endswith('\nThought: ')) and (not thought.startswith('\n')):# 在"Question:"和第一个"Thought:"之间添加换行符text_messages[-1].content += '\n'if action_input.startswith('```'):# 为代码块的正确渲染添加换行符action_input = '\n' + action_inputtext_messages[-1].content += thought + f'\nAction: {action}\nAction Input: {action_input}' + observation

这个函数实现了 ReAct 格式的核心流程：

1. 首先调用 _prepend_react_prompt 将 ReAct 提示词添加到消息中
2. 使用计数器限制 LLM 调用次数，防止无限循环
3. 开始响应总是以 "Thought: " 开头
4. 在循环中：
   • 调用 LLM 生成一段回复
   • 使用 _detect_tool 解析回复中的工具调用信息
   • 如果没有检测到工具调用，则完成推理并结束循环
   • 如果检测到工具调用，则执行工具并获取观察结果
   • 将观察结果和新的 "Thought: " 提示添加到响应中
   • 更新历史消息以包含完整的思考、行动和观察过程，为下一轮生成提供上下文

与Function Call方法的区别

注意在调用公共推理函数的时候，ReActChat的写法是self._call_llm(messages=text_messages)，而FnCallAgent的写法是self._call_llm(messages=messages, functions=[func.function for func in self.function_map.values()],extra_generate_cfg=extra_generate_cfg)，这是因为ReActChat已经把Function的信息和内容都汇聚到了文本中，因此它的Function为None，Function Call所调用的一系列与Function相关的处理与后处理都被跳过了。

提示词处理

_prepend_react_prompt 方法负责将 ReAct 格式的提示词添加到用户消息中：

def _prepend_react_prompt(self, messages: List[Message], lang: Literal['en', 'zh']) -> List[Message]:tool_descs = []for f in self.function_map.values():function = f.functionname = function.get('name', None)name_for_human = function.get('name_for_human', name)name_for_model = function.get('name_for_model', name)assert name_for_human and name_for_modelargs_format = function.get('args_format', '')tool_descs.append(TOOL_DESC.format(name_for_human=name_for_human,name_for_model=name_for_model,description_for_model=function['description'],parameters=json.dumps(function['parameters'], ensure_ascii=False),args_format=args_format).rstrip())tool_descs = '\n\n'.join(tool_descs)tool_names = ','.join(tool.name for tool in self.function_map.values())text_messages = [format_as_text_message(m, add_upload_info=True, lang=lang) for m in messages]text_messages[-1].content = PROMPT_REACT.format(tool_descs=tool_descs,tool_names=tool_names,query=text_messages[-1].content,)return text_messages

这个函数的主要工作：

1. 为每个工具生成描述，包括名称、描述和参数信息
2. 将所有工具描述合并为一个字符串
3. 生成用于提示中的工具名称列表
4. 将原始消息转换为文本格式
5. 使用 PROMPT_REACT 模板替换最后一条消息内容，并填入工具描述、工具名称和原始查询

这样确保了 LLM 在生成回复时遵循 ReAct 格式的思考、行动和观察循环。

工具解析

_detect_tool 方法负责从 LLM 回复中解析工具调用信息：

def _detect_tool(self, text: str) -> Tuple[bool, str, str, str]:special_func_token = '\nAction:'special_args_token = '\nAction Input:'special_obs_token = '\nObservation:'func_name, func_args = None, Nonei = text.rfind(special_func_token)j = text.rfind(special_args_token)k = text.rfind(special_obs_token)if 0 <= i < j:  # 如果文本包含 `Action` 和 `Action input`if k < j:  # 但不包含 `Observation`# 那么可能是 LLM 输出省略了停止词# 因为输出文本可能丢弃了停止词text = text.rstrip() + special_obs_token  # 将其添加回去k = text.rfind(special_obs_token)func_name = text[i + len(special_func_token):j].strip()func_args = text[j + len(special_args_token):k].strip()text = text[:i]  # 返回工具调用前的响应，即 `Thought`return (func_name is not None), func_name, func_args, text

这个函数通过查找特殊标记（Action、Action Input、Observation）来解析 LLM 输出中的工具调用信息：

1. 查找最后出现的 “Action:”、“Action Input:” 和 “Observation:” 标记
2. 如果找到了 Action 和 Action Input 但没有 Observation，则手动添加一个 Observation 标记
3. 解析出工具名称（func_name）和工具参数（func_args）
4. 返回一个元组，包含：
   • 是否存在工具调用
   • 工具名称
   • 工具参数
   • 思考部分的文本内容

这个函数确保了即使 LLM 的输出格式略有偏差，也能正确解析出工具调用信息，保证了 ReAct 过程的稳定执行。

关于中英文提示词的分析

在qwen_agent/agents/react_chat.py中，一开始就定义了英文提示词模板

TOOL_DESC = ('{name_for_model}: Call this tool to interact with the {name_for_human} API. ''What is the {name_for_human} API useful for? {description_for_model} Parameters: {parameters} {args_format}')PROMPT_REACT = """Answer the following questions as best you can. You have access to the following tools:{tool_descs}Use the following format:Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Thought: I now know the final answer
Final Answer: the final answer to the original input questionBegin!Question: {query}
Thought: """

其经_prepend_react_prompt 方法处理后的实例如下：

[Message({'role': 'user', 'content': 'Answer the following questions as best you can. You have access to the following tools:\n\nmultiply: Call this tool to interact with the multiply API. What is the multiply API useful for? 两个整数相乘 Parameters: [[{"name": "first_int", "type": "integer", "description": "第一个整数", "required": true}, {"name": "second_int", "type": "integer", "description": "第二个整数", "required": true}]] 此工具的输入应为JSON对象。\n\nadd: Call this tool to interact with the add API. What is the add API useful for? 两个整数相加 Parameters: [[{"name": "first_add", "type": "integer", "description": "第一个整数", "required": true}, {"name": "second_add", "type": "integer", "description": "第二个整数", "required": true}]] 此工具的输入应为JSON对象。\n\nUse the following format:\n\nQuestion: the input question you must answer\nThought: you should always think about what to do\nAction: the action to take, should be one of [multiply,add]\nAction Input: the input to the action\nObservation: the result of the action\n... (this Thought/Action/Action Input/Observation can be repeated zero or more times)\nThought: I now know the final answer\nFinal Answer: the final answer to the original input question\n\nBegin!\n\nQuestion: 3乘12的数值是多少？以及，11加49是多少？\nThought: '})]

那么中文体现在哪里呢，其实只在qwen_agent/utils/utils.py中的def format_as_multimodal_message( 下作了一些小小的翻译：

for f in [get_basename_from_url(f) for f in files]:if is_image(f):if has_zh:upload.append(f'![图片]({f})')else:upload.append(f'![image]({f})')else:if has_zh:upload.append(f'[文件]({f})')else:upload.append(f'[file]({f})')
upload = ' '.join(upload)
if has_zh:upload = f'（上传了 {upload}）\n\n'
else:upload = f'(Uploaded {upload})\n\n'

关于并行调用工具失效的分析

在上一章中，我们尝试通过在run函数中设置parallel_function_calls=True来实现工具的并行化调用，但并没有生效。事实上，由于工具都写在了提示词中，在qwen_agent/llm/base.py中的_call_model_service并没有调用self._chat_with_functions而是直接进入了self._chat。所以至少，这个参数如果你去设，它不会被人为删掉。

那么_chat的下一步是哪里呢？其实是qwen_agent/llm/oai.py中的_chat_complete_create函数

def _chat_complete_create(*args, **kwargs):# OpenAI API v1 does not allow the following args, must pass by extra_bodyextra_params = ['top_k', 'repetition_penalty']if any((k in kwargs) for k in extra_params):kwargs['extra_body'] = copy.deepcopy(kwargs.get('extra_body', {}))for k in extra_params:if k in kwargs:kwargs['extra_body'][k] = kwargs.pop(k)if 'request_timeout' in kwargs:kwargs['timeout'] = kwargs.pop('request_timeout')client = openai.OpenAI(**api_kwargs)return client.chat.completions.create(*args, **kwargs)

看起来好像没什么问题，试试看设置有没有用。

哈哈哈哈哈参数不正确，可以放心地死心了。