AI Agent入门学习（二）
·
阶段二：将rag_search替换为真实 Milvus 向量库 RAG 检索
一、基础概念
RAG（检索增强生成）给 Agent 提供外部知识库，让智能体可以拿到私有、实时、业务文档数据，减少幻觉。 Agent 负责思考调度；
RAG 是一种专门用来查资料的 Skill（工具）
- LLM：模型训练时学到的知识，存在内部参数。知识会过时、不知道你的私有文档，很容易编造内容 (幻觉)
- RAG：把外部文档（PDF、Word、markdown、数据库文本）提前向量化存入向量数据库 (Milvus、Chroma)；用户提问时先检索相关片段，再交给大模型
放到你的茶叶项目：国标文件就是私有知识库，大模型训练数据不一定包含最新国标，不能直接硬编码写死在代码。
二、流程设计
- 用户：帮我检测绿茶
- Agent (Thought)：我需要绿茶的国标标准
- Agent (Action)：调用 RAG‑Skill，检索绿茶国标文档
- RAG 返回国标原文片段
- Agent (Observation)：拿到国标，接下来需要收集水分、灰分、茶多酚指标
- 收集齐全指标 →调用茶叶检测 Skill
- 结合国标文档 + 检测结果，生成合规检测报告
Workflow(整体业务流程蓝图)
└── Agent(ReAct循环调度大脑)
├── Skill1：RAG知识库检索工具
├── Skill2：茶叶检测业务工具
└── Skill3：生成PDF报告工具RAG 完整技术链路
文档 →文本分割 →Embedding 向量化 →存入向量库 (Milvus) →检索 →把相关片段拼入 Prompt →交给 LLM
实现思路：
1、当前模拟版 rag_search →改成真实 Milvus 向量库 RAG 工具
2、把 RAG 注册成 Agent 的一个 Skill
3、用 Workflow 约束流程：优先调用 RAG 拿到国标再做检测
本次改造要点
- 移除内存模拟知识库，接入 Milvus 向量数据库实现真实 RAG
- RAG 封装成标准 Skill (Tool) 注册进 Agent
- 使用硬编码简易 Workflow 强制约束：必须先查到国标，才能执行茶叶检测
- 底层模型沿用 DeepSeek，保留原有 ReAct Agent 框架
Milvus 本身只负责向量计算、索引、查询逻辑，它自己不存元数据、不存向量原始文件，必须依赖两个外部组件。
1、etcd：元数据数据库（管家）
相当于 Milvus 的配置 / 元信息仓库
etcd = Milvus 的小账本，记录所有表结构、索引信息、文件位置。
2、MinIO：向量文件存储（仓库）
Milvus 把真正的向量、索引文件，全部打包成文件丢给 MinIO 保存
Milvus 本身容器里面只放临时缓存，不持久化向量。
三、实现步骤
第一步：安装依赖包
pip install openai python-dotenv pymilvus sentence-transformers第二步：安装Milvus向量数据库
Milvus 官方没有原生 Windows 安装包，推荐优先用 Docker
步骤 1： 安装 Docker Desktop
下载地址：https://www.docker.com/products/docker-desktop/
步骤2：Docker‑Desktop 中安装启动 milvusdb/milvus:v2.4.5
Milvus 不能单独一个容器直接跑，它依赖 2 个组件：etcd(元数据) + minio(向量文件存储)，必须用 docker‑compose 一次性启动 3 个容器。 端口：19530(gRPC)、9091(http)
方式一：直接使用现成 docker‑compose.yml（指定 v2.4.5，推荐）
- 新建一个文件夹，例如
D:\docker\milvus245 - 在文件夹里面新建文件，名字：
docker-compose.yml完整内容直接复制下面：
version: '3.5'
services:
etcd:
container_name: milvus-etcd
image: quay.io/coreos/etcd:v3.5.5
environment:
- ETCD_AUTO_COMPACTION_MODE=revision
- ETCD_AUTO_COMPACTION_RETENTION=1000
- ETCD_QUOTA_BACKEND_BYTES=4294967296
- ETCD_ENABLE_V2=true
volumes:
- ./volumes/etcd:/etcd/data
command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd/data
healthcheck:
test: ["CMD", "etcdctl", "endpoint", "health"]
interval: 30s
timeout: 20s
retries: 3
minio:
container_name: milvus-minio
image: minio/minio:RELEASE.2023-03-20T20-16-18Z
environment:
MINIO_ACCESS_KEY: minioadmin
MINIO_SECRET_KEY: minioadmin
volumes:
- ./volumes/minio:/minio/data
command: minio server /minio/data
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
interval: 30s
timeout: 20s
retries: 3
standalone:
container_name: milvus-standalone
image: milvusdb/milvus:v2.4.5
command: ["milvus", "run", "standalone"]
environment:
ETCD_ENDPOINTS: etcd:2379
MINIO_ADDRESS: minio:9000
volumes:
- ./volumes/milvus:/var/lib/milvus
ports:
- "19530:19530"
- "9091:9091"
depends_on:
- "etcd"
- "minio"
networks:
default:
name: milvus
3.打开 PowerShell /cmd，cd 进入这个目录
cd D:\docker\milvus245
4.后台启动
docker compose up -d
第一次会自动拉取三个镜像：
milvusdb/milvus:v2.4.5quay.io/coreos/etcd:v3.5.5minio/minio:RELEASE.2023‑03‑20T20‑16‑18Z
查看容器
docker compose ps三个容器状态必须都是 running
健康检查 http 接口
浏览器访问：http://127.0.0.1:9091/healthz
返回 ok 代表 Milvus 已经就绪
数据持久化：全部数据保存在当前目录 volumes 文件夹下
如出现报错信息：
Error response from daemon: failed to resolve reference "docker.io/milvusdb/milvus:v2.4.5": failed to do request: Head "[https://docker.mirrors.ustc.edu.cn/v2/milvusdb/milvus/manifests/v2.4.5?ns=docker.io](https://link.wtturl.cn/?target=https%3A%2F%2Fdocker.mirrors.ustc.edu.cn%2Fv2%2Fmilvusdb%2Fmilvus%2Fmanifests%2Fv2.4.5%3Fns%3Ddocker.io&scene=im&aid=582478&lang=zh)": dialing [docker.mirrors.ustc.edu.cn:443](https://link.wtturl.cn/?target=https%3A%2F%2Fdocker.mirrors.ustc.edu.cn%3A443&scene=im&aid=582478&lang=zh) container via direct connection because Docker Desktop has no HTTPS proxy: connecting to [docker.mirrors.ustc.edu.cn:443](https://link.wtturl.cn/?target=https%3A%2F%2Fdocker.mirrors.ustc.edu.cn%3A443&scene=im&aid=582478&lang=zh): dial tcp: lookup [docker.mirrors.ustc.edu.cn](https://link.wtturl.cn/?target=https%3A%2F%2Fdocker.mirrors.ustc.edu.cn&scene=im&aid=582478&lang=zh): no such host报错原因：拉取镜像失败：docker.mirrors.ustc.edu.cn 已经下线、域名解析失败。
解决方案：
第一步：修改 Docker‑Desktop 镜像配置
1. 打开 Docker Desktop → 右上角齿轮 Settings → Docker Engine
2. 把里面全部 JSON 删掉，粘贴下面这份可用配置
{
"registry-mirrors": [
"https://docker.m.daocloud.io"
],
"builder": {
"gc": {
"defaultKeepStorage": "20GB",
"enabled": true
}
},
"experimental": false
}3. 点 Apply & Restart，Docker 自动重启
DaoCloud 是目前国内相对稳定的公共加速源
第三步：.env 配置文件
# LLM
DEEPSEEK_API_KEY=sk-你的deepseek密钥
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-v4-flash
# Milvus配置
MILVUS_HOST=127.0.0.1
MILVUS_PORT=19530
MILVUS_COLLECTION=tea_standard第四步：完整代码 agent_rag.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)
报错异常处理：
'[WinError 10060] 由于连接方在一段时间后没有正确答复或连接的主机没有反应，连接尝试失败。' thrown while requesting HEAD [https://huggingface.co/sentence-transformers/all-MiniLM-L6-v2/resolve/main/./modules.json](https://link.wtturl.cn/?target=https%3A%2F%2Fhuggingface.co%2Fsentence-transformers%2Fall-MiniLM-L6-v2%2Fresolve%2Fmain%2F.%2Fmodules.json&scene=im&aid=582478&lang=zh)原因：WinError 10060 = 国内网络访问 huggingface.co 超时。
程序启动 SentenceTransformer("all‑MiniLM‑L6‑v2") 需要从 HuggingFace 下载模型文件 modules.json，直接连境外网站被卡住。
SentenceTransformer("all‑MiniLM‑L6‑v2") 这句代码，本质是加载一个预训练向量模型，而 modules.json 是这个模型的配置清单文件，没有它模型就不知道自己长什么样、没法正常工作。
方案 1（最简单，临时生效，优先试）
打开 PowerShell，先执行这条镜像环境变量命令
$env:HF_ENDPOINT="https://hf-mirror.com"
python tea_agent_milvus_workflow.py方案 2 永久设置环境变量
- Win+R 输入
SystemPropertiesAdvanced回车 - 环境变量 → 用户变量 →新建
- 变量名：
HF_ENDPOINT - 变量值：
https://hf-mirror.com
- 变量名：
- 全部窗口关掉，重新开一个新终端再运行代码
方案 3：提前手动下载模型到本地，离线加载
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
1
0- 0
已为社区贡献3条内容
所有评论(0)