教程：https://www.bilibili.com/video/BV1yjz5BLEoY

代码：https://github.com/shangxiang0907/HeiMa-AI-LLM-RAG-Agent-Dev

云开发平台：https://hzh.sealos.run

Agent智能体-11、[Agent项目]tools工具开发

介绍

→提示词：@zhisaotong_agent/config/agent.yml 中实际需要存储以下内容：external_data_path: data/external/records.csv，请帮我根据最佳实践，同时修改@zhisaotong_agent/utils/config_handler.py 中关于它的内容

←AI回复

已按你的需求改好，修改总结如下：

1. zhisaotong_agent/config/agent.yml

• 在 envs.default 下增加 external_data_path，与 agent、conversation 同级。
• 默认值设为 data/external/records.csv，并加了注释说明用途。

default:
  # 外部数据文件路径（相对于项目根目录），供 usage_report 等工具使用
  external_data_path: data/external/records.csv
  agent:
    ...

不同环境（如 dev、prod）如需覆盖，只需在对应 env 下再写一次 external_data_path 即可，会按现有逻辑与 default 合并。

2. zhisaotong_agent/utils/config_handler.py

• AgentConfig 增加字段

  • external_data_path: str = "data/external/records.csv"
    与 yml 默认值一致，并写进 docstring。

• from_dict
  从合并后的配置里读取顶层 external_data_path（即 data.get("external_data_path", cls.external_data_path)），保证未配置时用默认值。

• 便捷方法

  • get_external_data_abs_path() -> str：基于项目根目录解析为绝对路径，便于直接用于文件读写，风格与 PromptsConfig.get_prompt_abs_path 一致。

使用方式示例：

agent_conf = load_agent_config()
# 相对路径（与配置一致）
path_rel = agent_conf.external_data_path  # "data/external/records.csv"
# 绝对路径（用于 open 等）
path_abs = agent_conf.get_external_data_abs_path()

当前无新的 linter 报错；若你在 usage_report 等工具里需要用到该路径，可直接用 agent_config.external_data_path 或 agent_config.get_external_data_abs_path()。

→提示词：我现在要实现@zhisaotong_agent/PROJECT_OVERVIEW.md 中的agent_tools.py，下面是一个已有的实现：

import os
from utils.logger_handler import logger
from langchain_core.tools import tool
from rag.rag_service import RagSummarizeService
import random
from utils.config_handler import agent_conf
from utils.path_tool import get_abs_path

rag = RagSummarizeService()

user_ids = ["1001", "1002", "1003", "1004", "1005", "1006", "1007", "1008", "1009", "1010",]
month_arr = ["2025-01", "2025-02", "2025-03", "2025-04", "2025-05", "2025-06",
             "2025-07", "2025-08", "2025-09", "2025-10", "2025-11", "2025-12", ]

external_data = {}


@tool(description="从向量存储中检索参考资料")
def rag_summarize(query: str) -> str:
    return rag.rag_summarize(query)


@tool(description="获取指定城市的天气，以消息字符串的形式返回")
def get_weather(city: str) -> str:
    return f"城市{city}天气为晴天，气温26摄氏度，空气湿度50%，南风1级，AQI21，最近6小时降雨概率极低"


@tool(description="获取用户所在城市的名称，以纯字符串形式返回")
def get_user_location() -> str:
    return random.choice(["深圳", "合肥", "杭州"])


@tool(description="获取用户的ID，以纯字符串形式返回")
def get_user_id() -> str:
    return random.choice(user_ids)


@tool(description="获取当前月份，以纯字符串形式返回")
def get_current_month() -> str:
    return random.choice(month_arr)


