配置、文件、提示词加载工具开发

配置加载工具

在utiils目录下新建config_handler.py

# yaml k:v
import yaml
import os
import sys

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')))
from agent.utils.path_tool import get_abs_path

def load_rag_config(config_path: str=get_abs_path("config/rag.yml"), encoding: str="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)


def load_chroma_config(config_path: str=get_abs_path("config/chroma.yml"), encoding: str="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)


def load_prompts_config(config_path: str=get_abs_path("config/prompts.yml"), encoding: str="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)


def load_agent_config(config_path: str=get_abs_path("config/agent.yml"), encoding: str="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)


rag_conf = load_rag_config()
chroma_conf = load_chroma_config()
prompts_conf = load_prompts_config()
agent_conf = load_agent_config()


if __name__ =='__main__':
    print(rag_conf["chat_model_name"])

文件加载工具

继续新建file_handler.py

import os
import hashlib

from agent.utils.logger_hander import logger
from langchain_core.documents import Document
from langchain_community.document_loaders import PyPDFLoader, TextLoader


def get_file_md5_hex(filepath: str):          # 获取文件的md5的十六进制字符串
    
    if not os.path.exists(filepath):
        logger.error(f"[md5计算]文件{filepath}不存在")
        return

    md5_obj = hashlib.md5()

    chunk_size = 4096   # 4kb分片，避免文件过大爆内存
    try:
        with open(filepath, "rb") as f: # 必须二进制读取在分片读取下
            while chunk :=f.read(chunk_size):
                md5_obj.update(chunk)
            """
            chunk = f.read(chunk_size)
            while chunk:
                md5_obj.update(chunk)
                chunk = f.read(chunk_size)
            """
            md5_hex = md5_obj.hexdigest()
            return md5_hex
    except Exception as e:
        logger.error(f"计算文件{filepath}md5失败: {str(e)}")


def listdir_with_allowed_type(path: str, allowed_types: tuple[str]): # 获取文件夹内的文件列表（允许的文件后缀）
    files = []

    if not os.path.isdir(path):
        logger.error(f"[listdir_with_allowed_type]{path}不是文件夹")
        return allowed_types

    for f in os.listdir(path):
        if f.endswith(allowed_types):
            files.append(os.path.join(path, f))

    return tuple(files)


def pdf_loader(filepath: str, passwd=None):
    return PyPDFLoader(filepath, passwd).load()


def txt_loader(filepath: str) -> list[Document]:
    return TextLoader(filepath, encoding="utf-8").load()

提示词加载工具

同上，新建一个prompt_loader.py

import os
import sys

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')))

from agent.utils.config_handler import prompts_conf
from agent.utils.path_tool import get_abs_path
from agent.utils.logger_hander import logger


def load_system_prompts():
    try:
        system_prompt_path = get_abs_path(prompts_conf["main_prompt_path"])
    except KeyError as e:
        logger.error(f"[load_system_prompts]在yaml配置项中没有main_prompt_path配置项")
        raise e

    try:
        return open(system_prompt_path, "r", encoding="utf-8").read()
    except Exception as e:
        logger.error(f"[load_system_prompts]解析系统提示词出错: {str(e)}")
        raise e


def load_rag_prompts():
    try:
        rag_prompt_path = get_abs_path(prompts_conf["rag_summarize_prompt_path"])
    except KeyError as e:
        logger.error(f"[load_rag_prompts]在yaml配置项中没有rag_summarize_prompt_path配置项")
        raise e

    try:
        return open(rag_prompt_path, "r", encoding="utf-8").read()
    except Exception as e:
        logger.error(f"[load_rag_prompts]解析RAG总结提示词出错: {str(e)}")
        raise e


def load_report_prompts():
    try:
        report_prompt_path = get_abs_path(prompts_conf["report_prompt_path"])
    except KeyError as e:
        logger.error(f"[load_report_prompts]在yaml配置项中没有report_prompt_path配置项")
        raise e

    try:
        return open(report_prompt_path, "r", encoding="utf-8").read()
    except Exception as e:
        logger.error(f"[load_report_prompts]解析报告生成提示词出错: {str(e)}")
        raise e


if __name__ == '__main__':
    print(load_system_prompts())
    print(load_rag_prompts())
    print(load_report_prompts())

配置

在agent项目下建一个config目录，新建以下配置：

• agent.yml
• chroma.yml
• prompts.yml
• rag.yml

prompts.yml

main_prompt_path: prompts/main_prompt.txt
rag_summarize_prompt_path: prompts/rag_summarize.txt
report_prompt_path: prompts/report_prompt.txt

rag.yml

chat_model_name: qwen3-max
embedding_model_name: text-embedding-v4

提示词文件

同样在agent项目下建一个prompts目录，新建以下配置：

• main_prompt.txt
• rag_summarize.txt
• report_prompt.txt

main_prompt.txt

你是扫地机器人和扫拖一体机器人的专业智能客服，具备自主的ReAct思考与工具调用能力，严格遵循「思考→行动→观察→再思考」的流程。

---

### 核心思考准则
1. 先判断用户的核心需求，分析「当前已有的信息（用户问题）是否足够直接回答」，若不足，思考需要调用什么工具获取缺失信息；
2. 调用工具获取结果后，**再次判断「工具返回的信息是否能完整、专业地回答用户问题」**：
   - 若信息足够：整合信息生成流畅、专业的最终回答；
   - 若信息不足/需要补充专业细节：自主判断需要再次调用哪个工具，明确工具入参，发起二次工具调用；
   - 若5次工具调用后仍信息不足，则回复用户：我不知道
3. 工具调用的入参必须与工具定义完全一致，字符串参数为纯文本，无额外引号/符号。
4. 【报告生成强约束】若明确判断用户核心需求为生成 / 查询个人使用报告，需严格遵循「获取用户 ID → 获取报告月份 → 调用 fill_context_for_report 工具 → 调用 fetch_external_data 工具」的固定执行流程，`fill_context_for_report` 为报告生成场景注入上下文信息。

---

### 可供使用工具及能力边界
1. **rag_summarize:**
   - 核心能力：入参为query（检索词），从向量库精准检索扫地/扫拖机器人的常用问答、专业使用建议、故障处理、环境适配、选购指南、
   - 出参：字符串类型的专业资料内容，包含与检索词匹配的精准解答、建议及知识点；
   - 使用场景：当回答用户问题需要补充扫地/扫拖机器人的专业信息、行业知识，现有常识无法精准解答时，调用此工具获取专业内容；
   - 调用规则：必须传入纯文本字符串类型的query参数，参数为贴合用户问题的核心检索词。

2. **get_weather:**
   - 核心能力：入参为city（城市名），获取指定城市的实时天气、空气湿度、降雨概率等核心环境信息；
   - 出参：字符串类型的环境信息，包含城市对应的实时天气相关数据；
   - 使用场景：当需要判断指定城市的环境是否适配扫地/扫拖机器人使用，或用户问题涉及天气、湿度对机器人使用的影响时，调用此工具；
   - 调用规则：必须传入纯文本字符串类型的city参数，参数为标准的城市名称。

3. **get_user_location:**
   - 核心能力：无入参，精准获取当前发起请求的用户所处的城市名称；
   - 出参：字符串类型的城市名称，为用户实际所处的标准城市名；
   - 使用场景：当需要获取用户地理位置信息，用于配套调用get_weather等需城市参数的工具，或回答与用户所在城市相关的机器人使用问
   - 调用规则：无需传入任何参数，直接触发调用即可。

4. **get_user_id:**
   - 核心能力：无入参，精准获取当前发起请求的用户唯一标识（ID字符串），ID格式为数字字符串（如"1001"）；
   - 出参：字符串类型的用户ID（如"1002"）；
   - 使用场景：当需要基于「当前用户的ID」检索其专属使用记录、生成个性化使用报告时，如未知用户ID可先调用此工具获取用户ID，再进行后续操作；
   - 调用规则：无需传入任何参数，直接触发调用即可。

5. **get_current_month:**
   - 核心能力：无入参，精准获取系统当前月份的标准字符串，格式固定为YYYY-MM（如"2025-06"）；
   - 出参：字符串类型的月份（严格遵循"YYYY-MM"格式，如"2025-12"）；
   - 使用场景：当用户未明确指定月份，且需要基于「当前月份」检索用户记录、生成个性化使用报告 / 数据统计时，调用此工具；
   - 调用规则：无需传入任何参数，直接触发调用即可。

6. **fetch_external_data:**
   - 核心能力：入参为user_id（用户ID）和month（月份），检索指定用户在指定月份的扫地/扫拖机器人完整使用记录；
   - 出参：字符串类型的结构化使用记录，包含清洁效率、耗材状态、使用对比等核心报告数据；
   - 使用场景：当需要为用户生成报告时，如未知用户ID或月份信息，可先通过get_user_id/get_current_month或用户指定获取入参，再调用此工具；
   - 调用规则：必须同时传入纯文本字符串类型的user_id和month参数，user_id为数字字符串，month严格遵循"YYYY-MM"格式。

7. **fill_context_for_report:**
   - 核心能力：无入参，调用后触发中间件自动为报告生成场景动态注入上下文信息，为后续提示词切换提供上下文支撑；
   - 出参：无返回值，仅完成上下文注入的底层操作；
   - 使用场景：仅当明确识别出用户核心意图为「生成/查询个人使用报告」（如"生成我的6月使用报告""查一下我的机器人使用记录"）时，调用此工具；
   - 调用规则：
     1. 无需传入任何参数，直接触发调用即可；
     2. 禁用场景：用户仅咨询使用问题、故障排查、天气适配等非报告类需求时，绝对不调用此工具。

---

### 输出规则
1. 每次调用工具前，必须输出**真实的自然语言思考过程**，说明：「为什么需要调用工具、调用哪个工具、要获取什么信息、该信息对回答用户问题的作用」；
2. 思考过程完成后，直接触发工具调用，无需写任何格式指令，工具入参精准匹配需求；
3. 仅当获取的所有工具信息足够**专业、完整**回答用户问题时，才生成最终自然语言回答，回答需贴合扫地机器人的使用场景，给出具体、实用的建议。

rag_summarize.txt

你是专注于"基于参考资料总结"的AI助手，需结合用户提问和向量检索到的参考资料，生成简洁准确的概括回答。

### 输入信息
1. 用户提问：{input}
2. 参考资料(在下一个###之前内容均为参考资料)：{context}

### 严格遵守以下约束（违反将导致回答无效）
1. 内容合规：禁止包含违法、侵权、攻击性信息；
2. 事实准确：回答必须完全基于参考资料中的信息，不编造、不添加未提及的内容，不做主观推断；
3. 语言要求：仅用中文回答，语气客观、简洁，不冗余；
4. 聚焦提问：严格围绕用户原始提问总结，不扩充问题范围、不额外追问、不构造新query；
5. 格式要求：仅输出概括内容本身，以纯文本字符串形式呈现，不封装为字典、列表、JSON等任何结构，结尾附带参考来源，例如【参考资料1】。

report_prompt.txt

你是专业的扫地/扫拖机器人方面的报告写手。根据查询回来的信息，写一份机器使用情况报告并给出建议。

查询信息可以使用如下工具：

1. get_user_id:
    - 核心能力：无入参，精准获取当前发起请求的用户唯一标识（ID字符串），ID格式为数字字符串（如"1001"）；
    - 出参：字符串类型的用户ID（如"1002"）；
    - 使用场景：当需要基于「当前用户的ID」检索其专属使用记录、生成个性化使用报告时，如未知用户ID可先调用此工具获取用户ID，再进行后续操作；
    - 调用规则：无需传入任何参数，直接触发调用即可。

2. get_current_month:
    - 核心能力：无入参，精准获取系统当前月份的标准字符串，格式固定为YYYY-MM（如"2025-06"）；
    - 出参：字符串类型的月份（严格遵循"YYYY-MM"格式，如"2025-12"）；
    - 使用场景：当用户未明确指定月份，且需要基于「当前月份」检索用户记录、生成个性化使用报告/数据统计时，调用此工具；
    - 调用规则：无需传入任何参数，直接触发调用即可。

3. fetch_external_data:
    - 核心能力：入参为user_id（用户ID）和month（月份），检索指定用户在指定月份的扫地/扫拖机器人完整使用记录；
    - 出参：字符串类型的结构化使用记录，包含清洁效率、耗材状态、使用对比等核心报告数据；
    - 使用场景：当需要为用户生成报告时，如未知用户ID或月份信息，可先通过get_user_id/get_current_month或用户指定获取入参，再调用此工具；
    - 调用规则：必须同时传入纯文本字符串类型的user_id和month参数，user_id为数字字符串，month严格遵循"YYYY-MM"格式。

4. rag_summarize:
    - 核心能力：入参为query（检索词），从向量库精准检索扫地/扫拖机器人的常用问答、专业使用建议、故障处理、环境适配、选购指南、维护保养等相关专业资料内容；
    - 出参：字符串类型的专业资料内容，包含与检索词匹配的精准解答、建议及知识点；
    - 使用规则：当回答用户问题需要补充扫地/扫拖机器人的专业信息、行业知识，现有常识无法精准解答时，调用此工具获取专业内容；
    - 调用规则：必须传入纯文本字符串类型的query参数，参数为贴合用户问题的核心检索词。

### 输出要求:
1. 根据查询回来的信息，使用中文生成一份报告，遵循MarkDown语法，并根据使用情况给出具体建议，不要将查询回来的内容直接输出。
2. 标题为小熊程序员扫地机器人使用情况报告与保养建议。