开发语言：Python

大模型：Deepseek

现有业务：

输入一批茶叶检测指标参数 → 运行固定业务算法 → 输出检测结论、等级、风险提示

目标改造：把整套检测计算包装成 AI Agent，而不是单纯一个接口。

Agent 相比普通接口新增能力

1. 自然语言输入：可以文字录入参数，不用严格表单
2. 参数缺失时 Agent 主动提问补全参数
3. 自动调用原有茶叶检测算法（核心计算逻辑不动，不要把业务规则交给大模型算，避免结果不准）
4. 生成解读报告、风险分析、改进建议
5. 支持追问：“为什么判定为二级、哪一项指标超标”
6. 返回结构化结果，供网页 / App 展示

一、环境搭建

Agent = LLM 大脑 + 自定义工具 (调用原有检测算法) + 参数校验 + 对话记忆

• LLM 大模型：推理、理解、生成文字（大脑）
• Skill / Tool：一项一项可以调用的能力（工具：查国标、茶叶检测）
• ReAct：自主循环模式 思考→行动→观察，模型自己临时决定下一步
• Workflow（工作流）：人为编排好步骤、分支、条件、循环，约束 Agent 的执行路径

二、流程设计

Agent 完整工作流程

1. 接收用户输入，可以是表单 JSON 也可以是一段自然语言

示例输入：水分 7.2，灰分 5.1，茶多酚 18.5

2.LLM 解析文字，提取结构化检测参数

3. 校验必填指标是否齐全，如果缺失 →返回提示，让用户补充参数

4. 参数完整 →Agent 调用工具（你的原有茶叶检测算法）

5. 拿到原始检测结果

6. 交给 LLM 生成一份可读的检测分析报告

7. 同时返回两套数据

• 结构化原始结果（供 APP、网页渲染，保证数据准确）
• AI 自然语言解读报告

三、分阶段落地步骤

阶段一：python实现 Demo，熟悉agent核心组成部分和执行流程

1、新建一个独立测试项目，接入 DeepSeek 模型

2、写最简单 Agent，测试 @Tool 工具调用能否成功调用检测算法

3、实现基础：参数解析、工具调用、生成报告

阶段二：引入RAG 知识库

上传茶叶国标文档，Agent 可以基于国家标准给出解释

阶段三：引入 LangChain框架

工程化管理，生产级别实现验证

阶段四：工程化生产落地（持续学习）｜对接小程序 / App，上线可用系统

学习目标：从 Demo 变成可以给移动端调用的稳定服务，处理并发、异常、监控、部署。

学习知识点

1. 对外 REST 接口设计：统一入参出参 JSON，sessionId 会话管理，鉴权 token
2. 中间件：Redis（会话记忆、限流缓存）、Mysql（知识库元数据、业务记录）、Milvus 向量库，全部 Docker 部署
3. 异常处理：大模型超时、API 调用失败、工具调用异常、降级策略
4. 安全：输入过滤、接口限流、成本监控（统计 token 消耗，控制模型费用）
5. 架构选型
   • 方案 A：Agent 集成进现有 SpringBoot 项目（适合你的现状，不用新增域名，小程序直接调用）
   • 方案 B：Agent 独立微服务，HTTP 调用茶叶检测服务（后期业务量大再拆分）
6. 部署：Docker 打包、日志收集、可观测。

实操练习

1. 开发/api/tea‑agent/chat对外接口，模拟小程序调用。
2. 异常场景测试：参数缺失、大模型接口报错、网络超时。

产出

可部署的完整服务，小程序 / App 可以 HTTP 调用 Agent 接口获取检测结果。

四、架构设计

客户端(小程序/App/H5)
        ↓ HTTPS
Agent网关服务(SpringBoot)
 ├─会话管理层 Session+Redis持久记忆
 ├─Agent调度引擎(工作流Workflow)
 ├─工具调用模块 Tool‑Calling
 │    └─自定义业务工具 → 茶叶检测算法服务
 ├─RAG私有知识库模块
 │    ├─文档解析(PDF/Word/TXT茶叶国标、检测标准)
 │    ├─文本拆分、向量化
 │    ├─向量数据库(Milvus / Chroma)
 │    └─检索模块：根据用户问题召回相关标准文档片段
 ├─LLM模块（DeepSeek/通义千问）
 └─输出校验&结构化模块
        ↓