def generate_external_data():
    """
    {
        "user_id": {
            "month" : {"特征": xxx, "效率": xxx, ...}
            "month" : {"特征": xxx, "效率": xxx, ...}
            "month" : {"特征": xxx, "效率": xxx, ...}
            ...
        },
        "user_id": {
            "month" : {"特征": xxx, "效率": xxx, ...}
            "month" : {"特征": xxx, "效率": xxx, ...}
            "month" : {"特征": xxx, "效率": xxx, ...}
            ...
        },
        "user_id": {
            "month" : {"特征": xxx, "效率": xxx, ...}
            "month" : {"特征": xxx, "效率": xxx, ...}
            "month" : {"特征": xxx, "效率": xxx, ...}
            ...
        },
        ...
    }
    :return:
    """
    if not external_data:
        external_data_path = get_abs_path(agent_conf["external_data_path"])

        if not os.path.exists(external_data_path):
            raise FileNotFoundError(f"外部数据文件{external_data_path}不存在")

        with open(external_data_path, "r", encoding="utf-8") as f:
            for line in f.readlines()[1:]:
                arr: list[str] = line.strip().split(",")

                user_id: str = arr[0].replace('"', "")
                feature: str = arr[1].replace('"', "")
                efficiency: str = arr[2].replace('"', "")
                consumables: str = arr[3].replace('"', "")
                comparison: str = arr[4].replace('"', "")
                time: str = arr[5].replace('"', "")

                if user_id not in external_data:
                    external_data[user_id] = {}

                external_data[user_id][time] = {
                    "特征": feature,
                    "效率": efficiency,
                    "耗材": consumables,
                    "对比": comparison,
                }


@tool(description="从外部系统中获取指定用户在指定月份的使用记录，以纯字符串形式返回， 如果未检索到返回空字符串")
def fetch_external_data(user_id: str, month: str) -> str:
    generate_external_data()

    try:
        return external_data[user_id][month]
    except KeyError:
        logger.warning(f"[fetch_external_data]未能检索到用户：{user_id}在{month}的使用记录数据")
        return ""


@tool(description="无入参，无返回值，调用后触发中间件自动为报告生成的场景动态注入上下文信息，为后续提示词切换提供上下文信息")
def fill_context_for_report():
    return "fill_context_for_report已调用"

但是它可能不符合最佳实践和生产规范，帮我指出它的问题，并改造，但是我不希望影响它跟外部交互的接口和方式

←AI回复

已按你的要求完成改造，并保持与外部交互的接口和方式不变。下面是原实现中的问题和本次改造要点。

---

原实现中的问题

1. 配置使用方式错误
   使用了不存在的 agent_conf["external_data_path"]。项目里应通过 load_agent_config() 拿到 AgentConfig，再用 .get_external_data_abs_path() 取路径。

2. 全局可变状态且非线程安全
   用模块级 external_data = {} 并在 generate_external_data() 里直接修改，多线程下可能重复加载或读到半成品。应用“懒加载 + 单次加载 + 锁”的缓存方式。

3. CSV 解析脆弱
   手写按行 split(",") 无法正确处理带逗号的引号字段。应使用标准库 csv.reader（或 csv.DictReader）解析。

4. 导入与项目规范不一致
   使用了 from utils.xxx，而项目其他模块统一用 from zhisaotong_agent.xxx。日志应使用 get_logger(__name__)，与 rag_service 等一致。

5. fetch_external_data 返回值类型与描述不符
   描述为“以纯字符串形式返回”，实际返回的是 dict。已改为先取 dict，再格式化为可读字符串（如 "特征: xxx\n效率: xxx\n..."）后返回。

6. RAG 服务在导入时即初始化
   模块一加载就 RagSummarizeService()，会拖慢启动并强依赖向量库/模型。改为懒加载：首次调用 rag_summarize 时再创建服务。

7. 缺少与 PROJECT_OVERVIEW 一致的位置
   按文档应在 agent/tools/agent_tools.py，并已在该路径创建模块。

---

改造后的行为（接口不变）

• 工具签名与描述：所有 @tool 的入参、返回类型、description 与原先一致，调用方式不变。
• 配置：通过 load_agent_config() 和 agent_config.get_external_data_abs_path() 读取 external_data_path，不再使用 agent_conf。
• 外部数据：用 csv.reader 解析；懒加载 + threading.Lock 保证只加载一次且线程安全；未找到文件时记录警告并返回空数据，不抛错打断工具链。
• RAG：通过 _get_rag_service() 懒加载 RagSummarizeService，避免导入时初始化。
• 命名与导出：常量改为大写（如 USER_IDS、MONTH_ARR）；提供 TOOLS_LIST 和 TOOLS_BY_NAME，便于 react_agent 按 agent.yml 里的 name（如 rag_search、usage_report）做绑定。

