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

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

news 来源：原创 2025/6/21 14:09:41

在前面的几篇文章如《针对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 knowledge
    else:
        docs = result
    try:
        # 文档列表格式处理
        _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

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

  1. 字符串格式：尝试解析为JSON，失败则作为整体内容
  2. 文档列表格式：包含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_prompt
        else:
            messages = [Message(role=SYSTEM, content=knowledge_prompt)] + messages
    return messages

处理流程：

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

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 NotImplementedError
    return 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_prompt
    else:
        messages.insert(0, Message(SYSTEM, system_prompt))

    response = self._call_llm(messages=messages)
    return response

处理流程：

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

4. 两个类的比较

4.1 相同点

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

4.2 区别点

  1. 用途不同：

    • Assistant类是通用助手，集成了RAG和函数调用能力
    • BasicDocQA类专注于文档问答，适用于需要定位具体细节的问题

  2. 提示模板不同：

    • Assistant类使用知识库模板，强调知识来源
    • BasicDocQA类使用参考资料模板，强调对参考资料的理解和组织

  3. 处理流程不同：

    • Assistant类先格式化知识，再添加到系统消息
    • BasicDocQA类直接将知识作为参考资料添加到系统消息

5. 工作流程分析

5.1 Assistant类工作流程

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

5.2 BasicDocQA类工作流程

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

6. 实际应用场景

6.1 Assistant类适用场景

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

6.2 BasicDocQA类适用场景

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

7. 总结

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

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

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

相关文章：

  • 《比特城传奇：公钥、私钥与网络安全的守护之战》
  • 【Mac 从 0 到 1 保姆级配置教程 11】- Mac 基础配置 Finder、触控板、常用快捷键等
  • 数据库访问工具 dbVisitor v6.0.0 发布
  • 从零实现本地大模型RAG部署
  • PostgreSQL 16深度解析（从16.0-16.8）
  • 正则表达式 - 简介
  • 【算法】堆排序
  • jeecg拦截器 排除地址不让他拦截
  • 使用傅里叶级数和Python表示方波
  • 深度解析新能源汽车研发测试中的关键信号采集技术
  • Qt实现鼠标拖动窗口
  • Java 基础-32-枚举-枚举的应用场景
  • Flink介绍——实时计算核心论文之Storm论文详解
  • 以太网安全
  • 达芬奇20正式版 Davinci Resolve Studio 20 支持WIN
  • 【HarmonyOS Next之旅】DevEco Studio使用指南(十一)
  • Systemd构建容器化微服务集群管理系统
  • Ubuntu 22 Linux上部署DeepSeek R1保姆式操作详解（Xinference方式）
  • C++之红黑树
  • JAVA EE_多线程-初阶（二）
  • 什么网站做招聘收录好/百度导航下载安装手机导航
  • 制作html网站/google广告投放
  • 网站劫持必须做系统嘛/seo点击排名
  • 冠县哪里有做网站的/优化推广网站怎么做
  • 门户网站 管理系统/跨境电商培训机构哪个靠谱
  • 南宁新站seo/百度高级检索入口