外部资源：大模型API、向量库、MySQL、Redis

三大核心组件

1. LLM 大脑：负责思考、任务拆解、生成文字
2. Agent 工作流 (Workflow)：定义固定执行步骤，控制执行顺序、分支判断、循环重试
3. 私有 RAG：读取本地私有文档（茶叶国标、检测规范），回答的时候参考文档，减少幻觉
4. Tools 工具：调用业务接口、算法、数据库

茶叶项目业务例子 用户提问：水分7.5，灰分5.3，帮我检测茶叶等级，并对照国标说明是否合格

Agent 执行工作流步骤

1. 解析用户输入提取检测参数

2.RAG 检索茶叶国标文档，获取判定标准

3. 参数是否齐全？缺失 →反问用户补充参数

4. 参数齐全 →调用【茶叶检测工具】执行业务算法

5. 结合 RAG 召回的国标文档 + 算法计算结果，生成最终检测报告

6. 返回结构化 JSON 给小程序 / App

练习项目（循序渐进）

1. Demo1：调用大模型 API，结构化 JSON 输出
2. Demo2：RAG 私有文档问答（茶叶国标 PDF）
3. Demo3：LangChain 框架封装
4. Demo4：完整茶叶检测 Agent

• 阶段 1：1-2天｜大模型 API 调用 + Prompt
• 阶段 2：2‑3 天｜RAG 知识库完整链路
• 阶段 3：3‑4 天｜Agent 工具调用、工作流，完成业务 Demo
• 阶段 4：2‑3 天｜工程化改造、接口、部署，对接小程序 / App

五、开发流程

1、环境准备

（1）Python 3.11.x安装

 官网地址：https://www.python.org/downloads/windows/

下载 Windows installer (64‑bit)（现在电脑基本都是 64 位） 、

如果官网下载慢，使用国内清华镜像 https://mirrors.tuna.tsinghua.edu.cn/python/

验证安装是否成功

python --version
pip --version

pip 国内镜像切换：

pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

快速安装 AI 开发基础依赖包

pip install langchain langchain-deepseek pymilvus python-dotenv

2、阶段 1：1-2天｜大模型 API 调用 + Prompt

实现能力：

1. 调用 DeepSeek‑V4‑Flash
2. 工具调用 Function‑call：模拟茶叶检测算法
3. 多轮对话记忆
4. 用户自然语言输入 →提取指标 →调用检测工具 →输出报告

环境：python>=3.10，使用官方 openai 兼容接口，不需要 LangChain，最简版本，方便理解原理

1）安装依赖

pip install openai python-dotenv

2）项目文件

.env 文件（放在代码同目录）

DEEPSEEK_API_KEY=sk-你的deepseek密钥
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-v4-flash

agent.py

import os
import json
from dotenv import load_dotenv
from openai import OpenAI
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility
from sentence_transformers import SentenceTransformer

# ---------------------- 加载配置 ----------------------
load_dotenv()
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL")
)
MODEL = os.getenv("DEEPSEEK_MODEL")

MILVUS_HOST = os.getenv("MILVUS_HOST")
MILVUS_PORT = os.getenv("MILVUS_PORT")
COLLECTION_NAME = os.getenv("MILVUS_COLLECTION")

# Embedding模型，用于文档向量化
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
EMB_DIM = 384

# ---------------------- Milvus初始化函数 ----------------------
def init_milvus():
    """连接Milvus，没有集合则自动创建"""
    connections.connect(host=MILVUS_HOST, port=MILVUS_PORT)
    if utility.has_collection(COLLECTION_NAME):
        return Collection(COLLECTION_NAME)

    fields = [
        FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
        FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=1024),
        FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=EMB_DIM)
    ]
    schema = CollectionSchema(fields, description="茶叶国标知识库")
    coll = Collection(COLLECTION_NAME, schema)
    index_params = {
        "metric_type": "L2",
        "index_type": "IVF_FLAT",
        "params": {"nlist":128}
    }
    coll.create_index("embedding",index_params)
    return coll