文件位置：zhisaotong_agent/agent/tools/agent_tools.py，并已添加 agent/__init__.py 与 agent/tools/__init__.py。若你希望 usage_report 对应别的工具或组合逻辑，只需在绑定工具时改用其他函数或自定义封装即可。

代码

agent_tools.py

"""
智能体业务工具：RAG 查询、用户信息、外部数据、报告上下文等。

设计要点：
- 与 config/agent.yml 中 tools 的 name 对应关系由调用方（如 react_agent）按 name 绑定；
- 配置通过 config_handler 加载，不使用全局 dict；
- 外部数据使用 CSV 标准库解析、懒加载 + 线程安全缓存；
- 所有工具对外入参、返回值与描述保持不变，便于与 ReAct 等框架对接。
"""

from __future__ import annotations

import csv
import random
import threading
from typing import Any, Dict

from langchain_core.tools import tool

from zhisaotong_agent.rag.rag_service import RagSummarizeService
from zhisaotong_agent.utils.api_key import init_dashscope_api_key
from zhisaotong_agent.utils.config_handler import load_agent_config
from zhisaotong_agent.utils.logger_handler import get_logger

logger = get_logger(__name__)

# ---------------------------------------------------------------------------
# 服务与配置（与项目其他模块一致，使用统一配置入口）
# ---------------------------------------------------------------------------

_rag_service: RagSummarizeService | None = None


def _get_rag_service() -> RagSummarizeService:
    """懒加载 RAG 服务，避免在导入时强依赖向量库与模型初始化。"""
    global _rag_service
    if _rag_service is None:
        _rag_service = RagSummarizeService()
    return _rag_service


# 模拟数据：实际场景可由会话/登录态提供
USER_IDS = [
    "1001", "1002", "1003", "1004", "1005",
    "1006", "1007", "1008", "1009", "1010",
]
MONTH_ARR = [
    "2025-01", "2025-02", "2025-03", "2025-04", "2025-05", "2025-06",
    "2025-07", "2025-08", "2025-09", "2025-10", "2025-11", "2025-12",
]

# ---------------------------------------------------------------------------
# 外部数据：懒加载 + 线程安全缓存，使用标准 csv 解析与配置路径
# ---------------------------------------------------------------------------

_external_data_cache: Dict[str, Dict[str, Dict[str, str]]] = {}
_external_data_lock = threading.Lock()


def _load_external_data_from_file(file_path: str) -> Dict[str, Dict[str, Dict[str, str]]]:
    """
    从 CSV 文件加载外部使用记录，返回结构：
    { user_id: { month: { "特征": xxx, "效率": xxx, "耗材": xxx, "对比": xxx } } }
    约定：首行为表头，列顺序为 user_id, 特征, 效率, 耗材, 对比, 月份。
    """
    result: Dict[str, Dict[str, Dict[str, str]]] = {}
    with open(file_path, "r", encoding="utf-8") as f:
        reader = csv.reader(f)
        next(reader, None)  # 跳过表头
        for row in reader:
            if len(row) < 6:
                continue
            user_id = row[0].strip().strip('"')
            feature = row[1].strip().strip('"')
            efficiency = row[2].strip().strip('"')
            consumables = row[3].strip().strip('"')
            comparison = row[4].strip().strip('"')
            time_key = row[5].strip().strip('"')
            if user_id not in result:
                result[user_id] = {}
            result[user_id][time_key] = {
                "特征": feature,
                "效率": efficiency,
                "耗材": consumables,
                "对比": comparison,
            }
    return result


