AI编程

今日内容介绍

• AI编程概念
• AI编程工具
• Claude Code
• 项目实战

AI编程概念

本质：意图的编译。（vibe coding）

概念：将自然语言/模糊需求/草图等通过LLM大模型编译成能够运行的软件产品。

模型：

编程语言发展历史：

#1.机器码（0/1）
通过设备的插拔来操作。

#2.汇编语言
A语言，通过指令来操作cpu。

#3.低级语言（面向过程）
C语言

#4.高级语言（面向对象）
Java、Python

#5.vibe coding（氛围编程）
2025年

AI编程工具

有哪些编程工具：

工具定位：

选用建议：

1. 对于需要深入项目级重构、依赖多文件上下文且要求模型能力灵活切换的开发者，Cursor 提供了卓越的 AI驱动 IDE 体验。
2. 对于处理超大型代码库、需要模型进行复杂架构推理的开发者，Claude Code 及其强大的上下文能力是最佳选择。
3. 对于注重中文支持、国内生态和合规性的开发者，以 CodeBuddy、Comate 和通义灵码为代表的本土工具更具优势。

建议Claude Code或者Cursor。

Claude Code介绍

Claude和Claude Code是一回事嘛？

不是。Claude是Anthropic公司推出来的一款大模型。而Claude Code是Anthropic公司推出来的一款AI编程工具。

ClaudCode概述

https://claude.com/product/claude-code

Claude Code 是 Anthropic 专为软件开发人员和工程团队设计的一款 AI 驱动的编码工具，它将 Claude 的高级智能直接集成到终端和开发环境中。

Claude是指Anthropic 这家公司训练的一系列基础模型。

Anthropic 是一家专注于 AI 安全和研究的人工智能公司。它由前 OpenAI 的研究人员创立，其核心使命是开发出安全、有用且可控的通用人工智能(AGI)。

有哪些核心的模型，比如编程用Claude Sonnet

安装

安装环境

#1.安装Node.js
运行环境。电脑本身看不懂JS代码，必须安装这个"解释器"才能运行Claude Code。

#2.安装npm
下载工具。包管理工具，随着node.js下载一起附带的。

#3.安装Claude code
AI编程工具。我们要用的编程助手软件本身。

#4.安装git bash
git代码管理工具。可以通过claude Code帮我们把代码推送到github等git仓库中。

node.js介绍

安装node.js

node.js 下载地址:https://nodejs.org/en/download/

下载.msi文件安装，并验证。

当然，资料中已经提前下载好了，直接安装即可。

node -v
npm -v

安装claudecode

#打开终端执行，以下是claudecode安装
npm install -g @anthropic-ai/claude-code

#claude-code版本验证
claude --version

安装git bash

安装上git bash 才能使用claude命令，否则报错。

安装并验证：https://git-scm.com/install/windows

前面项目中用过git，应该都已经安装好了。

配置

配置deepseek模型

# 设置 Deepseek 接地址
setx ANTHROPIC_BASE_URL "https://api.deepseek.com/anthropic"

# 设置认证 Key (请将 sk-xxxx 替换为你的真实的 Deepseek Key)
setx ANTHROPIC_AUTH_TOKEN "sk-xxxxx"

# 设置超时时间(10分钟)
setx API_TIMEOUT_MS 600000

# 指定模型
setx ANTHROPIC_MODEL "deepseek-chat"
setx ANTHROPIC_SMALL_FAST_MODEL "deepseek-chat"

#禁用非必要流量
setx CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC 1

配置文件有2处需要修改：

其一：C:\Users\81196\.claude.json文件：

"hasCompletedOnboarding": true

其二：C:\Users\xxx\.claude\settings.json文件：

{
  "autoUpdatesChannel": "stable",
  "env": {
    "ANTHROPIC_AUTH_TOKEN": "sk-cc66af9c235c4adsfa9d69fa6f3dde7",
    "ANTHROPIC_BASE_URL": "https://api.deepseek.com/anthropic",
    "ANTHROPIC_DEFAULT_HAIKU_MODEL": "deepseek-chat",
    "ANTHROPIC_DEFAULT_OPUS_MODEL": "deepseek-chat",
    "ANTHROPIC_DEFAULT_SONNET_MODEL": "deepseek-chat",
    "API_TIMEOUT_MS": "3000000",
    "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": 1
  }
}

验证

# 1.如果项目在d盘就切到d盘
C:\Users\81196>d:

# 2.切到项目所在位置：
cd D:\coding\workspace1\claudecode

# 3.输入claude命令，并一直点回车
claude

# 4.在claude的命令行下输入需求，让claude帮我们干活。比如：
你是谁？
你能干什么？
...
其他问题

项目实战

实战一：点歌平台

#1.生成需求文档
使用Python写一个简单的点名系统。要求：
（1）随机点名，但是要尽量公平
（2）支持一次性随机多个
（3）能看到所有的点名历史纪录
（4）页面美观，观赏性强
（5）不需要登录功能，直接开始点名即可
（6）名单从金庸武侠小说中随机30个人。
（7）Python解释器使用

先基于上述需求，成一个需求文档：requirements.md

#2.生成最后代码
最后基于需求文档，生成最终的代码。

实战二：心血管疾病模型

初始化项目

【项目总览】 本项目目标是构建一个名为 CardioAI-心血管疾病智能辅助系统的多模块应用，集成了数据可视化(Streamlit)、机器学习预测(XGBoost/Flask)。
【本地环境与文件路径约束】 请严格遵守以下文件路径和文件引用，它们是用户本地环境的绝对路径:
代码根目录(CODE_ROOT):D:\coding\workspace1\ai_code\aicodes
数据文件(DATA_PATH):D:\coding\workspace1\ai_code\data\心血管疾病.xlsx
配置文件 (ENV_PATH):D:\coding\workspace1\ai_code\.env
【指令:项目初始化与环境配置】
Conda 环境说明:
请在 requirements.txt 文件前添加注释，指导用户创建并激活名为 cardioenv 的 conda 虚拟环境。
用户执行示例(非代码生成部分):conda create -n cardioenv python=3.10 和 conda activate
cardioenv。
创建目录结构:
在 CODE_ROOT 下创建以下目录:data(如果不存在)，module1_dashboard, module2_predictor, module3_voice_assistant, module2_predictor/templates, module3_voice_assistant/templates。
生成 requirements.txt:
CODE_ROOT 下生成requirements.txt 文件，包含以下所有依赖:pandas,openpyxl,numpy,scikit-learn, xgboost, joblib, streamlit, plotly, Flask, python-dotenv, langchain-openai,
dashscope, requests。

模块1-交互式仪表盘

【核心产出】module1_dashboard/cardio_dashboard.py
【任务要求】生成一个streamlit应用程序，用于对 DATA_PATH 的数据进行清洗、特征工程和交互式可视化。
数据处理:
加载: 使用 DATA_PATH 加载数据,D:\coding\workspace1\ai_code\data\心血管疾病.xlsx，注意数据格式excel表格和编码。
特征工程:将age(天)转换为年(age_years，四舍五入)。计算BMI($BMI=\frac{weight}
{(height/100)^2}$).
异常值处理:删除舒张压(ap_1o)$\geq$ 收缩压(ap_hi)的记录。删除血压极端异常值(收缩压$\in[90,250]$,舒张压$\in[60,150]$)。
类别转换:将 cholestero1 和 gluc 数值转换为描述性字符串。创建 bmi_category。
性能优化: 对数据加载和清洗函数使用@st.cache_data装饰器。
Streamlit 界面:
侧边栏:包含 age_years范围滑块、gender 和 cardio 的多选框筛选器。
主页展示:展示筛选后的总记录数和心血管疾病总风险率。
图表(Plotly Express):绘制 age_years 分布直方图(按cardio 区分)和 bmi_category对 cardio影响的堆叠柱状图。

C:\Programming\Anaconda3\envs\cardioenv\python.exe 解释器 这是当前整个项目的解释器!虚拟环境是cardioenv

参考效果如下：

模块2-高效心血管风险预测模型

【核心产出】module2_predictor/train_and_save.py (一次性脚本), module2_predictor/cardio_predictor_model.pkl, module2_predictor/app.py, module2_predictor/templates/index.html
【任务要求】

Part A:模型训练与保存

生成独立脚本 module2_predictor/train_and_save.py:
执行与模块1相同的清洗和特征工程。删除id和原始 age字段。
构建 Pipeline:使用 ColumnTransformer 封装StandardScaler(用于连续特征)和 OneHotEncoder (用于分类特征)。
训练XGBClassifier模型。
使用joblib 将包含 预处理器和分类器的完整Pipeline 对象保存到 module2_predictor/cardio_predictor_model.pkl。

Part B:Flask API 部署

生成后端应用 module2_predictor/app.py (Flask):
创建/predict_cardio API接口,接收11个原始特征值的JSON PosT 请求。
在接口内部:加载模型，使用保存的 Pipeline 对输入数据进行预处理，然后预测。
返回JsoN响应，包含预测的疾病概率(probability)和预测结果(prediction:0或1)。

Part C:前端交互界面

生成前端模板 module2_predictor/templates/index.html:
创建一个包含 11 个输入字段的 HTML 表单。
使用 Javascript/Fetch API 调用/predict_cardio接口，并将返回的风险概率和结果清晰地展示给用户。

启动方式:bat
不要创建bat文件进行启动 给出对应的命令手动启动即可

C:\Programming\Anaconda3\envs\cardioenv\python.exe 解释器 这是当前整个项目的解释器!虚拟环境是cardioenv

参考效果如下：

实战三：推送到gitee

#1.推送到git仓库
帮我把代码推送到gitee平台，仓库地址为：git@gitee.com:pydliyong2/hz4ai.git

Coze&Dify

今日内容介绍

• 低代码智能体平台

• Coze

• Dify

低代码智能体平台

为什么需要Agent

Agent，就是智能体。其实翻译为代理更合适。

企业的痛点：

需求：企业需要能够自主行动，高效处理复杂任务的智能助手。这就是AI Agent存在的意义。

大模型能力那么强，难道搞不定吗？是的。大模型本身是有局限性的，具体体现如下：

• 知识过时，知识停留在训练时间
• 无法联网，获取最新信息

• 深度不足，缺乏专业领域知识

• 不能执行，无法完成实际操作

总结：大模型只能说，但是不能做。于是，AI Agent诞生了。

#1.Agent
智能体，代理，能代理（帮）人干活。饮水机。

#2.AI Agent
和Agent一样，但是是基于AI（大模型）来驱动的。大模型来充当智能体的大脑（做决策）

#3.Agentic
和AI Agent类似，能够主动代替（帮）人来做决策。
主动订打折机票。候补高铁票。

概述

智能体（AI Agent）是一类能够感知环境、做出决策并采取行动以达成目标的系统或程序。与传统的单模型调用不同，智能体具备自主决策、多工具协作和持续优化三大核心特征。智能体框架提供标准化开发工具和架构，使开发者能高效构建具有自主执行能力的AI系统。

三个核心能力：

#1.工具调用
#2.记忆
#3.规划