def insert_docs(coll:Collection):
    """初始化导入国标测试文档，仅首次运行执行一次"""
    docs = [
        "绿茶国标：水分≤7.5%，灰分4.0~6.0%，茶多酚17‑22%。总分≥90特级，75‑89一级，60‑74二级，低于60不合格。",
        "红茶国标：水分≤8.0%，灰分4.5‑6.5%，茶多酚10‑18%。总分≥90特级，75‑89一级，60‑74二级，低于60不合格。"
    ]
    embeds = embedding_model.encode(docs)
    coll.insert([docs, embeds])
    coll.flush()
    print("国标文档已经导入Milvus向量库")

# ---------------------- Skill1 真实RAG检索工具 ----------------------
def rag_search(tea_type: str):
    """
    【Skill‑RAG】从Milvus向量库检索茶叶国标文档
    :param tea_type: green绿茶 / black红茶
    """
    coll = Collection(COLLECTION_NAME)
    coll.load()
    query_text = f"{tea_type} 茶叶国家标准"
    vec = embedding_model.encode([query_text])
    search_params = {"metric_type":"L2","params":{"nprobe":10}}
    res = coll.search(vec, "embedding", search_params, limit=1, output_fields=["content"])
    if len(res[0])>0:
        doc = res[0][0].entity.get("content")
        return {"standard":doc}
    else:
        return {"standard":"未检索到该品类国标文档"}

# ---------------------- Skill2 茶叶检测工具 ----------------------
def tea_detect_tool(moisture: float, ash: float, tea_polyphenols: float):
    """
    【Skill‑业务检测】茶叶品质检测
    :param moisture: 水分 %
    :param ash: 灰分 %
    :param tea_polyphenols: 茶多酚 %
    """
    score = 0
    if 5.0 <= moisture <= 7.5:
        score += 30
    if 4.0 <= ash <= 6.0:
        score += 30
    if 17 <= tea_polyphenols <= 22:
        score += 40

    grade = "不合格"
    if score >= 90:
        grade = "特级"
    elif score >= 75:
        grade = "一级"
    elif score >= 60:
        grade = "二级"

    return {
        "moisture": moisture,
        "ash": ash,
        "tea_polyphenols": tea_polyphenols,
        "score": score,
        "grade": grade
    }