def _get_external_data() -> Dict[str, Dict[str, Dict[str, str]]]:
    """获取外部数据（懒加载、线程安全、仅加载一次）。"""
    global _external_data_cache
    if _external_data_cache:
        return _external_data_cache
    with _external_data_lock:
        if _external_data_cache:
            return _external_data_cache
        agent_config = load_agent_config()
        file_path = agent_config.get_external_data_abs_path()
        try:
            loaded = _load_external_data_from_file(file_path)
            _external_data_cache.clear()
            _external_data_cache.update(loaded)
            logger.info("外部数据已加载，路径=%s，用户数=%d", file_path, len(_external_data_cache))
        except FileNotFoundError:
            logger.warning("外部数据文件不存在，路径=%s，将返回空数据", file_path)
        except Exception as e:
            logger.error("加载外部数据失败，路径=%s，error=%s", file_path, e, exc_info=True)
            raise
    return _external_data_cache


def _format_record_as_string(record: Dict[str, str]) -> str:
    """将单条使用记录格式化为纯字符串，便于工具返回。"""
    return "\n".join(f"{k}: {v}" for k, v in record.items())


# ---------------------------------------------------------------------------
# 工具定义（对外接口与描述保持不变）
# ---------------------------------------------------------------------------


@tool(description="从向量存储中检索参考资料")
def rag_summarize(query: str) -> str:
    return _get_rag_service().rag_summarize(query)


@tool(description="获取指定城市的天气，以消息字符串的形式返回")
def get_weather(city: str) -> str:
    return f"城市{city}天气为晴天，气温26摄氏度，空气湿度50%，南风1级，AQI21，最近6小时降雨概率极低"


@tool(description="获取用户所在城市的名称，以纯字符串形式返回")
def get_user_location() -> str:
    return random.choice(["深圳", "合肥", "杭州"])


@tool(description="获取用户的ID，以纯字符串形式返回")
def get_user_id() -> str:
    return random.choice(USER_IDS)


@tool(description="获取当前月份，以纯字符串形式返回")
def get_current_month() -> str:
    return random.choice(MONTH_ARR)


@tool(
    description="从外部系统中获取指定用户在指定月份的使用记录，以纯字符串形式返回， 如果未检索到返回空字符串"
)
def fetch_external_data(user_id: str, month: str) -> str:
    data = _get_external_data()
    try:
        record = data[user_id][month]
        return _format_record_as_string(record)
    except KeyError:
        logger.warning(
            "[fetch_external_data] 未能检索到用户：%s 在 %s 的使用记录数据",
            user_id,
            month,
        )
        return ""


@tool(
    description="无入参，无返回值，调用后触发中间件自动为报告生成的场景动态注入上下文信息，为后续提示词切换提供上下文信息"
)
def fill_context_for_report() -> str:
    return "fill_context_for_report已调用"


# ---------------------------------------------------------------------------
# 对外集合（供 react_agent 等按 name 绑定，如 rag_search -> rag_summarize）
# ---------------------------------------------------------------------------

TOOLS_LIST = [
    rag_summarize,
    get_weather,
    get_user_location,
    get_user_id,
    get_current_month,
    fetch_external_data,
    fill_context_for_report,
]

# 配置中 name 与工具函数的映射，便于按 agent.yml 的 tools.name 绑定
TOOLS_BY_NAME: Dict[str, Any] = {
    "rag_summarize": rag_summarize,
    "get_weather": get_weather,
    "get_user_location": get_user_location,
    "get_user_id": get_user_id,
    "get_current_month": get_current_month,
    "fetch_external_data": fetch_external_data,
    "fill_context_for_report": fill_context_for_report,
}

# 若 agent.yml 中工具名与函数名不一致，可在此做别名（如 rag_search -> rag_summarize）
TOOLS_BY_NAME.setdefault("rag_search", rag_summarize)
TOOLS_BY_NAME.setdefault("usage_report", fetch_external_data)

__all__ = [
    "rag_summarize",
    "get_weather",
    "get_user_location",
    "get_user_id",
    "get_current_month",
    "fetch_external_data",
    "fill_context_for_report",
    "TOOLS_LIST",
    "TOOLS_BY_NAME",
]