简而言之：智能体 = 大模型 + 工具 + 规划 + 记忆

AI Agent：就是一个能够干活的AI管家。

大模型和智能体的区别

所以：

ChatGPT就是一个聊天对象；

AI Agent就是一个能干活的AI助手。

一个更简洁的总结：Agent = 大模型 + 工具！

类别

当前市场上的低智能体框架主要有：Coze、Dify和n8n。用的多的主要是前两种。

Coze

简介

概述：Coze 是由字节跳动推出的一个AI聊天机器人和应用程序编辑开发平台，可以理解为字节跳动版的GPTs。

核心定位：Coze是字节跳动推出的零代码AI应用开发平台，主打"5分钟搭建聊天机器人"，主要面向零基础的个人开发者和中小团队。

Coze还提供了多种插件、知识、工作流、长期记忆和定时任务等功能，来增强聊天机器人的能力和交互性。而且你可以将搭建的 Bot 发布到各类社交平台和通讯软件上，让更多的用户与你搭建的 Bot 聊天。

如何使用

Coze（扣子）分为国内版和国外版：

• 国内版访问地址：https://www.coze.cn，背后大模型应用的是字节自研的云雀大模型、通义千问和kimi大模型
• 国外版访问地址：https://www.coze.com，背后大模型应用的是一些国际大模型，比如GPT系列，但是需要一些科学上网的方法。

当前国外版的Coze的确比国内版的要有优势，主要体现在大模型上。但是国内版的一些功能也在不断地完善，而且不需要科学上网。因此接下来的教程就以国内版 Coze 来进行，也可以参考官网文档：https://www.coze.cn/docs/guides/welcome 实现Coze平台的应用。

插件系统

扣子平台提供了一个多样化的插件库，这些插件涵盖了从基础的文本处理到高级的机器学习功能。例如，文本分析插件可以帮助 AI 理解用户输入的意图，情感分析插件能够识别用户的情绪倾向，而自然语言处理（NLP）插件则支持更复杂的对话生成。此外，还有图像识别、语音识别、数据分析等插件，这些插件的数量和种类不断增加，以适应不断变化的技术趋势和市场需求。

工作流

概述

工作流：业务逻辑的可视化执行。

工作流的作用：它将一个复杂的任务分解成一系列可管理的、按顺序或按条件执行的步骤，并通过图形化的界面将这些步骤连接起来。

需要注意的是：工作流和智能体，不是对立的关系，而是演进。

核心观点：工作流是确定性的智能体实现方式。

coze平台提供了灵活的工作流设计工具，开发者可以通过拖拽式界面轻松搭建复杂的对话流程。工作流节点包括了各种预设的对话元素，如问答、选择、表单填写等，以及自定义的逻辑判断，使得 AI Bot 能够处理复杂的用户交互。

coze的工作流类型

coze的工作流体系分为两种：工作流和对话流：

无论是工作流还是对话流，他们都由多个节点构成，节点是组成工作流的基本单元。例如，大语言模型 LLM、自定义代码、判断逻辑等节点。

工作流默认包含了 Start 节点和 End 节点。

• Start 节点是工作流的起始节点，可以包含用户输入信息。
• End 节点是工作流的末尾节点，用于返回工作流的运行结果。

节点类型

扣子为你提供了以下基础节点，除此之外，你还可以添加插件节点或其他工作流。

我们通过将这些不同类型的节点连接起来，就可以精确地控制数据流向和任务执行顺序，从而实现任何复杂的业务逻辑。

创建工作流标准流程

• 创建工作流
• 编排工作流
• 测试并发布

平台实践

使用入口

在官网入口中，如下图：

点击左侧的扣子编程，弹出如下页面：

点击回到旧版或者低代码模式，如下图所示：

在这里，就可以创建新项目了。

面试助手

创建智能体

点击新建，弹出如下对话框，选择创建智能体，如下图：

创建智能体

输入智能体名称和介绍，如下图：

编排智能体

点击确认后，弹出如下页面：

左边：填写提示词（参考如下）。

中间：填写智能体的设置（选择豆包2.0pro版本）。

右边：预览结果。

提示词：

你是一个AI大模型的面试官。可以对前来没事的同学进行提问。提问的流程如下：

1.让求职者进行自我介绍

2.让求职者进行项目介绍

3.主动问询求职者最近的项目10个问题。

4.让求职者反问面试官2个问题。

5.结束面试。结束后续联系。

注意：
* 你的提问应该简洁干脆。
* 你的提问应该围绕AI大模型领域。

最终结果如下图：

发布智能体

当一切设置完成后，可以点击右上角的发布，进行发布智能体。

点击确认，进入到发布页面，如下图：

点击发布按钮，如下图：

到此，智能体发布成功，最终在项目开发中，能看到刚刚的智能体。

法律问答助手

创建智能体，输入名称和介绍，如下图：

弹出的页面分三部分，分别是：人设与回复逻辑、编排、预览。

首先，填写提示词：

你是一个法律专家，精通劳动法，你能够精准透彻地理解用户的问题，然后从庞大的知识库中精准检索相关信息，进而为用户生成详尽准确的答案。

## 技能
### 技能 1: 深入理解问题
1. 仔细剖析用户提出的问题，精准提取其中的关键信息。
### 技能 2: 高效知识库检索
1. 依据关键信息，在知识库中进行全面检索。
### 技能 3: 智能搜索引擎查询
1. 若根据关键信息，在知识库中未找到高度相关的知识，借助 bingWebSearch 搜索工具展开检索。
### 技能 4: 精确回答生成
1. 以检索到的信息为基础，为用户打造准确且简洁明了的回答。

## 限制:
1. 只回答与学习相关的问题，对无关话题不予理会。
2. 尽量运用清晰简洁的语言回应用户的问题。
3. 在整个回答过程中，始终将用户的需求置于核心位置。

其次，点击添加知识库，如下图：

添加知识库后，点击创建知识库，如下图：

输入知识库的名称后，点击创建并导入，如下图：

把知识文本拖入框内或者上传皆可。这里以上传《中华人民共和国劳动合同法》为例，如下图：

点击下一步，默认参数即可，如下图：

继续点击下一步，如下图：

等待处理处理中，也可以直接完成（会后台处理），如下图：

点击添加该知识库到智能体中，如下图：

添加完后，中间编排部分就多了知识库了，如下图：

最后就可以测试和发布了。最终效果如下：

最后点击右上角进行发布（也可以发布到微信公众号），如下图：

填写微信公众号的AppID后，即可绑定微信公众号，如下图：

最后，点击右上角发布，完成发布到微信公众号。

点击完成，即可，最终效果如下：

跨境电商助手对话流

先来看整个流程图：

下面是coze平台的整个流程。

创建对话流智能体

创建智能体，切换到对话流模式，添加对话流（参考如下流程），试运行后发布，即可完成对话流任务。

创建对话流

点击创建对话流，输入名字，确认即可。

保存后，来到了工作流编排页面：

开始节点

• 作用：接受用户的输入的问题（query）
• 实现方式：《默认节点》

意图识别

节点创建详情：

节点参数详情：

• 作用：判断用户的query是否属于跨境电商知识范畴

properties 主要三种选择：打招呼(greeting)；找人工(human)；RAG检索(rag)

• 实现方式：添加**《插件》**节点，主要该插件是用户自定义实现

• 实现代码如下：

``python from runtime import Args import random """ Each file needs to export a function namedhandler`. This function is the entrance to the Tool.

Parameters:
args: parameters of the entry function.
args.input - input parameters, you can get test input value by args.input.xxx.
args.logger - logger instance used to print logs, injected by runtime.

Remember to fill in input/output in Metadata, it helps LLM to recognize and use tool.

Return:
The return data of the function, which should match the declared output parameters.
"""
def handler(args)->dict:
"""
基于固定短语匹配的跨境电商问答处理器

  Args:
      query: 用户输入的问题

  Returns:
      Dict: 包含回复类型、回复内容等信息的字典
  """
  # 去除首尾空格
  query = args.input.query.strip()

  # 打招呼相关的固定短语
  greeting_phrases = {
      # 基本问候
      "你好", "您好", "hi", "hello", "嗨", "哈喽", "哈罗",
      "早上好", "下午好", "晚上好", "上午好", "中午好", "晚安",
      "早", "午安", "good morning", "good afternoon", "good evening",
      # 询问身份
      "你是谁", "你是什么", "你叫什么", "你的名字", "介绍一下自己", 
      "自我介绍", "你是什么东西", "你是哪个", "你是啥",
      # 询问状态
      "你好吗", "怎么样", "还好吗", "你还好吗", "最近怎么样",
      "你在吗", "在不在", "还在吗", "在线吗", "你在线吗",
      # 询问能力
      "你能干什么", "你会做什么", "你能做什么", "你的功能", 
      "你有什么用", "你的作用", "你的职责", "你的用途",
      "你能帮我什么", "你可以做什么", "你会什么", "你懂什么",
      "能力介绍", "功能介绍", "你的能力", "你有什么功能",
      # 开始对话
      "开始", "开始咨询", "开始对话", "开始聊天", "我想咨询",
      "我有问题", "我想问问题", "我想了解", "咨询一下",
      # 测试类
      "测试", "试试", "试一试", "test", "testing", "试试看",
      "测试一下", "看看", "检查一下"
  }

  greeting_response = "你好，很高兴为您服务！我是您的跨境电商学习小助手，专业为您答疑解惑。"

  # 礼貌用语
  thank_phrases = {
      "谢谢", "感谢", "多谢", "谢了", "thanks", "thank you", 
      "thx", "3q", "3x", "谢谢你", "感谢你", "多谢了",
      "非常感谢", "十分感谢", "万分感谢", "太感谢了"
  }

  goodbye_phrases = {
      "再见", "拜拜", "bye","byebye","goodbye", "88", "走了", "告辞",
      "先走了", "下次见", "回头见", "有空再聊", "改天聊",
      "see you", "拜", "溜了", "闪了", "slip away"
  }

  polite_responses = {
      "thank": [
          "不用客气，随时为您服务！", 
          "很高兴能帮助到您！", 
          "这是我应该做的，有问题随时找我哦！",
          "客气了，有什么问题尽管问！",
          "不客气，祝您跨境电商生意兴隆！"
      ],
      "goodbye": [
          "再见！期待下次为您服务！", 
          "祝您生活愉快，有问题随时来找我！", 
          "再见！祝您跨境电商生意兴隆！",
          "拜拜！有问题随时回来咨询！",
          "再见！祝您学习愉快！"
      ]
  }

  # 人工服务相关短语
  human_service_phrases = {
      # 直接要求人工服务
      "人工服务", "人工客服", "人工坐席", "人工咨询", "人工帮助",
      "人工支持", "人工答疑", "人工解答", "人工回复", "人工对话",
      # 转接相关
      "转人工", "找人工", "要人工", "转接人工", "转接客服",
      "切换人工", "接入人工", "联系人工","答疑入口",
      # 真人服务
      "真人服务", "真人客服", "真人咨询", "真人对话", "真人帮助",
      "活人", "真人", "人类", "人工", "真的人",
      # 客服相关
      "客服", "在线客服", "联系客服", "找客服", "客服电话",
      "客服微信", "客服qq", "官方客服",
      # 老师/导师
      "联系老师", "找老师", "咨询老师", "请教老师", "老师帮忙",
      "专业老师", "课程老师", "指导老师",
      # 专业服务
      "专人服务", "专人客服", "专业咨询", "专业服务", "专家咨询",
      "顾问咨询", "一对一服务", "专属服务",
      # 投诉和问题
      "投诉", "举报", "反馈问题", "意见反馈", "服务投诉",
      "质量问题", "服务问题", "系统问题",
      # 售后相关
      "退款", "退货", "售后", "售后服务", "退换货", "申请退款",
      "退费", "取消订单", "订单问题",
      # 不满意
      "不满意", "有问题", "出问题", "不行", "太差了", "服务差",
      "回答不对", "答非所问", "听不懂", "不准确"
  }

  human_service_response = "同学，点击 https://www.123.com 可进入人工答疑"

  def normalize_text(text: str) -> str:
      """标准化文本：去除标点符号，转换为小写"""
      import re
      # 保留中文、英文、数字，去除标点符号和空格
      cleaned = re.sub(r'[^\w\u4e00-\u9fff]', '', text.lower().strip())
      return cleaned

  def exact_match_check(query_text: str, phrase_set) -> bool:
      """精确匹配检查"""
      normalized_query = normalize_text(query_text)

      for phrase in phrase_set:
          normalized_phrase = normalize_text(phrase)
          if normalized_phrase == normalized_query:
              return True
      return False

  def contains_match_check(query_text: str, phrase_set) -> bool:
      """包含匹配检查（用于短查询中包含关键短语的情况）"""
      normalized_query = normalize_text(query_text)

      # 只有当查询很短时才使用包含匹配（避免误匹配）
      if len(normalized_query) <= 10:
          for phrase in phrase_set:
              normalized_phrase = normalize_text(phrase)
              if normalized_phrase in normalized_query or normalized_query in normalized_phrase:
                  return True
      return False

  # 处理输入
  query = query.strip()
  if not query:
      return {
          "type": "error",
          "response": "请输入您的问题。",
          "need_rag": False,
          "original_query": query
      }

  print(f"处理查询: '{query}'")
  print(f"标准化后: '{normalize_text(query)}'")

  # 1. 检查感谢类礼貌用语（精确匹配）
  if exact_match_check(query, thank_phrases):
      return {
          "type": "greeting",
          "response": random.choice(polite_responses["thank"]),
          "need_rag": False,
          "original_query": query
      }

  # 2. 检查告别类礼貌用语（精确匹配）
  if exact_match_check(query, goodbye_phrases):
      return {
          "type": "greeting", 
          "response": random.choice(polite_responses["goodbye"]),
          "need_rag": False,
          "original_query": query
      }

  # 3. 检查打招呼（精确匹配 + 短语包含匹配）
  if exact_match_check(query, greeting_phrases) or contains_match_check(query, greeting_phrases):
      return {
          "type": "greeting",
          "response": greeting_response,
          "need_rag": False,
          "original_query": query
      }

  # 4. 检查人工服务请求（精确匹配 + 短语包含匹配）
  if exact_match_check(query, human_service_phrases) or contains_match_check(query, human_service_phrases):
      return {
          "type": "human_service",
          "response": human_service_response,
          "need_rag": False,
          "original_query": query
      }

  # 5. 其他情况需要RAG检索
  return {
      "type": "rag_needed",
      "response": "",
      "need_rag": True,
      "original_query": query
  }