# 工具注册列表
tools = [
    {
        "type": "function",
        "function": {
            "name": "rag_search",
            "description": "RAG检索工具，查询茶叶国家标准。参数tea_type：green绿茶 / black红茶。Workflow强制优先调用此工具获取国标。",
            "parameters": {
                "type": "object",
                "required": ["tea_type"],
                "properties": {
                    "tea_type": {"type": "string", "enum": ["green", "black"], "description": "茶叶类型 green绿茶 black红茶"}
                }
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "tea_detect_tool",
            "description": "茶叶指标检测，【Workflow约束】必须已经成功调用rag_search拿到国标文档，并且三项指标齐全，才能调用本工具。",
            "parameters": {
                "type": "object",
                "required": ["moisture", "ash", "tea_polyphenols"],
                "properties": {
                    "moisture": {"type": "number"},
                    "ash": {"type": "number"},
                    "tea_polyphenols": {"type": "number"}
                }
            }
        }
    }
]
tool_map = {
    "rag_search": rag_search,
    "tea_detect_tool": tea_detect_tool
}

# ---------------------- Workflow+ReAct Agent ----------------------
class WorkflowTeaAgent:
    def __init__(self):
        # 工作流状态变量，硬编码Workflow约束
        self.workflow_state = {
            "tea_type": None,
            "has_standard": False,    # 是否已经获取国标（Workflow关键标记）
            "standard_content":"",
            "moisture": None,
            "ash": None,
            "tea_polyphenols": None
        }
        self.messages = [
            {
                "role": "system",
                "content": """
你是茶叶检测助手，严格遵守Workflow硬性规则：
规则1：必须先拿到茶叶品类，没有品类先询问用户。
规则2：**必须优先调用rag_search工具获取国标，没有国标绝对不能执行检测**。
规则3：国标获取成功之后，再收集三项指标：水分、灰分、茶多酚；缺失则主动提问补齐。
规则4：国标已拿到+三项指标齐全，才可以调用tea_detect_tool。
规则5：最后结合国标和检测结果生成完整报告。
禁止跳过步骤、颠倒顺序，禁止自己编造国标内容。
                """.strip()
            }
        ]

    def run(self, user_input: str):
        self.messages.append({"role": "user", "content": user_input})
        max_loop = 8
        loop = 0
        while loop < max_loop:
            loop +=1
            resp = client.chat.completions.create(
                model=MODEL,
                messages=self.messages,
                tools=tools,
                tool_choice="auto",
                temperature=0.1
            )
            msg = resp.choices[0].message

            # 不需要调用工具：反问用户（缺品类/缺指标）
            if not msg.tool_calls:
                self.messages.append(msg)
                return {"finished":True,"reply":msg.content}

            # 执行工具 Action
            for call in msg.tool_calls:
                func_name = call.function.name
                args = json.loads(call.function.arguments)
                print(f"\n【Action】调用工具:{func_name} 参数:{args}")
                func = tool_map[func_name]
                obs = func(**args)
                print(f"【Observation】工具返回 {obs}")

                # 更新Workflow状态
                if func_name == "rag_search":
                    self.workflow_state["has_standard"] = True
                    self.workflow_state["standard_content"] = obs["standard"]
                    self.workflow_state["tea_type"] = args["tea_type"]

                self.messages.append(msg)
                self.messages.append({
                    "role":"tool",
                    "tool_call_id":call.id,
                    "content":json.dumps(obs,ensure_ascii=False)
                })

            # Workflow约束：只有已经拿到国标，才允许结束并生成报告
            if func_name == "tea_detect_tool":
                if not self.workflow_state["has_standard"]:
                    return {"finished":True,"reply":"流程错误：尚未获取国标，不能执行检测。"}
                final_resp = client.chat.completions.create(model=MODEL,messages=self.messages,temperature=0.3)
                final_msg = final_resp.choices[0].message
                self.messages.append(final_msg)
                return {
                    "finished":True,
                    "detect_data":obs,
                    "standard":self.workflow_state["standard_content"],
                    "reply":final_msg.content
                }
        return {"finished":False,"reply":"任务循环超时"}


if __name__ == "__main__":
    # 初始化Milvus向量库
    coll = init_milvus()
    # 如果是空集合，导入测试国标文档（只运行一次）
    if coll.num_entities ==0:
        insert_docs(coll)

    print("==== Milvus‑RAG + Workflow约束茶叶Agent ====")
    agent = WorkflowTeaAgent()
    while True:
        inp = input("输入指令(exit重置):")
        if inp.strip().lower() == "exit":
            agent = WorkflowTeaAgent()
            print("会话重置\n")
            continue
        res = agent.run(inp)
        print("\nAgent回复：",res["reply"])
        if "detect_data" in res:
            print("检测结果：",res["detect_data"])
            print("查询到国标：",res["standard"])
        print("-"*70)

运行：python agent.py

两种测试案例：

案例1(参数齐全):
水分7.2，灰分5.1，茶多酚19.3

案例2(缺少参数):
帮我检测茶叶，水分7.3，灰分5.2

• 参数齐全 →Agent 自动调用tea_detect_tool →生成检测报告
• 参数不全 →Agent 反问你，补齐缺失指标

当前版本局限 & 后续优化方向

1. 当前记忆保存在内存，程序重启记忆丢失。生产环境要改成 Redis 持久化（每个 sessionId 对应一组 messages）
2. 当前是单会话；多用户并发需要每个用户新建独立TeaAgent实例
3. RAG 这里做简易内存模拟；后续可以把 rag_search 改成真实 Milvus 向量库检索。