当前位置：首页 > news >正文

Qwen-Agent框架的文件相关操作：从Assistant到BasicDocQA

news 来源：原创 2025/8/24 8:18:02

在前面的几篇文章如《针对Qwen-Agent框架的Function Call及ReAct的源码阅读与解析：Agent基类篇》、《基于Qwen-Agent框架的Function Call及ReAct方式调用自定义工具》、
《针对Qwen-Agent框架的源码阅读与解析：FnCallAgent与ReActChat篇》中，我们已经理解了Agent的具体运作机制与原理，这里我们再以文件相关操作为例，学习一下Agent的实际应用。

事实上，Qwen-Agent是一个强大的智能助手框架，它集成了RAG（检索增强生成）能力和函数调用能力，能够处理各种复杂任务。本文将深入探讨Qwen-Agent框架中的文件相关操作，特别是从Assistant类到BasicDocQA类的实现，分析代码结构、工作原理以及各个组件的功能。

文章目录

2. Assistant类分析
- 2.1 类定义与初始化
- 2.2 知识格式化函数
- 2.3 核心运行方法
- 2.4 知识提示预处理
- 2.5 辅助函数
3. BasicDocQA类分析
- 3.1 类定义与初始化
- 3.2 提示模板
- 3.3 核心运行方法
4. 两个类的比较
- 4.1 相同点
- 4.2 区别点
5. 工作流程分析
- 5.1 Assistant类工作流程
- 5.2 BasicDocQA类工作流程
6. 实际应用场景
- 6.1 Assistant类适用场景
- 6.2 BasicDocQA类适用场景
7. 总结

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

2. Assistant类分析

2.1 类定义与初始化

Assistant类是Qwen-Agent框架中的一个核心类，它继承自FnCallAgent，集成了RAG能力和函数调用能力。

class Assistant(FnCallAgent):"""This is a widely applicable agent integrated with RAG capabilities and function call ability."""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,rag_cfg: Optional[Dict] = None):super().__init__(function_list=function_list,llm=llm,system_message=system_message,name=name,description=description,files=files,rag_cfg=rag_cfg)

初始化参数包括：

function_list：可调用的函数列表
llm：语言模型实例
system_message：系统消息
name和description：Agent的名称和描述
files：文件列表
rag_cfg：RAG配置

2.2 知识格式化函数

format_knowledge_to_source_and_content函数用于将检索结果格式化为源和内容的标准格式：

def format_knowledge_to_source_and_content(result: Union[str, List[dict]]) -> List[dict]:knowledge = []if isinstance(result, str):# 字符串格式处理result = f'{result}'.strip()try:docs = json5.loads(result)except Exception:print_traceback()knowledge.append({'source': '上传的文档', 'content': result})return knowledgeelse:docs = resulttry:# 文档列表格式处理_tmp_knowledge = []assert isinstance(docs, list)for doc in docs:url, snippets = doc['url'], doc['text']assert isinstance(snippets, list)_tmp_knowledge.append({'source': f'[文件]({get_basename_from_url(url)})','content': '\n\n...\n\n'.join(snippets)})knowledge.extend(_tmp_knowledge)except Exception:print_traceback()knowledge.append({'source': '上传的文档', 'content': result})return knowledge

该函数支持两种输入格式：

字符串格式：尝试解析为JSON，失败则作为整体内容
文档列表格式：包含URL和文本片段的字典列表

输出为统一的[{'source': '来源', 'content': '内容'}, ...]格式。

2.3 核心运行方法

_run方法是Assistant类的核心方法，负责处理用户查询：

def _run(self,messages: List[Message],lang: Literal['en', 'zh'] = 'en',knowledge: str = '',**kwargs) -> Iterator[List[Message]]:"""Q&A with RAG and tool use abilities."""new_messages = self._prepend_knowledge_prompt(messages=messages, lang=lang, knowledge=knowledge, **kwargs)return super()._run(messages=new_messages, lang=lang, **kwargs)

该方法首先调用_prepend_knowledge_prompt方法添加知识提示，然后调用父类的_run方法处理消息。

2.4 知识提示预处理

_prepend_knowledge_prompt方法用于在消息前添加知识提示：

def _prepend_knowledge_prompt(self,messages: List[Message],lang: Literal['en', 'zh'] = 'en',knowledge: str = '',**kwargs) -> List[Message]:messages = copy.deepcopy(messages)if not knowledge:# 从文件中检索知识*_, last = self.mem.run(messages=messages, lang=lang, **kwargs)knowledge = last[-1][CONTENT]logger.debug(f'Retrieved knowledge of type `{type(knowledge).__name__}`:\n{knowledge}')if knowledge:knowledge = format_knowledge_to_source_and_content(knowledge)logger.debug(f'Formatted knowledge into type `{type(knowledge).__name__}`:\n{knowledge}')else:knowledge = []snippets = []for k in knowledge:snippets.append(KNOWLEDGE_SNIPPET[lang].format(source=k['source'], content=k['content']))knowledge_prompt = ''if snippets:knowledge_prompt = KNOWLEDGE_TEMPLATE[lang].format(knowledge='\n\n'.join(snippets))if knowledge_prompt:if messages[0][ROLE] == SYSTEM:messages[0][CONTENT] += '\n\n' + knowledge_promptelse:messages = [Message(role=SYSTEM, content=knowledge_prompt)] + messagesreturn messages

处理流程：

如果未提供知识，则从文件中检索
格式化知识为标准格式
将知识片段格式化为多语言模板
将格式化的知识添加到系统消息中