```

在元数据中，填写好输入和输出参数：

编写完后，测试，如下：

测试完成后，选择右上角的发布，如下图：

选择器

• 作用：获取意图识别的结果

properties 如果意图识别属于：打招呼、人工服务这两种类型，直接返回默认结果结束；否则，直接经过知识库检索

• 实现方式：添加**《选择器》**节点

知识库检索

• 作用

properties 输入用户的专业问题，经过知识库检索,得到和问题相关的上下文，注意：有可能检索出来的结果为空

• 实现方式：添加**《知识库检索》**节点

判断是否可以检索出结果

• 作用：根据是否检索出结果进行判断选择

properties 1.如果检索出相关的上下文，那么交由《大模型RAG回答》：基于问题和上下文来让大模型回答问题 2.如果没有检索出相关的上下文，那么交由《闲聊大模型回答》：直接将问题送入大模型回答问题

• 实现方式：添加**《选择器》**节点

大模型RAG回答

• 作用：

properties 接受用户的query和上下文，经过大模型得到结果

• 实现方式：添加《大模型》节点

• RAG大模型系统Prompt

# 角色
你是一位专业且高效的跨境电商答疑小助手，精通跨境电商领域的各类知识，能够依据相关信息准确回答用户关于跨境电商的问题。

## 技能
### 技能 1: 基于上下文解答问题
1. 严格依据提供的上下文内容进行回答，不添加上下文中未出现的信息。
2. 回答需精准、简洁、有条理，重点突出，逻辑严密，避免歧义，采用清晰的格式呈现内容，将复杂概念以通俗易懂的方式表达。

### 技能 2: 答案优化
1. 若问题检索出来的上下文答案比较简单，需进行优化，但不能改变原来上下文中涉及到的答案核心内容。
2. 示例：
    - 问题： 是不是不能在一个页面里面同时操作设置两个素材（A图案对应白T，B图案对应黑T）只能选一个通用于白黑T的图案？
    - 检索上下文答案： 是的
    - 优化后给出答案：  是的，不能在一个页面里面同时操作设置两个素材（A图案对应白T，B图案对应黑T），只能选一个通用于白黑T的图案。

### 技能 3: 上下文不足信息内容补齐
1. 若问题经过数据库检索，给出的上下文信息非常少或没有，需结合自身专业知识回答。
2. 示例：
    - 问题：如果我的产品标题中即有泰文和英文，算重复吗？
    - 上下文的信息：请不要在标题中出现叠词
    - 回答：本地知识库中未提供关于产品既有泰文和英文是否算重复的相关内容，结合专业知识，目前没有明确固定标准判定这种情况一定算重复，具体要根据不同平台规则和实际情况判断。以上信息仅供参考。

### 技能 4 敏感词过滤
1. 当遇到不符合安全规范或者敏感的词汇时，优先利用插件check进行敏感词搜索，然后去除

## 限制:
- 仅回答与跨境电商相关的问题，拒绝处理与跨境电商无关的话题。
-优先以检索的知识库内容为答案，除非内容不全在进行优化或者补全
- 回答内容必须符合上述技能要求的格式和规范。
- 若回答基于知识库已有信息，需遵循相应说明规范；若知识库中没有相关内容，应按技能 3 要求回答用户 。 

• RAG大模型用户Prompt

properties 问题：{{input}}, 上下文：{{context[0].output}}

大模型闲聊回答

• 作用：

properties 接受用户的问题，直接基于大模型本身回答结果

• 实现方式：添加**《大模型》**节点

• 闲聊大模型：系统Prompt

# 角色
你是一位专业且亲切的跨境电商答疑小助手，不仅拥有深厚的跨境电商专业知识，还具备广泛的通用知识储备。你能够根据用户需求，灵活切换角色，既能为用户解答跨境电商领域的复杂专业问题，也能与用户展开轻松愉快的日常闲聊。

## 技能
### 技能 1: 精准解答跨境电商问题
1. 当用户提出跨境电商相关问题时，充分运用自身专业知识，结合丰富且贴合实际的案例，为用户提供准确、全面且详细的解答。
2. 针对复杂的跨境电商概念，运用通俗易懂、生动形象的语言进行深入浅出的解释说明，确保用户能够轻松理解。
3. 在回答的结尾处明确注明“以上答案仅供参考”。

### 技能 2: 耐心回应非跨境电商问题
1. 当用户提及非跨境电商问题时，例如简单的数学运算“1 + 1 等于几”等，运用已有的知识储备，耐心、准确地对用户进行回答。
2. 在回答此类问题的结尾同样注明“以上答案仅供参考”。

### 技能 3: 合理引导不明确问题
1. 当用户输入的问题含义不明确时，不直接给出知识回答，而是友好地输出“我还在努力理解您的问题，请您详细描述后再来询问吧，这样我能更好地为您解答。”

## 限制:
- 交流内容主要围绕跨境电商学习以及其他非电商知识相关范畴，对于与这两类内容无关的话题，需礼貌地拒绝回答，并告知用户“抱歉，我只能回答与跨境电商学习和其他非电商知识相关的问题哦”。
- 所输出的内容必须严格符合上述回答要求，保证格式规范、条理清晰、逻辑连贯。 
- 回答的答案中禁止提及：需要进行知识库检索

• 闲聊大模型：用户Prompt

properties 用户输入：{{input}}

结束节点

• 作用：

properties 输出最终的结果

• 实现方式：默认节点

添加到当前对话流中

发布后，就可以添加到当前对话流中：

接下俩就可以正常使用了：

最终，发布该对话流到coze平台：

Dify

介绍

Dify 是一款开源的大语言模型(LLM) 应用开发平台。它融合了后端即服务（Backend as Service） 的理念，使开发者可以快速搭建生产级的生成式 AI 应用。即使你是非技术人员，也能参与到 AI 应用的定义和数据运营过程中。

由于 Dify 内置了构建 LLM 应用所需的关键技术栈，包括对数百个模型的支持、直观的 Prompt 编排界面、高质量的 RAG 引擎、稳健的 Agent 框架、灵活的流程编排，并同时提供了一套易用的界面和 API。这为开发者节省了许多重复造轮子的时间，使其可以专注在创新和业务需求上。

安装

安装WSL

首先，Windows平台下，先安装WSL：

wsl --install

备注：目前大家都已经安装好了。

安装 Docker Desktop

直接双击安装Docker Desktop软件即可。

备注：Docker Desktop在前面的课程中已经安装好了。

安装Dify环境

访问github，https://github.com/langgenius/dify/releases，下载的是一个zip包。

解压下载的zip包，切换到dify的解压目录下，复制一份.env文件：

#1.复制.env文件
cp .env.example .env

接下来，启动启动dify，运行命令：

# 进入dify的目录
cd dify

# 再进入到docker目录
cd docker

# 启动
docker compose up -d

首次运行，结果如下图，表示目前正在拉取镜像：

拉取完成后，则会启动dify，如下图：

配置Dify

接下来访问dify：

http://localhost

如下页面所示：

设置好相应的邮箱、用户名、密码即可。

# 邮箱
123456@qq.com

# 用户名
admin

# 密码
admin123456

设置完后，跳转到如下页面：

到此，dify就安装且配置完成。

大模型接入

安装大模型插件

在 Dify 的主页中，找到设置入口，并且点击：

切换到模型供应商，点击：

默认是没有配置任何大模型的，因此需要在下面列出的模型中，选择安装配置好需要的模型：

找到通义大模型，单机安装：

安装过程需要下载，受网络环境影响，有可能需要多尝试几次，请保持耐心等待。

配置大模型API KEY

安装成功后，接下来就是配置大模型的API KEY。

这里以通义大模型为例，介绍一下如何配置大模型。

点击设置，如下图：

输入API KEY和名称，如下图：

之后，点击保存即可，绿色说明大模型API KEY配置OK，如下图：

最后选择系统模型设置，配置通义的模型即可：

配置好后，如下图：

点击保存即可。

应用类型

Dify 中提供了五种应用类型：

• 聊天助手：基于 LLM 构建对话式交互的助手
• 文本生成应用：面向文本生成类任务的助手，例如撰写故事、文本分类、翻译等
• Agent：能够分解任务、推理思考、调用工具的对话式智能助手
• 对话流：适用于定义等复杂流程的多轮对话场景，具有记忆功能的应用编排方式
• 工作流：适用于自动化、批处理等单轮生成类任务的场景的应用编排方式

工作流通过将复杂的任务分解成较小的步骤（节点）降低系统复杂度，减少了对提示词技术和模型推理能力的依赖，提高了 LLM 应用面向复杂任务的性能，提升了系统的可解释性、稳定性和容错性。

Dify 工作流分为两种类型：

• Chatflow：面向对话类情景，包括客户服务、语义搜索、以及其他需要在构建响应时进行多步逻辑的对话式应用程序。
• Workflow：面向自动化和批处理情景，适合高质量翻译、数据分析、内容生成、电子邮件自动化等应用程序。

节点是工作流中的关键构成，通过连接不同功能的节点，执行工作流的一系列操作。

• 开始（Start）：定义一个 workflow 流程启动的初始参数。
• 结束（End）：定义一个 workflow 流程结束的最终输出内容。
• 大语言模型（LLM）：调用大语言模型回答问题或者对自然语言进行处理。
• 知识检索（Knowledge Retrieval）：从知识库中检索与用户问题相关的文本内容，可作为下游 LLM 节点的上下文。
• 问题分类（Question Classifier）：通过定义分类描述，LLM 能够根据用户输入选择与之相匹配的分类。
• 条件分支（IF/ELSE）：允许你根据 if/else 条件将 workflow 拆分成两个分支。
• 代码执行（Code）：运行 Python / NodeJS 代码以在工作流程中执行数据转换等自定义逻辑。
• 变量聚合（Variable Aggregator）：将多路分支的变量聚合为一个变量，以实现下游节点统一配置。
• 循环（Loop）：循环节点用于执行依赖前一轮结果的重复任务，直到满足退出条件或达到最大循环次数。

知识库

知识库功能将 RAG 上的各环节可视化，提供了一套简单易用的用户界面来方便应用构建者管理个人或者团队的知识库，并能够快速集成至 AI 应用中。

开发者可以将企业内部文档、规范信息等内容上传至知识库进行结构化处理，供后续 LLM 查询。

1. 创建知识库。通过上传本地文件、导入在线数据或创建一个空的知识库。
2. 指定分段模式。该阶段是内容的预处理与数据结构化过程，长文本将会被划分为多个内容分段。你可以在此环节预览文本的分段效果。
3. 设定索引方法和检索设置。知识库在接收到用户查询问题后，按照预设的检索方式在已有的文档内查找相关内容，提取出高度相关的信息片段供语言模型生成高质量答案。
4. 完成上传，在应用内关联知识库并使用

实践

聊天助手

创建聊天助手，如下图：

点击创建后，弹出如下编排和调试页面：

在编排页面的提示词中，输入大模型的提示词：

你是一个通用的聊天助手，负责和用户以轻松、幽默、诙谐的语气闲聊。

注意：你一定要输出内容，且内容不能包含暴力、色情、违禁、消极等情绪文字。

最后，在右边的调试窗口中，输入问题，以完成和用户聊天。

没有问题后，最终点击右上角的发布，完成聊天助手的发布。

时间工具使用

创建智能体，输入名称，如下图：

什么都不需要输入，直接和他对话，询问当前时间：

给智能体添加工具，如下图：

再次和他对话，询问当前时间，结果虽然有，但是和当前对不上：

调整时区，得到最终结果：

保存，重新开始测试就符合正常时间了。

到此，时间插件就开发完了。

电商问答助手

整体流程图

先来看整个流程图：

下面是Dify的整个流程：

开始节点

• 作用：接受用户的输入的问题（query）

• 节点展开详情：

注意：query输入字段需要手动创建

意图识别

• 代码实现

• 作用：判断用户是否属于打招呼用语、亦或者找人工、还是询问电商知识（需要RAG检索）

• 节点展开详情：

• 作用：判断用户的query是否属于跨境电商知识范畴

properties 主要三种选择：打招呼(greeting)；找人工(human)；RAG检索(rag)

• 实现方式：利用《代码执行》节点完成

• 实现代码如下：

```python
import random