if __name__ == "__main__":
    """
    自测：在项目根目录下运行
        python -m zhisaotong_agent.agent.tools.agent_tools
    仅验证工具可调用与返回格式，不强制要求 DASHSCOPE_API_KEY 或外部数据文件存在。
    """
    # 先手动加载 API Key 到环境变量，供 RAG 等依赖使用
    init_dashscope_api_key()

    print("=== agent_tools 自测 ===\n")

    # 1. 无外部依赖的工具
    print("[1] get_weather('北京') ->", get_weather.invoke({"city": "北京"}))
    print("[2] get_user_location() ->", get_user_location.invoke({}))
    print("[3] get_user_id() ->", get_user_id.invoke({}))
    print("[4] get_current_month() ->", get_current_month.invoke({}))
    print("[5] fill_context_for_report() ->", fill_context_for_report.invoke({}))

    # 2. 外部数据（文件不存在时返回空字符串）
    out = fetch_external_data.invoke({"user_id": "1001", "month": "2025-01"})
    print("[6] fetch_external_data('1001', '2025-01') ->", repr(out)[:80] + ("..." if len(repr(out)) > 80 else ""))

    # 3. RAG 工具（需配置 DASHSCOPE_API_KEY 与知识库）
    print("\n[7] rag_summarize (可选，依赖 API 与向量库)")
    if not init_dashscope_api_key():
        print("    未配置 DASHSCOPE_API_KEY，已跳过")
    else:
        try:
            ans = rag_summarize.invoke({"query": "扫地机器人如何保养？"})
            print("    ->", ans[:120] + "..." if len(ans) > 120 else ans)
        except Exception as e:
            print("    失败:", e)

    print("\n=== 自测结束 ===")

运行

python -m zhisaotong_agent.agent.tools.agent_tools

结果

2026-03-16 08:47:17,190 - zhisaotong_agent.utils.api_key - INFO - api_key.py:49 - DashScope API Key 已成功加载并设置到环境变量。
=== agent_tools 自测 ===

[1] get_weather('北京') -> 城市北京天气为晴天，气温26摄氏度，空气湿度50%，南风1级，AQI21，最近6小时降雨概率极低
[2] get_user_location() -> 合肥
[3] get_user_id() -> 1004
[4] get_current_month() -> 2025-05
[5] fill_context_for_report() -> fill_context_for_report已调用
2026-03-16 08:47:17,199 - __main__ - INFO - agent_tools.py:121 - 外部数据已加载，路径=/home/devbox/project/zhisaotong_agent/data/external/records.csv，用户数=10
[6] fetch_external_data('1001', '2025-01') -> '特征: 65㎡公寓 | 单身 | 木地板\n效率: 覆盖率:85%\\n日均清扫:45㎡\\n漏扫区域:沙发底部（高度不足）\n耗材: 主刷寿命:剩余60天\...

[7] rag_summarize (可选，依赖 API 与向量库)
2026-03-16 08:47:17,200 - zhisaotong_agent.utils.api_key - INFO - api_key.py:49 - DashScope API Key 已成功加载并设置到环境变量。
2026-03-16 08:47:17,540 - zhisaotong_agent.rag.rag_service - INFO - rag_service.py:132 - RagSummarizeService 初始化完成。
2026-03-16 08:47:18,125 - zhisaotong_agent.rag.rag_service - INFO - rag_service.py:153 - 向量检索完成，query='扫地机器人如何保养？', 命中文档数=3
2026-03-16 08:47:26,229 - zhisaotong_agent.rag.rag_service - INFO - rag_service.py:197 - RAG 总结调用成功，query='扫地机器人如何保养？', answer_len=215
    -> 扫地机器人保养需每日使用后用干软布擦拭机身外壳，清理防撞条缝隙的毛发和线头；每次清扫后清理主刷、尘盒及拖布（宠物家庭需全面清理并每周除味消毒）；每日检查边刷旋转是否卡顿，及时停机清理以防电机烧毁；每周检查充电座电源适配器有无发热或异响；每月...

=== 自测结束 ===