2.5 辅助函数

get_current_date_str函数用于获取当前日期的字符串表示：

def get_current_date_str(lang: Literal['en', 'zh'] = 'en',hours_from_utc: Optional[int] = None,
) -> str:# 获取当前时间if hours_from_utc is None:cur_time = datetime.datetime.now()else:cur_time = datetime.datetime.utcnow() + datetime.timedelta(hours=hours_from_utc)# 根据语言格式化日期字符串if lang == 'en':date_str = 'Current date: ' + cur_time.strftime('%A, %B %d, %Y')elif lang == 'zh':cur_time = cur_time.timetuple()date_str = f'当前时间：{cur_time.tm_year}年{cur_time.tm_mon}月{cur_time.tm_mday}日，星期'date_str += ['一', '二', '三', '四', '五', '六', '日'][cur_time.tm_wday]date_str += '。'else:raise NotImplementedErrorreturn date_str

该函数支持中英文两种格式，可以指定UTC时差。

3. BasicDocQA类分析

3.1 类定义与初始化

BasicDocQA类继承自Assistant类，专门用于文档问答：

class BasicDocQA(Assistant):"""This is an agent for doc QA."""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] = DEFAULT_NAME,description: Optional[str] = DEFAULT_DESC,files: Optional[List[str]] = None,rag_cfg: Optional[Dict] = None):super().__init__(function_list=function_list,llm=llm,system_message=system_message,name=name,description=description,files=files,rag_cfg=rag_cfg)

默认名称和描述：

DEFAULT_NAME = 'Basic DocQA'
DEFAULT_DESC = '可以根据问题，检索出知识库中的某个相关细节来回答。适用于需要定位到具体位置的问题，例如"介绍表1"等类型的问题'

3.2 提示模板

BasicDocQA类使用特定的提示模板，支持中英文：

PROMPT_TEMPLATE_ZH = """请充分理解以下参考资料内容，组织出满足用户提问的条理清晰的回复。
#参考资料：
{ref_doc}"""PROMPT_TEMPLATE_EN = """Please fully understand the content of the following reference materials and organize a clear response that meets the user's questions.
# Reference materials:
{ref_doc}"""PROMPT_TEMPLATE = {'zh': PROMPT_TEMPLATE_ZH,'en': PROMPT_TEMPLATE_EN,
}

3.3 核心运行方法

BasicDocQA类重写了_run方法，使用不同的文档问答提示：

def _run(self, messages: List[Message], lang: str = 'en', **kwargs) -> Iterator[List[Message]]:"""This agent using different doc qa prompt with Assistant"""# 使用Memory agent进行数据管理*_, last = self.mem.run(messages=messages, **kwargs)knowledge = last[-1][CONTENT]messages = copy.deepcopy(messages)system_prompt = PROMPT_TEMPLATE[lang].format(ref_doc=knowledge)if messages[0][ROLE] == SYSTEM:messages[0][CONTENT] += '\n\n' + system_promptelse:messages.insert(0, Message(SYSTEM, system_prompt))response = self._call_llm(messages=messages)return response

处理流程：

使用mem.run检索相关知识
使用特定的文档问答提示模板
将格式化的提示添加到系统消息中
调用语言模型生成回复

4. 两个类的比较

4.1 相同点

都继承自Agent基类，具备Agent的基本能力
都集成了RAG能力，可以检索知识库
都支持中英文两种语言
都通过添加系统提示来引导模型回答

4.2 区别点

用途不同：
- Assistant类是通用助手，集成了RAG和函数调用能力
- BasicDocQA类专注于文档问答，适用于需要定位具体细节的问题
提示模板不同：
- Assistant类使用知识库模板，强调知识来源
- BasicDocQA类使用参考资料模板，强调对参考资料的理解和组织
处理流程不同：
- Assistant类先格式化知识，再添加到系统消息
- BasicDocQA类直接将知识作为参考资料添加到系统消息

5. 工作流程分析

5.1 Assistant类工作流程

初始化Assistant实例，配置参数
接收用户消息
如果未提供知识，从文件中检索相关知识
格式化知识为标准格式（源和内容）
将格式化的知识添加到系统消息中
调用语言模型生成回复

5.2 BasicDocQA类工作流程

初始化BasicDocQA实例，配置参数
接收用户消息
从文件中检索相关知识
使用特定的文档问答提示模板
将格式化的提示添加到系统消息中
调用语言模型生成回复

6. 实际应用场景

6.1 Assistant类适用场景

通用问答系统
需要调用外部函数的场景
多种知识源集成的场景
需要展示知识来源的场景

6.2 BasicDocQA类适用场景

特定文档问答
需要定位文档中具体细节的场景
专注于文档内容理解和组织的场景
例如：“介绍表1”、"第三章说了什么"等具体位置问题

7. 总结

Qwen-Agent框架中的Assistant类和BasicDocQA类展示了框架在文件相关操作中的灵活性和强大能力。Assistant类作为通用助手，集成了RAG能力和函数调用能力，可以处理各种复杂任务；而BasicDocQA类专注于文档问答，适用于需要定位具体细节的问题。

两个类都通过添加系统提示来引导模型回答，但使用了不同的提示模板和处理流程。这种设计使得框架可以根据不同场景灵活配置，提供最适合的回答。

在实际应用中，开发者可以根据具体需求选择合适的类，或者基于这些类进行扩展，构建更加专业和高效的智能助手系统。