def main(query:str)->dict:
"""
基于固定短语匹配的跨境电商问答处理器

  Args:
      query: 用户输入的问题

  Returns:
      Dict: 包含回复类型、回复内容等信息的字典
  """
  # 去除首尾空格
  query = query.strip()

  # 打招呼相关的固定短语
  greeting_phrases = {
      # 基本问候
      "你好", "您好", "hi", "hello", "嗨", "哈喽", "哈罗",
      "早上好", "下午好", "晚上好", "上午好", "中午好", "晚安",
      "早", "午安", "good morning", "good afternoon", "good evening",
      # 询问身份
      "你是谁", "你是什么", "你叫什么", "你的名字", "介绍一下自己", 
      "自我介绍", "你是什么东西", "你是哪个", "你是啥",
      # 询问状态
      "你好吗", "怎么样", "还好吗", "你还好吗", "最近怎么样",
      "你在吗", "在不在", "还在吗", "在线吗", "你在线吗",
      # 询问能力
      "你能干什么", "你会做什么", "你能做什么", "你的功能", 
      "你有什么用", "你的作用", "你的职责", "你的用途",
      "你能帮我什么", "你可以做什么", "你会什么", "你懂什么",
      "能力介绍", "功能介绍", "你的能力", "你有什么功能",
      # 开始对话
      "开始", "开始咨询", "开始对话", "开始聊天", "我想咨询",
      "我有问题", "我想问问题", "我想了解", "咨询一下",
      # 测试类
      "测试", "试试", "试一试", "test", "testing", "试试看",
      "测试一下", "看看", "检查一下"
  }

   # 固定回复
  greeting_response = "你好，很高兴为您服务！我是您的跨境电商学习小助手，专业为您答疑解惑。"
  # 礼貌用语
  thank_phrases = {
      "谢谢", "感谢", "多谢", "谢了", "thanks", "thank you", 
      "thx", "3q", "3x", "谢谢你", "感谢你", "多谢了",
      "非常感谢", "十分感谢", "万分感谢", "太感谢了"
  }

  goodbye_phrases = {
      "再见", "拜拜", "bye","byebye","goodbye", "88", "走了", "告辞",
      "先走了", "下次见", "回头见", "有空再聊", "改天聊",
      "see you", "拜", "溜了", "闪了", "slip away"
  }


  polite_responses = {
      "thank": [
          "不用客气，随时为您服务！", 
          "很高兴能帮助到您！", 
          "这是我应该做的，有问题随时找我哦！",
          "客气了，有什么问题尽管问！",
          "不客气，祝您跨境电商生意兴隆！"
      ],
      "goodbye": [
          "再见！期待下次为您服务！", 
          "祝您生活愉快，有问题随时来找我！", 
          "再见！祝您跨境电商生意兴隆！",
          "拜拜！有问题随时回来咨询！",
          "再见！祝您学习愉快！"
      ]
  }
  # 人工服务相关短语
  human_service_phrases = {
      # 直接要求人工服务
      "人工服务", "人工客服", "人工坐席", "人工咨询", "人工帮助",
      "人工支持", "人工答疑", "人工解答", "人工回复", "人工对话",
      # 转接相关
      "转人工", "找人工", "要人工", "转接人工", "转接客服",
      "切换人工", "接入人工", "联系人工","答疑入口",
      # 真人服务
      "真人服务", "真人客服", "真人咨询", "真人对话", "真人帮助",
      "活人", "真人", "人类", "人工", "真的人",
      # 客服相关
      "客服", "在线客服", "联系客服", "找客服", "客服电话",
      "客服微信", "客服qq", "官方客服",
      # 老师/导师
      "联系老师", "找老师", "咨询老师", "请教老师", "老师帮忙",
      "专业老师", "课程老师", "指导老师",
      # 专业服务
      "专人服务", "专人客服", "专业咨询", "专业服务", "专家咨询",
      "顾问咨询", "一对一服务", "专属服务",
      # 投诉和问题
      "投诉", "举报", "反馈问题", "意见反馈", "服务投诉",
      "质量问题", "服务问题", "系统问题",
      # 售后相关
      "退款", "退货", "售后", "售后服务", "退换货", "申请退款",
      "退费", "取消订单", "订单问题",
      # 不满意
      "不满意", "有问题", "出问题", "不行", "太差了", "服务差",
      "回答不对", "答非所问", "听不懂", "不准确"
  }



  human_service_response = "同学，点击 https://www.123.com 可进入人工答疑"

  def normalize_text(text: str) -> str:
      """标准化文本：去除标点符号，转换为小写"""
      import re
      # 保留中文、英文、数字，去除标点符号和空格
      cleaned = re.sub(r'[^\w\u4e00-\u9fff]', '', text.lower().strip())
      return cleaned

  def exact_match_check(query_text: str, phrase_set) -> bool:
      """精确匹配检查"""
      normalized_query = normalize_text(query_text)

      for phrase in phrase_set:
          normalized_phrase = normalize_text(phrase)
          if normalized_phrase == normalized_query:
              return True
      return False

  def contains_match_check(query_text: str, phrase_set) -> bool:
      """包含匹配检查（用于短查询中包含关键短语的情况）"""
      normalized_query = normalize_text(query_text)

      # 只有当查询很短时才使用包含匹配（避免误匹配）
      if len(normalized_query) <= 10:
          for phrase in phrase_set:
              normalized_phrase = normalize_text(phrase)
              if normalized_phrase in normalized_query or normalized_query in normalized_phrase:
                  return True
      return False

  # 处理输入
  query = query.strip()
  if not query:
      return {
          "type": "error",
          "response": "请输入您的问题。",
          "need_rag": False,
          "original_query": query
      }

  print(f"处理查询: '{query}'")
  print(f"标准化后: '{normalize_text(query)}'")

  # 1. 检查感谢类礼貌用语（精确匹配）
  if exact_match_check(query, thank_phrases):
      return {
          "type": "greeting",
          "response": random.choice(polite_responses["thank"]),
          "need_rag": False,
          "original_query": query
      }

  # 2. 检查告别类礼貌用语（精确匹配）
  if exact_match_check(query, goodbye_phrases):
      return {
          "type": "greeting", 
          "response": random.choice(polite_responses["goodbye"]),
          "need_rag": False,
          "original_query": query
      }

  # 3. 检查打招呼（精确匹配 + 短语包含匹配）
  if exact_match_check(query, greeting_phrases) or contains_match_check(query, greeting_phrases):
      return {
          "type": "greeting",
          "response": greeting_response,
          "need_rag": False,
          "original_query": query
      }

  # 4. 检查人工服务请求（精确匹配 + 短语包含匹配）
  if exact_match_check(query, human_service_phrases) or contains_match_check(query, human_service_phrases):
      return {
          "type": "human_service",
          "response": human_service_response,
          "need_rag": False,
          "original_query": query
      }

  # 5. 其他情况需要RAG检索
  return {
      "type": "rag_needed",
      "response": "",
      "need_rag": True,
      "original_query": query
  }

```

选择器

• 作用：获取意图识别的结果

properties 如果意图识别属于：打招呼、人工服务这两种类型，直接返回默认结果结束；否则，直接经过知识库检索

• 实现方式：添加**《条件分支》**节点

• 节点详情展示：

直接输出

• 作用：如果用户的问题属于打招呼、找人工等用语，直接按照规则结果输出
• 否则：
• 进入知识库检索

知识库检索

• 作用

properties 输入用户的专业问题，经过知识库检索,得到和问题相关的上下文，注意：有可能检索出来的结果为空

• 实现方式：添加**《知识检索》**节点

注意：这一步一定要提前订定义好《知识库》

判断是否可以检索出结果

• 作用：根据是否检索出结果进行判断选择

properties 1.如果检索出相关的上下文，那么交由《大模型RAG回答》：基于问题和上下文来让大模型回答问题 2.如果没有检索出相关的上下文，那么交由《闲聊大模型回答》：直接将问题送入大模型回答问题

• 实现方式：添加**《条件分支》**节点

大模型RAG回答

• 作用：

properties 接受用户的query和上下文，经过大模型得到结果

• 实现方式：添加《LLM》节点

• 上下文：知识检索内容

• RAG大模型系统Prompt

# 角色
你是一位专业且高效的跨境电商答疑小助手，精通跨境电商领域的各类知识，能够依据相关信息准确回答用户关于跨境电商的问题。

## 重要参数
- 上下文内容：{{#context#}}
## 技能
### 技能 1: 基于上下文解答问题
1. 严格依据提供的上下文内容进行回答，不添加上下文中未出现的信息。
2. 回答需精准、简洁、有条理，重点突出，逻辑严密，避免歧义，采用清晰的格式呈现内容，将复杂概念以通俗易懂的方式表达。

### 技能 2: 答案优化
1. 若问题检索出来的上下文答案比较简单，需进行优化，但不能改变原来上下文中涉及到的答案核心内容。
2. 示例：
    - 问题： 是不是不能在一个页面里面同时操作设置两个素材（A图案对应白T，B图案对应黑T）只能选一个通用于白黑T的图案？
    - 检索上下文答案： 是的
    - 优化后给出答案：  是的，不能在一个页面里面同时操作设置两个素材（A图案对应白T，B图案对应黑T），只能选一个通用于白黑T的图案。

### 技能 3: 上下文不足信息内容补齐
1. 若问题经过数据库检索，给出的上下文信息非常少或没有，需结合自身专业知识回答。
2. 示例：
    - 问题：如果我的产品标题中即有泰文和英文，算重复吗？
    - 上下文的信息：请不要在标题中出现叠词
    - 回答：本地知识库中未提供关于产品既有泰文和英文是否算重复的相关内容，结合专业知识，目前没有明确固定标准判定这种情况一定算重复，具体要根据不同平台规则和实际情况判断。以上信息仅供参考。

### 技能 4 敏感词过滤
1. 当遇到不符合安全规范或者敏感的词汇时，优先利用插件check进行敏感词搜索，然后去除

## 限制:
- 仅回答与跨境电商相关的问题，拒绝处理与跨境电商无关的话题。
- 优先以检索的知识库内容为答案，除非内容不全在进行优化或者补全
- 回答内容必须符合上述技能要求的格式和规范。
- 若回答基于知识库已有信息，需遵循相应说明规范；若知识库中没有相关内容，应按技能 3 要求回答用户 。 

• RAG大模型用户Prompt

properties 用户输入问题：{x}query

直接输出

• 作用：直接输出经过RAG检索后大模型的结果

大模型闲聊回答

• 作用：

properties 接受用户的问题，直接基于大模型本身回答结果

• 实现方式：添加**《大模型》**节点

• 注意点：如果问题的问题不相关时，可能召回的内容也不相关。因此需要做一次过滤，就可以保证召回的内容是相关的。

• 闲聊大模型：系统Prompt

# 角色
你是一位专业且亲切的跨境电商答疑小助手，不仅拥有深厚的跨境电商专业知识，还具备广泛的通用知识储备。你能够根据用户需求，灵活切换角色，既能为用户解答跨境电商领域的复杂专业问题，也能与用户展开轻松愉快的日常闲聊。

## 技能
### 技能 1: 精准解答跨境电商问题
1. 当用户提出跨境电商相关问题时，充分运用自身专业知识，结合丰富且贴合实际的案例，为用户提供准确、全面且详细的解答。
2. 针对复杂的跨境电商概念，运用通俗易懂、生动形象的语言进行深入浅出的解释说明，确保用户能够轻松理解。
3. 在回答的结尾处明确注明“以上答案仅供参考”。

### 技能 2: 耐心回应非跨境电商问题
1. 当用户提及非跨境电商问题时，例如简单的数学运算“1 + 1 等于几”等，运用已有的知识储备，耐心、准确地对用户进行回答。
2. 在回答此类问题的结尾同样注明“以上答案仅供参考”。

### 技能 3: 合理引导不明确问题
1. 当用户输入的问题含义不明确时，不直接给出知识回答，而是友好地输出“我还在努力理解您的问题，请您详细描述后再来询问吧，这样我能更好地为您解答。”

## 限制:
- 交流内容主要围绕跨境电商学习以及其他非电商知识相关范畴，对于与这两类内容无关的话题，需礼貌地拒绝回答，并告知用户“抱歉，我只能回答与跨境电商学习和其他非电商知识相关的问题哦”。
- 所输出的内容必须严格符合上述回答要求，保证格式规范、条理清晰、逻辑连贯。 
- 回答的答案中禁止提及：需要进行知识库检索

• user

properties 用户问题：{x}query

直接输出

测试运行

RAG检索：

闲聊：

项目简介

背景介绍

针对扫地机器人的客服系统，可以完成咨询问答以及报告生成任务 ，主要面向客户，toC使用。

项目架构

整体架构如下：

代码架构

代码架构如下所示：

扩展：

在项目开始之前，需要在服务端（或者本机）使用ollama部署模型如下：

ollama run qwen3:8b ollama run gemma3:1b ollama run deepseek-r1:1.5b ollama run bge-m3:latest

数据集介绍

具体保存位置如下：

ZST_code/data

架构如下：

本项目数据来源于扫地机器人的产品手册及相关知识库，方便为用户解答使用问题。

具体问题包括：基础与技术类、产品功能类、智能交互类、维护与保养类、选购指南类、技术发展类、使用技巧类、配件与耗材类、安全与隐私类、特殊场景类等问题。举例如下：

模型选型及实现

Agent模块构建（尽量理解）

Agent简介

Agent组成部分：

项目中Agent是基于ReAct框架实现的，它主要包括推理（Reasoning）模块以及行动(Acting)模块。

ReAct核心思想：

1、推理模块

【CoT】:Agent每一步生成自然语言推理逻辑，解释当前决策的原因

【错误回溯机制】：当行动失败时，Agent能分析原因并调整策略（如：“用户输入与目前所有的工具不匹配，但是提示词工程中要求必须使用工具，因此可能需要使用Rag工具，但用户输入与Rag工具不匹配，我们重新思考一下确认应该使用的工具。”）

2、行动模块

【工具调用】:调用外部API、第三方数据库以及计算器等方式。

【环境状态感知】：实时接收行动结果，作为下一步决策的输入（如：“获取到北京气温25度，建议用户带薄外套。”）

具体实现

具体代码位置如下：

ZST_code/Agent

1、Action实现

具体代码位置如下：

ZST_code/Agent/Action.py

具体代码如下：

from pydantic import BaseModel,Field
from typing import List,Optional,Dict,Any

class Action(BaseModel):
    # 工具名称
    tool_name: str=Field(descrition="The name of a tool.")
    # 调用参数
    args: Optional[Dict[str,Any]] = Field(description="Input args of a tool,containing names and values")

    def __str__(self):
        # 工具名称以及调用参数组合在一起
        rst = f"Action(tool_name={self.tool_name}"
        # 判断参数是否存在
        if self.args:
            for k,v in self.args.items():
                rst += f",{k}={v}"
        rst += ")"
        return rst

2、ReAct实现

具体代码位置如下：

ZST_code/Agent/ReAct.py

具体代码实现如下：

import re
from typing import List, Optional,Tuple

from langchain_community.chat_message_histories.in_memory import ChatMessageHistory
from langchain_core.language_models.chat_models import BaseChatModel
from langchain.output_parsers import PydanticOutputParser,OutputFixingParser
from langchain.schema.output_parser import StrOutputParser
from langchain.tools.base import BaseTool
from langchain_core.prompts import ChatPromptTemplate,MessagesPlaceholder
from langchain_core.tools import render_text_description
from pydantic import ValidationError
from langchain_core.prompts import HumanMessagePromptTemplate

from Agent.Action import Action
from Utils.PrintHandler import PrintHandler,JsonFileHandler
from Utils.FileReader import ConfigHandler

conf = ConfigHandler()

class ReActAgent:
    @staticmethod
    def extract_json_from_action_rst(text:str) -> str | None:
        """从action结果中提取json"""
        pattern = re.compile(r'```json(.*?)```',flags=re.DOTALL)
        matches = pattern.findall(text)
        if matches:
            json_str = matches[-1]
            return json_str
        return None

    @staticmethod
    def format_observations(response:str,action:Action,observation:str) -> str:
        """将observation格式化"""
        rst = re.sub(r"```json(.*?)```","",response,flags=re.DOTALL)
        rst += '\n' + str(action) + '\n返回的结果是：\n' + observation
        return rst

    def __init__(self,
                 llm:BaseChatModel,
                 tools:List[BaseTool],
                 main_prompt_file_path:str):
        """初始化过程
        input:
            llm:使用的大模型
            tools:工具集
            main_prompt_file_path:主提示词的存储路径
        """
        self.llm = llm
        self.tools = tools
        # 输出格式解析器
        self.output_parser = PydanticOutputParser(pydantic_object=Action)
        try:
            # 带自动修复的输出解析器
            self.robust_parser = OutputFixingParser.from_llm(
                parser=self.output_parser,
                llm=llm
            )
        except Exception as e:
            print(f'不能修复输出，错误：{e}')
            print("请尝试提升提示词质量或者使用一个更加合适的大模型")

        # 主提示词文档路径
        self.main_prompt_file_path = main_prompt_file_path
        # 初始化提示工程
        self.__init_prompts()
        # 初始化agent的链路
        self.__init_chains()
        # 显示详细信息的处理器
        self.verbose_handler = PrintHandler() if conf.logging_flag == 0 else JsonFileHandler()

    def __init_prompts(self):
        with open(self.main_prompt_file_path,'r',encoding='utf-8') as f:
            self.prompt = ChatPromptTemplate.from_messages(
                [MessagesPlaceholder(variable_name='chat_history'),
                 HumanMessagePromptTemplate.from_template((f.read())),]
            ).partial(
                tools=render_text_description(self.tools),
                tool_names=",".join([tool.name for tool in self.tools]),
                format_instructions=self.output_parser.get_format_instructions()
            )

    def __init_chains(self):
        self.main_chains = (self.prompt | self.llm | StrOutputParser)

    def execute(self,
                task:str,
                uuid:int,
                chat_history:ChatMessageHistory,
                verbose=False) -> str:
        """对外接口
        input:
            task:用户任务
            uuid:用户的uid
            chat_history:对话历史
            verbose:是否显示详细信息
        """
        short_term_memory = []
        action,response = self.step_by_step(
            task=task,
            uuid=uuid,
            short_term_memory=short_term_memory,
            chat_history=chat_history,
            verbose=verbose
        )
        answer = self.exec_actions(action)
        self.verbose_handler.on_tool_end(answer)

        short_term_memory.append(self.format_observations(response,action,answer))
        chat_history.add_user_message(task)
        chat_history.add_ai_message(answer)
        return answer

    def step_by_step(self,
                     task:str,
                     uuid:int,
                     short_term_memory=None,
                     chat_history=None,
                     verbose=False) -> Tuple[Action,str]:
        """按步骤执行agent思考过程"""
        inputs = {
            "input":task,
            "uuid":uuid,
            "agent_scratchpad":"\n".join(short_term_memory),
            "chat_history":chat_history.messages if chat_history else chat_history
        }
        response = ""
        for s in self.main_chains.stream(inputs):
            response += s

        action_rst_json = self.extract_json_from_action_rst(response)

        action = self.robust_parser.parse(
            action_rst_json if isinstance(action_rst_json,dict) else response
        )
        return action,response

    def exec_actions(self,action:Action) -> str:
        """执行action
        input:
            action:Action对象
        """
        tool = self.confirm_tool(action.tool_name)
        if tool is None:
            observation = (
                f"错误：没找到待执行的工具或者指令'{action.tool_name}'"
                f"从提供的工具或者指令列表中选择，注意按对顶格式来输出最终内容"
            )
        else:
            try:
                observation = tool.run(action.args)
            except ValidationError as e:
                observation = (
                    f"执行动作的时候参数有非法错误 in exec_actions:{str(e)},args:{action.args}"
                )
        return observation

    def confirm_tool(self,
                     tool_name:str) -> Optional[BaseTool]:
        for tool in self.tools:
            if tool.name == tool_name:
                return tool
        return None

LLM模型选型及构建

代码位置：

ZST_code/Models/Factory.py

具体实现：

from langchain_ollama import ChatOllama,OllamaEmbeddings
import os
import sys
sys.path.append(sys.path.append(os.path.dirname(os.getcwd())))
from Utils.FileReader import ConfigHandler

# 模型配置文件获取
conf = ConfigHandler()
model_conf = conf.read_yaml("Configs/config.yml")

class ChatModelFactory:
    """ChatModel模型工厂定义"""
    def __init__(self) -> None:
        self.base_url = model_conf['server_url']

    def get_model(self):
        llm = ChatOllama(model=model_conf['model_name'],base_url=self.base_url)
        return llm


class EmbeddingModelFactory:
    """向量模型工厂定义"""
    def __init__(self):
        self.base_url = model_conf['server_url']

    def get_model(self):
        embeddings = OllamaEmbeddings(model=model_conf['embedding_name'],base_url=self.base_url)
        return embeddings
# ChatModel实例化
llm = ChatModelFactory().get_model()
# 向量模型实例化
embed_model = EmbeddingModelFactory().get_model()

工具模块构建

具体代码位置如下：

ZST_code/Tools

整体架构如下：

工具集声明

工具集中包括所有我们可以使用的工具声明，具体代码如下：

import warnings
warnings.filterwarnings("ignore")

from langchain.tools import StructuredTool
from .RagQATool import rag_qa
from .ReportGenerationTool import generator

rag_qa_tool = StructuredTool.from_function(
    func=rag_qa,
    name="RagQA",
    description="使用RAG检索知识进行问答"
)

report_generation_tool = StructuredTool.from_function(
    func=generator,
    name="ReportGeneration",
    description="从第三方系统查询信息，完成报告"
)

RAG工具模块构建

1、文档读取模块构建

具体代码如下：

from langchain_community.document_loaders import UnstructuredWordDocumentLoader,PyPDFLoader
import yaml
import pandas as pd

class FileReader:
    """知识库文档读取"""
    def __init__(self):
        pass

    def read_docx(self,file_path):
        """读取word文档"""
        try:
            loader = UnstructuredWordDocumentLoader(file_path)
            doc = loader.load_and_split()
            return doc
        except FileNotFoundError as e:
            print(f'错误：文件{file_path}未找到')
            return None
        except Exception as e:
            print(f'读取文件出错：{e}')
            return None

    def read_md(self,file_path):
        """读取markdown文档"""
        try:
            with open(file_path,'r',encoding='utf-8') as f:
                content = f.read()
                return content
        except FileNotFoundError as e:
            print(f'错误：文件{file_path}未找到')
            return None
        except Exception as e:
            print(f'读取文件出错：{e}')
            return None

    def read_pdf(self,file_path):
        """读取pdf文档"""
        try:
            loader = PyPDFLoader(file_path)
            pages = loader.load_and_split()
            return pages
        except FileNotFoundError as e:
            print(f'错误：文件{file_path}未找到')
            return None
        except Exception as e:
            print(f'读取文件出错：{e}')
            return None

    def read_csvs(self,file_path,sep=','):
        """读取csv文档"""
        try:
            data = pd.read_csv(file_path,sep=sep)
            return data
        except FileNotFoundError as e:
            print(f'错误：文件{file_path}未找到')
            return None
        except Exception as e:
            print(f'读取文件出错：{e}')
            return None


class ConfigHandler:
    """配置文件读取"""
    def __init__(self):
        pass

    def read_yaml(self,file_path="Configs/rag_config.yml",encoding='utf-8'):
        with open(file_path,'r',encoding=encoding) as f:
            return yaml.load(f.read(),Loader=yaml.FullLoader)


reader = FileReader()
config_reader = ConfigHandler()

2、向量数据库模块构建

具体位置如下：

ZST_code/Tools/VecStore.py

也可以直接调用如下代码：

ZST_code/Utils/create_db.py

• 整体架构

构建向量数据库包括文本加载、切分、数据库构建、保存及加载等步骤。

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from langchain_community.vectorstores import Chroma
import os
import sys
sys.path.append('../..')
sys.path.append('.')
sys.path.append('./')
from Utils.FileReader import reader,config_reader
from Models.Factory import embed_model

# RAG配置文件初始化
conf = config_reader.read_yaml("Configs/rag_config.yml")

class VecStore:
    def __init__(self,dir=conf['DATA_PATH'],save_dir=conf['CHROMA_PATH']):
        # 知识库路径
        self.dir = dir
        # 向量数据库保存路径
        self.save_dir = save_dir

    def load_documents(self):
        """知识库加载，可以加载word,markdown,pdf,csv文档"""
        files = os.listdir(self.dir)
        docs = []
        for file in files:
            file_path = os.path.join(self.dir,file)
            if file.endswith('.docx'):
                txts = reader.read_docx(file_path)
            elif file.endswith('.md'):
                txts = reader.read_md(file_path)
            elif file.endswith('.pdf'):
                txts = reader.read_pdf(file_path)
            elif file.endswith('.csv'):
                txts = reader.read_csvs(file_path)
            else:
                txts = []
            docs.extend(txts)
        return docs

    def split_text(self,documents):
        """文档切分，使用固定字符长度进行切分"""
        text_spliter = RecursiveCharacterTextSplitter(
            chunk_size=200,
            chunk_overlap=80,
            length_function=len,
            add_start_index=True
        )
        chunks = text_spliter.split_documents(documents)
        return chunks

    def get_db(self):
        """获取向量数据库"""
        documents = self.load_documents()
        chunks = self.split_text(documents)
        db = Chroma.from_documents(documents=chunks,embedding=embed_model,persist_directory=self.save_dir)
        return db

    def save_to_chroma(self,db):
        """将向量数据库保存"""
        db.persist()
        print(f"Saved chunks to {conf['CHROMA_PATH']}")

    def load_chroma(self):
        """向量数据库加载"""
        db = Chroma(persist_directory=self.save_dir,embedding_function=embed_model)
        return db

if __name__=="__main__":
    """向量数据库构建"""
    vec = VecStore()
    db = vec.get_db()
    vec.save_to_chroma(db)

运行结果如下图所示：

3、RAG工具调用

代码位置如下：

ZST_code/Tools/RagQATool.py

具体代码如下：

from langchain.chains import RetrievalQA
from Models.Factory import llm
from Tools.VecStore import VecStore

vec = VecStore()
db = vec.load_chroma()

def rag_qa(query):
    qa = RetrievalQA.from_chain_type(
        llm=llm,
        # stuff:将检索到的文档都给LLM让他生成答案
        # map_reduce:单独发送给LLM，生成摘要
        # refine:迭代处理文档
        # map_rerank:让LLM对每一个文档评分给出一个置信度分数，重排，取置信度最高的文档
        chain_type="stuff",
        retriever=db.as_retriever(search_kwargs={"k":5}),
        return_source_documents=True
    )
    response = qa(query)
    return response['result']

功能测试代码如下：

import sys
sys.path.append('../..')
sys.path.append('.')
sys.path.append('./')
from Tools.RagQATool import rag_qa

print(rag_qa("为什么有些机器人会'迷路'？"))

运行结果如下图所示：

报告生成工具构建

报告生成需要经过以下几个步骤：从query中进行信息提取-->从第三方系统中获取数据-->将所有信息组合在一起再调用大模型进行报告生成。

1、信息提取

具体代码如下：

from Models.Factory import llm
import json
from tenacity import retry,stop_after_attempt,wait_fixed
import re

# 如果一次调用大模型失败，多尝试几次确保调用成功
@retry(stop=stop_after_attempt(3),wait=wait_fixed(2))
def invoke_llm(prompt):
    return llm.invoke(prompt)

def extract_info(query):
    system_prompt = """
    你是一个优秀的信息提取专家，你的任务是从输入文本中获取时间这个参数值。
    返回JSON格式：{"时间":"2025-01"}.
    如果输入文本中没有相关参数，则返回空。
    举例如下：
    输入文本：使用2025年1月份的
    返回结果：{"时间":"2025-01"}
    输入文本：根据我的扫地机器人的使用情况，生成一份使用报告
    返回结果：{}
    """
    data = {}
    human_prompt = f"输入文本：{query},请返回提取的json。"
    # 提示词组合
    prompt = [
        {
            "role":"system",
            "content":system_prompt
        },
        {
            "role":"user",
            "content":human_prompt
        }
    ]
    # 大模型返回结果
    response = invoke_llm(prompt)
    if response:
        # 获取返回结果内容
        content = response.content
        # 正则提取结果
        match = re.search(r'\{.*\}',content)
        # 正则提取成功的话，获取比较干净的json
        if match:
            cleaned_json = match.group(0)
            data = json.loads(cleaned_json)
        else:
            print("未找到有效的json")
    return data

功能测试代码如下：

import sys
sys.path.append('../..')
sys.path.append('.')
sys.path.append('./')
from Tools.ExtractInfo import extract_info

print(extract_info("根据我的扫地机器人的使用情况，生成一份使用报告,要2025年6月份的"))

运行结果：

2、第三方系统中获取数据

具体代码如下：

import sys
sys.path.append('../..')
sys.path.append('.')
sys.path.append('./')
from Utils.FileReader import FileReader
import pandas as pd
from datetime import datetime

def fetch_data_from_external_systems(uuid=1001,info_dict={}):
    """根据传入的信息，从第三方系统中获取返回值

    Args:
        uuid (int): 用户ID，示例：{"uuid":"1001"}
        info_dict(json):从query中提取的时间，如：{"时间":"2025-01"}

    Returns:
        result: 查询到的信息，list格式，示例：result = [1001,"65㎡公寓 | 单身 | 木地板","覆盖率:85%\n日均清扫:45㎡\n漏扫区域:沙发底部（高度不足）","主刷寿命:剩余60天\nHEPA滤网:剩余40%","优于65%同面积用户（清洁频率更高）","2025-06"]
    """
    result = None
    try:
        reader = FileReader()
        data = reader.read_csvs(r'data\query_from_others\query_data.csv')
        # query 中有时间信息
        if '时间' in info_dict.keys():
            # 获取query中提取的时间
            query_time = info_dict['时间']
            # 根据query中提取的时间和uid获取扫地机器人的使用情况
            result = data[(data['用户ID'] == uuid) & (data['时间'] == query_time)]
            # 如果query中提取的时间在数据库中不存在数据的话，默认使用25年6月份的数据
            if len(result) == 0:
                result = data[(data['用户ID'] == uuid) & (data['时间'] == '2025-06')]
                # print('当前时间没有该用户的扫地机器人使用情况') 实际开发需要
        # query中没有时间的话
        else:
            # 获取当前年月
            current_year_month = datetime.now().strftime("%Y-%m")
            # 获取当前年月对应的数据
            result = data[(data['用户ID'] == uuid) & (data['时间'] == current_year_month)]
            # 如果当前年月没有对应数据的话，默认使用25年8月份
            if len(result) == 0:
                result = data[(data['用户ID'] == uuid) & (data['时间'] == "2025-08")]
                # print('当前时间没有该用户的扫地机器人使用情况') 实际开发需要
    except Exception as e:
        print(f'从第三方系统获取信息出错：{e}')
    return result

功能测试代码如下：

import sys
sys.path.append('../..')
sys.path.append('.')
sys.path.append('./')
from Tools.FetchFromOtherSys import fetch_data_from_external_systems
from Tools.ExtractInfo import extract_info

# query中有时间信息
a = extract_info("根据我的扫地机器人的使用情况，生成一份使用报告,要2025年6月份的")
b = fetch_data_from_external_systems(info_dict=a)
print('有时间的数据生成：',b)

# query中没有时间信息
a1 = extract_info("根据我的扫地机器人的使用情况，生成一份使用报告")
b1 = fetch_data_from_external_systems(info_dict=a1)
print('没时间的数据生成：',b1)

运行结果如下图所示：

3、报告生成

具体代码如下：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import PromptTemplate
import os
import sys
sys.path.append('../..')
sys.path.append('.')
sys.path.append('./')
sys.path.append(os.path.dirname(os.getcwd()))
from Models.Factory import llm
from .ExtractInfo import extract_info
from .FetchFromOtherSys import fetch_data_from_external_systems
from Utils.FileReader import config_reader

model_conf = config_reader.read_yaml('Configs/config.yml')

def generator(query,uuid=1001,verbose=False):
    try:
        # 报告生成提示词
        prompt = []
        # 读取报告生成提示词
        with open(model_conf['reporter_prompt_path'],'r',encoding='utf-8') as f:
            for line in f:
                prompt.append(line.strip())
        # 使用大模型进行信息提取，提取时间
        report_param = extract_info(query)
        # 如果格式不是字典的 话，报错
        if not isinstance(report_param,dict):
            return f"信息提取返回结果格式不对，请检查，返回信息为：{report_param}"
        # 从第三方系统中获取数据
        information = fetch_data_from_external_systems(uuid,report_param)
        # 获取数据是空的话，返回空
        if information is None:
            return ""
        # 构造模板
        template = PromptTemplate(
            template="".join(prompt),
            input_variable=["INFORMATION"]
        )
        # 模板格式化
        formatted_prompt = template.format(INFORMATION=information)
        # 从格式化的模板中获取最终使用的提示词模板
        final_template = PromptTemplate.from_template(formatted_prompt)
        # 运行链路
        chain = {"query":RunnablePassthrough} | final_template | llm | StrOutputParser()
        # 根据链路获取大模型生成的答案
        answer = chain.invoke(query)
        return answer
    except Exception as e:
        print(e)
        return None

功能测试代码如下：

import sys
sys.path.append('../..')
sys.path.append('.')
sys.path.append('./')
from Tools.ReportGenerationTool import generator

b = generator("根据我的扫地机器人的使用情况，生成一份使用报告,要2025年6月份的")
print(b)

运行结果如下：

提示词模块构建

具体位置如下：

ZST_code/prompts

架构如下：

主提示词文本内容如下：

你是强大的AI助手，可以使用工具与指令自动化解决问题。

你的任务是:
{input}

你可以使用以下工具或指令，它们又称为动作或actions:
{tools}

你必须遵循以下约束来完成任务。
1. 每次决策只使用一种工具。
2. 你所调用的指令或者工具在给定的下述工具列表中, {tool_names}。
3. 保证你的回答中不会包含违法或者侵犯性的信息。
4. 如果执行某个工具失败，请尝试改变参数或参数格式再次调用。
5. 你生成的回复必须符合上文中给定的事实信息。不可以编造信息。DO NOT MAKE UP INFORMATION.
6. 保证你生成的动作可以精确执行。
7. 使用中文回答。
8. 不要构造query。用户输入什么，问题就是什么，不要扩充问题。
9. 如果query跟可使用的工具无关，请直接返回“我不知道”。

输出形式如下：
{format_instructions}
请确保你输出的JSON代码块以```json\n\n```包裹。

报告生成提示词文本：

你是专业的报告写手。根据查询回来的信息，写一份机器使用情况报告并给出建议。
信息格式如下：
​```
用户ID,特征,清洁效率,耗材,对比,时间
1002,70㎡公寓 | 情侣 | 瓷砖,"覆盖率：88%  \n避障失败：2次/周（电线缠绕）","边刷磨损：中度  \n尘盒清理：每3天",拖地功能使用率低于同类30%,2025-01
​```
返回结果如上格式。
最终查回来的信息如下:
{INFORMATION}
根据查询回来的信息，使用中文生成一份报告，要求将查询回来的信息组装成一份文本内容，并根据使用情况给出具体建议，不要将查询回来的内容直接输出。

模型评估

具体位置如下：

ZST_code/Eval

整体内容如下：

评估指标简介

1、检索效果的评估：precision,recall,mrr(平均倒数排名)

MRR计算公式：

2、生成评估指标：rouge-1,rouge-l,bertscore,faithfulness

3、项目总评估：qps,平均延时(avg_latency)

代码实现

具体代码如下：

import json
import time
import numpy as np
import re
from rouge_score import rouge_scorer
from transformers import AutoModel, AutoTokenizer
import torch
from bert_score import BERTScorer
from bert_score.utils import get_model, lang2model, model2layers
from transformers import pipeline
from sklearn.metrics import precision_score, recall_score
import pandas as pd
import jieba
import sys,os
sys.path.append('../..')
sys.path.append('.')
sys.path.append('./')
from Agent.ReAct import ReActAgent
from Models.Factory import ChatModelFactory
from Tools.Tools import *
from Tools.RagQATool import rag_qa
from langchain_community.chat_message_histories.in_memory import ChatMessageHistory
from tqdm import tqdm
import random
import ast

device = "cuda" if torch.cuda.is_available() else "cpu"
stopwords = []
with open(r'D:\course\single_agent-main\single_agent-main\agent_project_bak\stopwords.txt','r',encoding='utf-8') as f:
    for line in f:
        stopwords.append(line.strip())

# 1. 准备评测数据
eval_data = pd.read_csv(os.path.join(os.getcwd(),'Eval/eval_data.csv'))

# 2.获取agent
llm = ChatModelFactory().get_model()

tools = [
    rag_qa_tool,
    report_generation_tool,
]

agent = ReActAgent(
    llm=llm,
    tools=tools,
    main_prompt_file_path="prompts/main.txt"
)

def launch_agent(query,agent,uuid=1001):
    chat_history = ChatMessageHistory()
    reply = agent.execute(query,uuid,chat_history=chat_history,verbose=True)
    return reply

# 3. 评测指标计算函数
def evaluate_retrieval(true_docs,retrieved_docs):
    """计算评估指标"""
    y_true = [1 if doc in true_docs else 0 for doc in retrieved_docs] # True
    y_pred = [1] * len(retrieved_docs) # all

    precision = precision_score(y_true,y_pred,zero_division=0)
    recall = recall_score(y_true,y_pred)
    mrr = 1 / (true_docs.index(retrieved_docs[0]) + 1) if retrieved_docs[0] in true_docs else 0
    return {
        "Precision@5":precision,
        "Recall@5":recall,
        "MRR":mrr
    }

def load_bert_scorer():
    #加载自定义模型和分词器
    model_path = os.path.join(os.getcwd(),'bert-base-chinese')
    tokenizer = AutoTokenizer.from_pretrained(model_path)
    # 手动加载模型配置
    num_layers = model2layers.get(os.path.basename(model_path),12)
    model_type = lang2model.get('zh','bert-base-chinese')
    model = AutoModel.from_pretrained(model_path)

    return BERTScorer(
        model_type=model_type,
        num_layers=num_layers,
        lang='zh',
        use_fast_tokenizer=True,
        rescale_with_baseline=False,
        device=model.device
    )



def evaluate_generation(precision,reference,context):
    """计算生成指标"""
    # ROUGE
    scorer = rouge_scorer.RougeScorer(['rouge1','rougeL'],use_stemmer=False)
    rouge = scorer.score(reference,precision)

    # BERTScore
    bert_scorer = load_bert_scorer()
    _,_,bert_f1 = bert_scorer.score([precision],[reference])

    # 忠实度（简易版：检测关键实体是否一致）
    random.seed(42)
    key_entities =set([word for word in jieba.lcut(reference) if word not in stopwords])
    precision_entities = set([word for word in jieba.lcut(precision) if word not in stopwords])
    faithful = sum(1 for e in key_entities if e in precision_entities) / max(1,len(key_entities))

    return {
        'ROUGE-1':rouge['rouge1'].fmeasure,
        'ROUGE-L':rouge['rougeL'].fmeasure,
        'BERTScore':bert_f1.mean().item(),
        'Faithfulness':faithful
    }

def preprocess_text(text):
    """统一的文本预处理函数"""
    # 转换为小写
    text = text.lower()
    text = re.sub(r'\n','',text)
    # 移除标点符号
    text = re.sub(r'[- *]+', '，', text).replace("，，|：，",'，')
    # 移除多余空格
    text = re.sub(r'\s+', ' ', text)
    return text

# 4. 主评测流程
results = []
latencies = []
eval_data['reference_docs'] = eval_data['reference_docs'].apply(ast.literal_eval)
for i in tqdm(range(len(eval_data))):
    item = eval_data.iloc[i]
    start_time = time.time()
    query = item['query']
    reference = ''.join(item['reference_docs'])
    reference_answer = item['reference_answer']
    response = launch_agent(query,agent,1001)

    if '<think>' in response and '</think>' in response:
        answer = re.sub(r"<think>.*?</think>","",response,flags=re.DOTALL)
    elif '<think>' in response and '</think>\n\n答：' in response:
        answer = re.sub(r"<think>.*?</think>\n\n答：","",response,flags=re.DOTALL)
    else:
        answer = response
    # 执行RAG流程
    retrieved_docs = rag_qa(item['query'])

    # 记录延迟
    latency = time.time() - start_time
    latencies.append(latency)
    # 评估各模块
    retrieval_metrics = evaluate_retrieval(retrieved_docs,retrieved_docs)
    ## 文本预处理
    reference_processed = preprocess_text(reference)
    precision_processed = preprocess_text(answer)
    generation_metrics = evaluate_generation(precision_processed,reference_processed,retrieved_docs)

    results.append({
        **retrieval_metrics,
        **generation_metrics
    })

# 5. 汇总结果
aggregates = {
    "Retrieval":{
        "Precision@5": np.mean([r['Precision@5'] for r in results]),
        "Recall@5": np.mean([r['Recall@5'] for r in results]),
        "MRR": np.mean([r['MRR'] for r in results]),
    },
    "Generation":{
        "ROUGE-1":np.mean([r['ROUGE-1'] for r in results]),
        "ROUGE-L":np.mean([r['ROUGE-L'] for r in results]),
        "BERTScore":np.mean([r['BERTScore'] for r in results]),
        "Faithfulness":np.mean([r['Faithfulness'] for r in results]),
    },
    "Performance":{
        "Avg_Latency":np.mean(latencies),
        "QPS":1 / np.mean(latencies)
    }
}

print('=' * 50 + '\n综合评测结果：\n' + '=' * 50)
print(json.dumps(aggregates,indent=2))

运行结果如下：

日志打印

简介说明

两类：一类本地测试在控制台打印日志；一类在生产环境/测试环境打印日志，保存到json文件中

代码实现

代码位置：

ZST_code/Utils/PrintHandler.py

代码如下：

from abc import ABC,abstractmethod
from typing import Any,Dict,Optional
import json
from datetime import datetime

class VerboseHandler(ABC):
    """抽象基类，用于处理详细输出"""
    @abstractmethod
    def on_tool_start(self,tool_name:str,input_data:Any) -> None:
        pass

    @abstractmethod
    def on_tool_end(self,tool_name:str,output_data:Any,metadata:Optional[Dict]=None)->None:
        pass

    @abstractmethod
    def on_tool_error(self,tool_name:str,error:Exception) -> None:
        pass


class PrintHandler(VerboseHandler):
    """具体的详细输出处理器，将信息打印到控制台"""
    def __init__(self,verbose:bool=True,indent:int=2):
        self.verbose = verbose
        self.indent = indent

    def on_tool_start(self,tool_name:str,input_data:Any) -> None:
        if not self.verbose:
            return
        print(f"\n{'='*50}")
        print(f'工具开始执行：{tool_name}')
        print(f"时间：{datetime.now().isoformat()}")
        print("输入数据：")
        self._print_formatted(input_data)
        print(f"\n{'='*50}")

    def on_tool_end(self,tool_name:str,output_data:Any,metadata:Optional[Dict]=None) ->None:
        if not self.verbose:
            return
        print(f"\n{'=' * 50}")
        print(f'工具执行完成：{tool_name}')
        print(f"时间：{datetime.now().isoformat()}")
        print("输出数据：")
        self._print_formatted(output_data)
        if metadata:
            print("元数据：")
            self._print_formatted(metadata)
        print(f"\n{'=' * 50}")

    def on_tool_error(self,tool_name:str,error:Exception) -> None:
        if not self.verbose:
            return
        print(f"\n{'=' * 50}")
        print(f'工具执行出错：{tool_name}')
        print(f"时间：{datetime.now().isoformat()}")
        print(f"错误信息：{str(error)}")
        print(f"\n{'=' * 50}")

    def _print_formatted(self,data:Any) -> None:
        try:
            if isinstance(data,(dict,list)):
                formatted = json.dumps(data,indent=self.indent,ensure_ascii=False)
                print(formatted)
            else:
                print(str(data))
        except Exception as e:
            print(f"无法格式化输出：{e}")
            print(str(data))


class JsonFileHandler(VerboseHandler):
    """将详细输出保存到JSON文件"""
    def __init__(self,file_path):
        self.file_path = file_path
        self.logs = []

    def on_tool_start(self,tool_name:str,input_data:Any) -> None:
        log_entry = {
            "event":"tool_start",
            "tool_name":tool_name,
            "timestamp":datetime.now().isoformat(),
            "input_data":input_data
        }
        self.logs.append(log_entry)
        self._save_logs()

    def on_tool_end(self,tool_name:str,output_data:Any,metadata:Optional[Dict]=None) ->None:
        log_entry = {
            "event": "tool_end",
            "tool_name": tool_name,
            "timestamp": datetime.now().isoformat(),
            "output_data": output_data,
            "metadata": metadata or {}
        }
        self.logs.append(log_entry)
        self._save_logs()

    def on_tool_error(self,tool_name:str,error:Exception) -> None:
        log_entry = {
            "event": "tool_error",
            "tool_name": tool_name,
            "timestamp": datetime.now().isoformat(),
            "error": str(error)
        }
        self.logs.append(log_entry)
        self._save_logs()

    def _save_logs(self) -> None:
        """将日志保存到文件中"""
        with open(self.file_path,'w',encoding='utf-8') as f:
            json.dump(self.logs,f,indent=2,ensure_ascii=False)

运行结果如下图所示：

模型部署

服务端

代码位置：

ZST_code/main.py

# 具体代码实现过程
from Agent.ReAct import ReActAgent
from Models.Factory import ChatModelFactory
from Tools.Tools import *
from langchain_community.chat_message_histories.in_memory import ChatMessageHistory
from flask import Flask,request,jsonify
import re

app = Flask(__name__)

def launch_agent(query,agent,uuid=1001):
    chat_history = ChatMessageHistory()
    reply = agent.execute(query,uuid,chat_history=chat_history,verbose=True)
    return reply

@app.route("/predict",methods=["POST"])
def main():
    try:
        data = request.get_json()
        query = data.get("sentence","")
        uuid = data.get("uuid",1001)
        if not query:
            return jsonify({"error":"请输入文本"}),400
        llm = ChatModelFactory().get_model()

        tools = [
            rag_qa_tool,
            report_generation_tool,
        ]

        agent = ReActAgent(
            llm=llm,
            tools=tools,
            main_prompt_file_path="prompts/main.txt"
        )

        response = launch_agent(query,agent,uuid)

        if '<think>' in response and '</think>' in response:
            answer = re.sub(r"<think>.*?</think>", "", response, flags=re.DOTALL)
        elif '<think>' in response and '</think>\n\n答：' in response:
            answer = re.sub(r"<think>.*?</think>\n\n答：", "", response, flags=re.DOTALL)
        else:
            answer = response
        return jsonify({"answer":answer}),200

    except Exception as e:
        print(e)
        return jsonify({"error":str(e)}),500

if __name__=="__main__":
    app.run(host="0.0.0.0",port=8001)

运行结果如下图所示：

客户端

注意：在进行问答任务及报告生成任务时，完成速度依赖于部署服务的机器性能，如果是利用32G的V100部署云端服务器，问答任务耗时20-40s，报告生成任务耗时1分钟左右；但如果是笔记本自带的GPU ，比如3060，问答任务可能耗时10多分钟，报告生成任务耗时1个多小时。整体耗时大家要根据自己的机器来看。

代码位置：

ZST_code/app_test_web.py

具体代码实现：

import streamlit as st
import requests

# Streamlit页面配置
st.title("扫地机器人智能客服系统")
st.write("输入一个句子，点击生成按钮进行问答及报告生成任务")

# 输入框
sentence = st.text_area("请输入句子：",height=100)

# 生成按钮
if st.button("生成"):
    if not sentence.strip():
        st.error("请输入有效的句子！")
    else:
        try:
            # 调用FLask API
            response = requests.post("http://192.168.0.102:8001/predict",json={'sentence':sentence,"uuid":1001})
            response.raise_for_status()
            result = response.json()
            if 'answer' in result:
                st.success(f"生成结果：{result['answer']}")
            else:
                st.error(f"错误：{result.get('error','未知错误')}")
        except requests.exceptions.RequestException as e:
            st.error(f'无法连接到API：{str(e)}')

# 添加说明
st.markdown("""
### 使用说明
1. 在文本框中输入一个句子。
2. 点击“生成”按钮，系统会调用模型进行问答及报告生成任务。
""")

运行结果如下：

问答任务运行结果如下：

报告生成任务运行结果如下：

补充：配置文件

代码位置：

D:\course\single_agent-main\single_agent-main\agent_project\Configs

结构如下图所示：

模型配置文件

代码位置如下所示：

D:\course\single_agent-main\single_agent-main\agent_project\Configs\config.yml

具体代码如下：

# 生成模型名称
model_name: qwen3:8b
# model_name: deepseek-r1:1.5b
#model_name: gemma3:1b
# 向量模型名称
embedding_name: bge-m3:latest
# ollama部署模型的ip+port
server_url: http://127.0.0.1:11434
# 报告生成提示词文件路径
reporter_prompt_path: prompts/report_prompt.txt

# 打印日志时的控制按钮,取0的时候是本地，使用PrintHandler；取1的时候，是生产环境，使用JsonFileHandler
logging_flag: 0

RAG配置文件

代码位置如下所示：

D:\course\single_agent-main\single_agent-main\agent_project\Configs\rag_config.yml

具体代码如下：

# 向量数据库保存路径
CHROMA_PATH: Chroma
# 原始知识库路径
DATA_PATH: Data