AgentV2: Medical AI Agent with multi-route RAG

.gitattributes

@@ -0,0 +1,5 @@
+*.pdf filter=lfs diff=lfs merge=lfs -text
+*.db filter=lfs diff=lfs merge=lfs -text
+*.xlsx filter=lfs diff=lfs merge=lfs -text
+*.jsonl filter=lfs diff=lfs merge=lfs -text
+*.json filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1 @@
+.env

.ipynb_checkpoints/test-checkpoint.py

@@ -0,0 +1,21 @@
+import requests
+import time
+import json
+
+url = "http://0.0.0.0:8103/"
+data = {"question": "平日里蜂蜜加白醋一起喝有什么疗效？"}
+#data = {"question": "听说用酸枣仁泡水喝能养生,是真的吗？"}
+
+start_time = time.time()
+
+data = json.dumps(data)
+
+# 向服务发送请求
+res = requests.post(url, data=data)
+
+cost_time = time.time() - start_time
+
+print('单次查询的耗时:', cost_time, 's')
+
+res = json.loads(res.text)
+print(res)

agent.py

@@ -0,0 +1,131 @@
+import os
+import uvicorn
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+import json
+import datetime
+from langchain_milvus import Milvus, BM25BuiltInFunction
+from model import OpenAIEmbeddings
+from dotenv import load_dotenv
+
+# 加载 .env 文件中的环境变量, 隐藏 API Keys
+load_dotenv()
+
+os.environ["TOKENIZERS_PARALLELISM"] = "false"
+
+# ====== 开关: True=本地vLLM, False=ChatGPT ======
+USE_LOCAL_LLM = False
+
+if USE_LOCAL_LLM:
+    from model import create_local_llm_client as create_client
+    from model import generate_local_answer as generate_answer
+else:
+    from model import create_chatgpt_client as create_client
+    from model import generate_chatgpt_answer as generate_answer
+
+client_llm = create_client()
+print(f"创建 {'本地 vLLM' if USE_LOCAL_LLM else 'ChatGPT'} 客户端成功......")
+
+app = FastAPI()
+
+# 允许所有域的请求
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 创建 Embedding 模型
+embedding_model = OpenAIEmbeddings()
+print("创建 Embedding 模型成功......")
+
+# 设置默认的 Milvus 数据库文件路径
+URI = "./milvus_agent.db"
+
+# 创建 Milvus 连接
+milvus_vectorstore = Milvus(
+    embedding_function=embedding_model,
+    builtin_function=BM25BuiltInFunction(),
+    vector_field=["dense", "sparse"],
+    index_params=[
+        {
+            "metric_type": "IP",
+            "index_type": "IVF_FLAT",
+        },
+        {
+            "metric_type": "BM25",
+            "index_type": "SPARSE_INVERTED_INDEX"
+        }
+    ],
+    connection_args={"uri": URI},
+)
+
+retriever = milvus_vectorstore.as_retriever()
+print("创建 Milvus 连接成功......")
+
+
+def format_docs(docs):
+    return "\n\n".join(doc.page_content for doc in docs)
+
+
+@app.post("/")
+async def chatbot(request: Request):
+    global milvus_vectorstore, retriever
+
+    json_post_raw = await request.json()
+    json_post = json.dumps(json_post_raw)
+    json_post_list = json.loads(json_post)
+
+    query = json_post_list.get('question')
+
+    # 召回 & 排序
+    # 在集合中搜索问题并检索语义 top-10 匹配项, 而且已经配置了 reranker 的处理, 采用RRF算法
+    recall_rerank_milvus = milvus_vectorstore.similarity_search(
+        query,
+        k=10,
+        ranker_type="rrf",
+        ranker_params={"k": 100}
+    )
+
+    if recall_rerank_milvus:
+        # 检索结果存放在列表中
+        context = format_docs(recall_rerank_milvus)
+    else:
+        context = []
+
+    # 为LLM定义系统和用户提示, 这个提示是由从Milvus检索到的文档组装而成的.
+    SYSTEM_PROMPT = """
+    System: 你是一个非常得力的医学助手, 你可以通过从数据库中检索出的信息找到问题的答案.
+    """
+
+    USER_PROMPT = f"""
+    User: 利用介于<context>和</context>之间的从数据库中检索出的信息来回答问题, 具体的问题介于<question>和</question>之间. 如果提供的信息为空, 则按照你的经验知识来给出尽可能严谨准确的回答, 不知道的时候坦诚的承认不了解, 不要编造不真实的信息.
+    <context>
+    {context}
+    </context>
+
+    <question>
+    {query}
+    </question>
+    """
+
+    # 使用 LLM 模型, 根据提示生成回复 (根据 USE_LOCAL_LLM 开关自动选择 vLLM 或 ChatGPT)
+    response = generate_answer(client_llm, SYSTEM_PROMPT + USER_PROMPT)
+
+    now = datetime.datetime.now()
+    time = now.strftime("%Y-%m-%d %H:%M:%S")
+
+    answer = {
+        "response": response,
+        "status": 200,
+        "time": time
+    }
+
+    return answer
+
+
+if __name__ == '__main__':
+    # 主函数中直接启动fastapi服务
+    uvicorn.run(app, host='0.0.0.0', port=8103, workers=1)

agent2.py

@@ -0,0 +1,206 @@
+import os
+import uvicorn
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+import json
+import datetime
+from openai import OpenAI
+from langchain_milvus import Milvus, BM25BuiltInFunction
+from vector import OpenAIEmbeddings
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from langchain_core.stores import InMemoryStore
+from langchain_classic.retrievers.parent_document_retriever import ParentDocumentRetriever
+from dotenv import load_dotenv
+
+# 加载 .env 文件中的环境变量, 隐藏 API Keys
+load_dotenv()
+
+os.environ["TOKENIZERS_PARALLELISM"] = "false"
+app = FastAPI()
+
+
+# ============================================================
+# OpenAI LLM 客户端封装 (替代讲义中的 DeepSeek)
+# ============================================================
+
+def create_openai_client():
+    """创建 OpenAI 客户端"""
+    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+    return client
+
+
+def generate_openai_answer(client, prompt):
+    """使用 OpenAI 生成回复"""
+    response = client.chat.completions.create(
+        model="gpt-4o-mini",
+        messages=[
+            {"role": "user", "content": prompt}
+        ],
+        temperature=0.7,
+    )
+    return response.choices[0].message.content
+
+
+# 允许所有域的请求
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 创建 Embedding 模型
+embedding_model = OpenAIEmbeddings()
+print("创建 Embedding 模型成功......")
+
+# 设置默认的 Milvus 数据库文件路径
+URI = "./milvus_agent.db"
+URI1 = "./pdf_agent.db"
+
+# 创建 Milvus 连接
+milvus_vectorstore = Milvus(
+    embedding_function=embedding_model,
+    builtin_function=BM25BuiltInFunction(),
+    vector_field=["dense", "sparse"],
+    index_params=[
+        {
+            "metric_type": "IP",
+            "index_type": "IVF_FLAT",
+        },
+        {
+            "metric_type": "BM25",
+            "index_type": "SPARSE_INVERTED_INDEX"
+        }
+    ],
+    connection_args={"uri": URI},
+)
+
+retriever = milvus_vectorstore.as_retriever()
+print("创建 Milvus 连接成功......")
+
+
+docstore = InMemoryStore()
+
+# 文本分割器
+child_splitter = RecursiveCharacterTextSplitter(
+                    chunk_size=200,
+                    chunk_overlap=50,
+                    length_function=len,
+                    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
+                )
+
+parent_splitter = RecursiveCharacterTextSplitter(
+                    chunk_size=1000,
+                    chunk_overlap=200
+                )
+
+pdf_vectorstore = Milvus(
+    embedding_function=embedding_model,
+    builtin_function=BM25BuiltInFunction(),
+    vector_field=["dense", "sparse"],
+    index_params=[
+        {
+            "metric_type": "IP",
+            "index_type": "IVF_FLAT",
+        },
+        {
+            "metric_type": "BM25",
+            "index_type": "SPARSE_INVERTED_INDEX"
+        }
+    ],
+    connection_args={"uri": URI1},
+    consistency_level="Bounded",
+    drop_old=False,
+)
+
+# 设置父子文档检索器
+parent_retriever = ParentDocumentRetriever(
+    vectorstore=pdf_vectorstore,
+    docstore=docstore,
+    child_splitter=child_splitter,
+    parent_splitter=parent_splitter,
+)
+
+print("创建 Parent Milvus 连接成功......")
+
+# 创建大语言模型, 采用 OpenAI 
+client_llm = create_openai_client()
+print("创建 OpenAI LLM 成功......")
+
+
+def format_docs(docs):
+    return "\n\n".join(doc.page_content for doc in docs)
+
+
+@app.post("/")
+async def chatbot(request: Request):
+    global milvus_vectorstore, retriever
+
+    json_post_raw = await request.json()
+    json_post = json.dumps(json_post_raw)
+    json_post_list = json.loads(json_post)
+
+    query = json_post_list.get('question')
+
+    # 1: Milvus 召回 & 排序
+    # 在集合中搜索问题并检索语义 top-10 匹配项, 而且已经配置了 reranker 的处理, 采用RRF算法
+    recall_rerank_milvus = milvus_vectorstore.similarity_search(
+                            query,
+                            k=10,
+                            ranker_type="rrf",
+                            ranker_params={"k": 100}
+                        ) 
+
+    if recall_rerank_milvus:
+        # 检索结果存放在列表中
+        context = [r.page_content for r in recall_rerank_milvus]
+        context = format_docs(recall_rerank_milvus)
+    else:
+        context = ""
+
+    # 2: PDF 文档的 Milvus 召回
+    # 父文档检索器按照query进行召回
+    res = ""
+    retrieved_docs = parent_retriever.invoke(query)
+
+    if retrieved_docs is not None and len(retrieved_docs) >= 1:
+        res = retrieved_docs[0].page_content
+        print("PDF res: ", res)
+
+    context = context + "\n" + res
+
+    # 为LLM定义系统和用户提示, 这个提示是由从Milvus检索到的文档组装而成的.
+    SYSTEM_PROMPT = """
+        System: 你是一个非常得力的医学助手, 你可以通过从数据库中检索出的信息找到问题的答案.
+        """
+
+    USER_PROMPT = f"""
+        User: 利用介于<context>和</context>之间的从数据库中检索出的信息来回答问题, 具体的问题介于<question>和</question>之间. 如果提供的信息为空, 则按照你的经验知识来给出尽可能严谨准确的回答, 不知道的时候坦诚的承认不了解, 不要编造不真实的信息.
+        <context>
+        {context}
+        </context>
+
+        <question>
+        {query}
+        </question>
+        """
+
+    # 3. 使用 OpenAI 最新版本模型, 根据提示生成回复
+    response = generate_openai_answer(client_llm, SYSTEM_PROMPT + USER_PROMPT.format(context, query))
+
+    now = datetime.datetime.now()
+    time = now.strftime("%Y-%m-%d %H:%M:%S")
+
+    answer = {
+        "response": response,
+        "status": 200,
+        "time": time
+    }
+
+    return answer
+
+
+if __name__ == '__main__':
+    # 主函数中直接启动fastapi服务
+    uvicorn.run(app, host='0.0.0.0', port=8103, workers=1)

data/dialog.jsonl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:879089741db2827a13e1a6f61716be405a95b40a168623a577ad0f22615c7911
+size 15035121

data/train.json

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:cbd0d98a702e355753bf9f1a2f14f034323464156654ca0cd7d1d8b2e97f6864
+size 1834602

milvus_agent.db

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3fb6f3a55a098a6eac5d6b916bb55ac65827941084f862007ded0669e2671f8e
+size 28672

model.py

@@ -0,0 +1,102 @@
+import os
+from openai import OpenAI
+from langchain.embeddings.base import Embeddings
+from dotenv import load_dotenv
+
+# 加载 .env 文件中的环境变量, 隐藏 API Keys
+load_dotenv()
+
+
+# ============================================================
+# 模型1: 嵌入模型, 采用 OpenAI text-embedding-3-small
+# ============================================================
+
+class OpenAIEmbeddings(Embeddings):
+    """基于 OpenAI Embedding API 的自定义嵌入类"""
+
+    def __init__(self):
+        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+
+    def embed_documents(self, texts):
+        embeddings = []
+        for text in texts:
+            response = self.client.embeddings.create(
+                model="text-embedding-3-small",
+                input=[text],
+            )
+            embeddings.append(response.data[0].embedding)
+        return embeddings
+
+    def embed_query(self, text):
+        # 查询文档
+        return self.embed_documents([text])[0]
+
+
+# ============================================================
+# 模型2: 本地大语言模型, 通过 vLLM 服务 (OpenAI 兼容接口)
+# 启动命令: vllm serve ./Qwen3-Next-80B-A3B-Thinking-AWQ-4bit --dtype auto --trust-remote-code --max-model-len 4096 --port 8000
+# ============================================================
+
+VLLM_MODEL_NAME = "./Qwen3-Next-80B-A3B-Thinking-AWQ-4bit"
+VLLM_BASE_URL = "http://localhost:8000/v1"
+
+
+def create_local_llm_client():
+    """创建本地 vLLM 客户端 (OpenAI 兼容接口)"""
+    client = OpenAI(api_key="none", base_url=VLLM_BASE_URL)
+    return client
+
+
+def generate_local_answer(client, question):
+    """调用本地 vLLM 生成回答"""
+    response = client.chat.completions.create(
+        model=VLLM_MODEL_NAME,
+        messages=[
+            {"role": "system", "content": "你是一个能力非常强大的助手."},
+            {"role": "user", "content": question}
+        ],
+        max_tokens=2048,
+        stream=False
+    )
+    return response.choices[0].message.content
+
+
+# ============================================================
+# 模型3: 远程大语言模型, 采用 ChatGPT (OpenAI API)
+# ============================================================
+
+def create_chatgpt_client():
+    """创建 OpenAI ChatGPT 客户端"""
+    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+    return client
+
+
+def generate_chatgpt_answer(client, question):
+    """调用 ChatGPT 生成回答"""
+    response = client.chat.completions.create(
+        model="gpt-4o",
+        messages=[
+            {"role": "system", "content": "你是一个能力非常强大的助手."},
+            {"role": "user", "content": question}
+        ],
+        stream=False
+    )
+    return response.choices[0].message.content
+
+
+# ============================================================
+# 测试入口
+# ============================================================
+
+if __name__ == "__main__":
+    # 测试本地 vLLM 模型
+    client = create_local_llm_client()
+    output = generate_local_answer(client, "你好啊,千与千寻")
+    print('-' * 50)
+    print(output)
+
+    # 测试远程 ChatGPT
+    #client = create_chatgpt_client()
+    #output = generate_chatgpt_answer(client, "你好啊,千与千寻")
+    #print('-' * 50)
+    #print(output)

pdf_agent.db

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:dd578ee2a4c3f4c9a3834d1486815e1f0002cd5b1de31784fa714cea1168f134
+size 452308992

pdf_documents/01.内科学_第9版_全书签_可复制检索.pdf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3cf35d1734c7816e3ec58545d28ce67d5155c962a0472f3f5da1357b6a10b41e
+size 486582267

pdf_documents/02.外科学_第9版_全书签_可复制检索.pdf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:fdad0ad721ab78a5bf07507ff22d0e25ca8ab0e0f7041b770c7d35f90194711f
+size 433303834

pdf_documents/03.妇产科学_第9版_全书签_可复制检索.pdf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:10f84dadb1feba7c255d95d02df9447e2d48e828dec7d99d1d152e8a763a389a
+size 248708226

pdf_documents/04.儿科学_第9版_全书签_可复制检索.pdf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:28b99f77d6f44dc9e43697cf58168cdca9db74965efaa8764a89567066c13010
+size 102781905

pdf_documents/05.神经病学_第9版_全书签_可复制检索.pdf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:064bb8ce4efa9715680713906992e05dc1a99dc8feabefbe8edf13bea53022e5
+size 812868931

pdf_documents/06.系统解剖学_第9版_全书签_可复制检索.pdf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:8356c6cacfa84b04df444c3714e8391790da78682e82c4b4a484f6f0975bd617
+size 128118784

pdf_documents/07.局部解剖学_第9版_全书签_可部分复制检索.pdf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:cf5762f141bcc8fc4ae2ae0540d64001293e30cd9e2a72d80d1946e798b56286
+size 103522604

pdf_documents/08.组织学与胚胎学_第9版_全书签_可复制检索.pdf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:eac553218200f3ca71c49cae4c56cfc323e9f11f3d3f03340d8f33ee53406e42
+size 49850201

pdf_documents/09.生物化学与分子生物学_第9版_全书签_可复制检索.pdf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:63bb2e1cf400dc6727834a2d0d8a25a8aa12795586f3b7b16eaed5f7f04db214
+size 284284534

pdf_documents/10.生理学(可复制).pdf

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:9d6423f99687a487e52479589a81ce9cea69cc4832d9c3b065b8f44d8a00111c
+size 412517165

pdf_output/pdf_detailed_text.xlsx

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d2fb7a2bba1bc86d83071e35c1f523341ed14162e9b8d9f2f8a56610d3e2a6a7
+size 6481056

pdf_output/pdf_extraction_summary.xlsx

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:13c43bc65f29f3b878879ea496d369e7b14b238efc8f54a09a0019e739375bac
+size 5700

pdf_output/pdf_processing.log

@@ -0,0 +1,24 @@
+2026-02-09 02:07:34,106 - INFO - 在 ./pdf_documents 中找到 10 个PDF文件
+2026-02-09 02:07:34,109 - INFO - 处理进度: 1/10 - 01.内科学_第9版_全书签_可复制检索.pdf
+2026-02-09 02:10:28,298 - INFO - 成功处理: 01.内科学_第9版_全书签_可复制检索.pdf - 962 页
+2026-02-09 02:10:28,312 - INFO - 处理进度: 2/10 - 02.外科学_第9版_全书签_可复制检索.pdf
+2026-02-09 02:13:01,894 - INFO - 成功处理: 02.外科学_第9版_全书签_可复制检索.pdf - 830 页
+2026-02-09 02:13:01,911 - INFO - 处理进度: 3/10 - 03.妇产科学_第9版_全书签_可复制检索.pdf
+2026-02-09 02:14:33,064 - INFO - 成功处理: 03.妇产科学_第9版_全书签_可复制检索.pdf - 495 页
+2026-02-09 02:14:33,080 - INFO - 处理进度: 4/10 - 04.儿科学_第9版_全书签_可复制检索.pdf
+2026-02-09 02:15:26,939 - INFO - 成功处理: 04.儿科学_第9版_全书签_可复制检索.pdf - 478 页
+2026-02-09 02:15:26,955 - INFO - 处理进度: 5/10 - 05.神经病学_第9版_全书签_可复制检索.pdf
+2026-02-09 02:16:26,155 - INFO - 成功处理: 05.神经病学_第9版_全书签_可复制检索.pdf - 494 页
+2026-02-09 02:16:26,162 - INFO - 处理进度: 6/10 - 06.系统解剖学_第9版_全书签_可复制检索.pdf
+2026-02-09 02:16:29,044 - ERROR - 处理文件失败 pdf_documents/06.系统解剖学_第9版_全书签_可复制检索.pdf: Unexpected EOF
+2026-02-09 02:16:29,044 - INFO - 处理进度: 7/10 - 07.局部解剖学_第9版_全书签_可部分复制检索.pdf
+2026-02-09 02:16:32,187 - INFO - 成功处理: 07.局部解剖学_第9版_全书签_可部分复制检索.pdf - 318 页
+2026-02-09 02:16:32,192 - INFO - 处理进度: 8/10 - 08.组织学与胚胎学_第9版_全书签_可复制检索.pdf
+2026-02-09 02:18:05,497 - INFO - 成功处理: 08.组织学与胚胎学_第9版_全书签_可复制检索.pdf - 300 页
+2026-02-09 02:18:05,501 - INFO - 处理进度: 9/10 - 09.生物化学与分子生物学_第9版_全书签_可复制检索.pdf
+2026-02-09 02:19:53,250 - INFO - 成功处理: 09.生物化学与分子生物学_第9版_全书签_可复制检索.pdf - 559 页
+2026-02-09 02:19:53,270 - INFO - 处理进度: 10/10 - 10.生理学(可复制).pdf
+2026-02-09 02:20:26,372 - INFO - 成功处理: 10.生理学(可复制).pdf - 466 页
+2026-02-09 02:20:27,907 - INFO - 结果已保存到 pdf_output
+2026-02-09 02:20:27,913 - INFO - 处理完成: 9/10 个文件成功
+2026-02-09 02:20:27,914 - INFO - 平均每文件: 707778 字符, 2.0 个表格

pdf_output/progress_batch_10.csv

@@ -0,0 +1,11 @@
+file_name,status,pages_processed
+01.内科学_第9版_全书签_可复制检索.pdf,Success,962
+02.外科学_第9版_全书签_可复制检索.pdf,Success,830
+03.妇产科学_第9版_全书签_可复制检索.pdf,Success,495
+04.儿科学_第9版_全书签_可复制检索.pdf,Success,478
+05.神经病学_第9版_全书签_可复制检索.pdf,Success,494
+06.系统解剖学_第9版_全书签_可复制检索.pdf,Error,0
+07.局部解剖学_第9版_全书签_可部分复制检索.pdf,Success,318
+08.组织学与胚胎学_第9版_全书签_可复制检索.pdf,Success,300
+09.生物化学与分子生物学_第9版_全书签_可复制检索.pdf,Success,559
+10.生理学(可复制).pdf,Success,466

preprocess.py

@@ -0,0 +1,259 @@
+import os
+import glob
+import logging
+import pandas as pd
+from tqdm import tqdm
+from typing import List, Dict, Optional
+from pathlib import Path
+import pdfplumber
+
+
+# 工业级PDF批量处理器, 生产一线级别的代码
+class PDFBatchProcessor:
+    def __init__(self, output_dir: str = "./output"):
+        self.output_dir = Path(output_dir)
+        self.output_dir.mkdir(exist_ok=True)
+
+        # 配置日志系统
+        logging.basicConfig(
+            level=logging.INFO,
+            format='%(asctime)s - %(levelname)s - %(message)s',
+            handlers=[
+                logging.FileHandler(self.output_dir / "pdf_processing.log"),
+                logging.StreamHandler()
+            ]
+        )
+        self.logger = logging.getLogger(__name__)
+
+    # 查找指定路径下的所有PDF文件
+    def find_pdf_files(self, input_path: str) -> List[Path]:
+        path = Path(input_path)
+        if path.is_file() and path.suffix.lower() == '.pdf':
+            return [path]
+        elif path.is_dir():
+            # 递归查找所有PDF文件
+            pdf_files = list(path.glob("**/*.pdf"))
+            self.logger.info(f"在 {input_path} 中找到 {len(pdf_files)} 个PDF文件")
+            return pdf_files
+        else:
+            raise ValueError(f"路径不存在,或不是PDF文件: {input_path}")
+
+    # 提取单个PDF文件的内容
+    def extract_pdf_content(self,
+                            pdf_path: Path,
+                            extract_text: bool = True,
+                            extract_tables: bool = True,
+                            table_settings: Optional[dict] = None) -> Dict:
+        """
+        Args:
+            pdf_path: PDF文件路径
+            extract_text: 是否提取文本
+            extract_tables: 是否提取表格
+            table_settings: 表格提取配置
+        """
+        result = {
+            "file_name": pdf_path.name,
+            "file_path": str(pdf_path),
+            "metadata": {},
+            "pages": [],
+            "error": None
+        }
+
+        try:
+            with pdfplumber.open(pdf_path) as pdf:
+                # 提取元数据
+                result["metadata"] = pdf.metadata
+
+                for page_num, page in enumerate(pdf.pages, 1):
+                    page_result = {"page_number": page_num, "text": "", "tables": []}
+
+                    # 提取文本
+                    if extract_text:
+                        try:
+                            # 布局模式根据需求调整
+                            text = page.extract_text(layout=False)
+                            page_result["text"] = text if text else ""
+                        except Exception as e:
+                            self.logger.warning(f"页面 {page_num} 文本提取失败: {str(e)}")
+                            pass
+
+                    # 提取表格
+                    if extract_tables:
+                        try:
+                            tables = page.extract_tables(table_settings or {})
+                            if tables:
+                                page_result["tables"] = tables
+                        except Exception as e:
+                            self.logger.warning(f"页面 {page_num} 表格提取失败: {str(e)}")
+                            pass
+
+                    # 添加当前页面page的提取结果
+                    result["pages"].append(page_result)
+
+                # 单一PDF文档提取完毕后, 写日志处理
+                self.logger.info(f"成功处理: {pdf_path.name} - {len(pdf.pages)} 页")
+
+        # 单一PDF文档提取失败后, 写日志处理
+        except Exception as e:
+            # 明确记录一下哪篇PDF文档处理失败, 并记录失败原因, 便于后续回溯与 "bad case分析"
+            error_msg = f"处理文件失败 {pdf_path}: {str(e)}"
+            result["error"] = error_msg
+            self.logger.error(error_msg)
+
+        return result
+
+    # 批量处理PDF文件
+    def process_batch(self, pdf_files: List[Path],
+                      save_format: str = "excel",
+                      **extract_kwargs) -> pd.DataFrame:
+        """
+        Args:
+            pdf_files: PDF文件列表
+            save_format: 保存格式 (excel, csv, parquet)
+            **extract_kwargs: 提取参数
+        """
+        all_results = []
+
+        for i, pdf_file in tqdm(enumerate(pdf_files, 1)):
+            self.logger.info(f"处理进度: {i}/{len(pdf_files)} - {pdf_file.name}")
+
+            result = self.extract_pdf_content(pdf_file, **extract_kwargs)
+            all_results.append(result)
+
+            # 实时保存进度 (针对大批量处理)
+            if i % 10 == 0:
+                self._save_intermediate_results(all_results, f"batch_{i}")
+
+        # 保存最终结果
+        return self._save_results(all_results, save_format)
+
+    # 保存处理结果
+    def _save_results(self, results: List[Dict], format: str) -> pd.DataFrame:
+        # 扁平化结果, 以��保存
+        flat_data = []
+
+        for result in results:
+            if result["error"]:
+                flat_data.append(
+                    {
+                        "file_name": result["file_name"],
+                        "status": "Error",
+                        "error_message": result["error"],
+                        "page_count": 0,
+                        "text_length": 0,
+                        "table_count": 0
+                    }
+                )
+                continue
+
+            total_text = ""
+            total_tables = 0
+
+            for page in result["pages"]:
+                total_text += page["text"]
+                total_tables += len(page["tables"])
+
+            flat_data.append({
+                "file_name": result["file_name"],
+                "status": "Success",
+                "error_message": "",
+                "page_count": len(result["pages"]),
+                "text_length": len(total_text),
+                "table_count": total_tables,
+                "author": result["metadata"].get("Author", ""),
+                "creation_date": result["metadata"].get("CreationDate", "")
+            })
+
+        # for循环处理完毕后, 所有数据封装成 Pandas 的 DataFrame 格式
+        df = pd.DataFrame(flat_data)
+
+        # 根据格式保存
+        if format.lower() == "excel":
+            df.to_excel(self.output_dir / "pdf_extraction_summary.xlsx", index=False)
+
+            # 同时保存详细文本内容
+            detailed_results = []
+            for result in results:
+                if not result["error"]:
+                    for page in result["pages"]:
+                        if page["text"]:
+                            detailed_results.append({
+                                "file_name": result["file_name"],
+                                "page_number": page["page_number"],
+                                "text_content": page["text"]
+                            })
+
+            if detailed_results:
+                pd.DataFrame(detailed_results).to_excel(
+                    self.output_dir / "pdf_detailed_text.xlsx", index=False
+                )
+
+        elif format.lower() == "csv":
+            df.to_csv(self.output_dir / "pdf_extraction_summary.csv", index=False)
+
+        self.logger.info(f"结果已保存到 {self.output_dir}")
+        return df
+
+    # 保存中间结果 (工业界一线生产环境, 异常因素很多, 防止处理中断丢失数据)
+    def _save_intermediate_results(self, results: List[Dict], batch_name: str):
+        try:
+            temp_df = pd.DataFrame([{
+                "file_name": r["file_name"],
+                "status": "Error" if r["error"] else "Success",
+                "pages_processed": len(r["pages"])
+            } for r in results])
+
+            temp_df.to_csv(self.output_dir / f"progress_{batch_name}.csv", index=False)
+        except Exception as e:
+            self.logger.warning(f"保存中间结果失败: {str(e)}")
+
+
+# 高级表格提取配置
+ADVANCED_TABLE_SETTINGS = {
+    "vertical_strategy": "lines",
+    "horizontal_strategy": "lines",
+    "snap_tolerance": 4,
+    "join_tolerance": 10,
+    "edge_min_length": 3,
+    "min_words_vertical": 2,
+    "min_words_horizontal": 1
+}
+
+
+def main():
+    # 实例化PDF处理器对象
+    processor = PDFBatchProcessor(output_dir="./pdf_output")
+
+    try:
+        # 查找PDF文件
+        pdf_files = processor.find_pdf_files("./pdf_documents")
+
+        if not pdf_files:
+            processor.logger.warning("未找到PDF文件")
+            return
+
+        # 批量处理
+        results_df = processor.process_batch(
+            pdf_files,
+            save_format="excel",
+            extract_text=True,
+            extract_tables=True,
+            table_settings=ADVANCED_TABLE_SETTINGS
+        )
+
+        # 打印摘要统计
+        success_count = len(results_df[results_df["status"] == "Success"])
+        processor.logger.info(f"处理完成: {success_count}/{len(pdf_files)} 个文件成功")
+
+        if success_count > 0:
+            avg_text_length = results_df[results_df["status"] == "Success"]["text_length"].mean()
+            avg_tables = results_df[results_df["status"] == "Success"]["table_count"].mean()
+            processor.logger.info(f"平均每文件: {avg_text_length:.0f} 字符, {avg_tables:.1f} 个表格")
+
+    # 处理过程中发生错误, 记录日志
+    except Exception as e:
+        processor.logger.error(f"处理过程发生错误: {str(e)}")
+
+
+if __name__ == "__main__":
+    main()

test.py

@@ -0,0 +1,21 @@
+import requests
+import time
+import json
+
+url = "http://0.0.0.0:8103/"
+data = {"question": "平日里蜂蜜加白醋一起喝有什么疗效？"}
+#data = {"question": "听说用酸枣仁泡水喝能养生,是真的吗？"}
+
+start_time = time.time()
+
+data = json.dumps(data)
+
+# 向服务发送请求
+res = requests.post(url, data=data)
+
+cost_time = time.time() - start_time
+
+print('单次查询的耗时:', cost_time, 's')
+
+res = json.loads(res.text)
+print(res)

vector.py

@@ -0,0 +1,262 @@
+import os
+from pydantic import BaseModel
+from tqdm import tqdm
+import json
+import uuid
+import time
+import pandas as pd
+from openai import OpenAI
+from langchain.embeddings.base import Embeddings
+from langchain_core.documents import Document
+from langchain_milvus import Milvus, BM25BuiltInFunction
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from langchain_classic.retrievers.parent_document_retriever import ParentDocumentRetriever
+from langchain_core.stores import InMemoryStore
+from dotenv import load_dotenv
+
+# 加载 .env 文件中的环境变量, 隐藏 API Keys
+load_dotenv()
+
+
+# ============================================================
+# 嵌入模型, 采用 OpenAI text-embedding-3-small
+# ============================================================
+
+class OpenAIEmbeddings(Embeddings):
+    """基于 OpenAI Embedding API 的自定义嵌入类"""
+
+    def __init__(self):
+        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+
+    def embed_documents(self, texts):
+        embeddings = []
+        for text in texts:
+            response = self.client.embeddings.create(
+                model="text-embedding-3-small",
+                input=[text],
+            )
+            embeddings.append(response.data[0].embedding)
+        return embeddings
+
+    def embed_query(self, text):
+        # 查询文档
+        return self.embed_documents([text])[0]
+
+
+# ============================================================
+# Milvus 向量数据库封装类 (第一路召回: JSONL 文本数据)
+# ============================================================
+
+class Milvus_vector():
+    def __init__(self, uri="./milvus_agent.db"):
+        self.URI = uri
+        self.embeddings = OpenAIEmbeddings()
+
+        # 定义索引类型
+        self.dense_index = {
+            "metric_type": "IP",
+            "index_type": "IVF_FLAT",
+        }
+        self.sparse_index = {
+            "metric_type": "BM25",
+            "index_type": "SPARSE_INVERTED_INDEX"
+        }
+
+    def create_vector_store(self, docs):
+        init_docs = docs[:10]
+        self.vectorstore = Milvus.from_documents(
+            documents=init_docs,
+            embedding=self.embeddings,
+            builtin_function=BM25BuiltInFunction(),  # output_field_names="sparse",
+            index_params=[self.dense_index, self.sparse_index],
+            vector_field=["dense", "sparse"],
+            connection_args={
+                "uri": self.URI,
+            },
+            # 支持 ("Strong", "Session", "Bounded", "Eventually")
+            consistency_level="Bounded",
+            drop_old=False,
+        )
+        print("已初始化创建 Milvus ‼")
+
+        count = 10
+        temp = []
+        for doc in tqdm(docs[10:]):
+            temp.append(doc)
+            if len(temp) >= 5:
+                self.vectorstore.aadd_documents(temp)
+                count += len(temp)
+                temp = []
+                print(f"已插入 {count} 条数据......")
+                time.sleep(1)
+
+        print(f"总共插入 {count} 条数据......")
+        print("已创建 Milvus 索引完成 ‼")
+
+        return self.vectorstore
+
+
+# ============================================================
+# PDF 父子文档检索器 (第二路召回: PDF 文档数据)
+# ============================================================
+
+class Pdf_retriever():
+    def __init__(self, uri="./pdf_agent.db"):
+        self.URI = uri
+        self.embeddings = OpenAIEmbeddings()
+
+        # 定义索引类型
+        self.dense_index = {
+            "metric_type": "IP",
+            "index_type": "IVF_FLAT",
+        }
+        self.sparse_index = {
+            "metric_type": "BM25",
+            "index_type": "SPARSE_INVERTED_INDEX"
+        }
+
+        self.docstore = InMemoryStore()
+
+        # 文本分割器
+        self.child_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=200,
+            chunk_overlap=50,
+            length_function=len,
+            separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
+        )
+        self.parent_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000,
+            chunk_overlap=200
+        )
+
+    def create_pdf_vector_store(self, docs):
+        self.milvus_vectorstore = Milvus(
+            embedding_function=self.embeddings,
+            builtin_function=BM25BuiltInFunction(),
+            vector_field=["dense", "sparse"],
+            index_params=[
+                {
+                    "metric_type": "IP",
+                    "index_type": "IVF_FLAT",
+                },
+                {
+                    "metric_type": "BM25",
+                    "index_type": "SPARSE_INVERTED_INDEX"
+                }
+            ],
+            connection_args={"uri": self.URI},
+            consistency_level="Bounded",
+            drop_old=False,
+        )
+
+        # 设置父子文档检索器
+        self.retriever = ParentDocumentRetriever(
+            vectorstore=self.milvus_vectorstore,
+            docstore=self.docstore,
+            child_splitter=self.child_splitter,
+            parent_splitter=self.parent_splitter,
+        )
+
+        # 添加文档
+        count = 0
+        temp = []
+        for doc in tqdm(docs):
+            temp.append(doc)
+            if len(temp) >= 10:
+                # ParentDocumentRetriever()不支持异步等待操作
+                self.retriever.add_documents(temp)
+                count += len(temp)
+                temp = []
+                print(f"已插入 {count} 条数据......")
+                time.sleep(1)
+
+        print(f"总共插入 {count} 条数据......")
+        print("基于PDF文档数据的 Milvus 索引完成 ‼")
+
+        return self.retriever
+
+
+# ============================================================
+# 数据预处理: 从 JSONL 文件加载文档 (第一路)
+# ============================================================
+
+def prepare_document(file_path=['./data/dialog.jsonl', './data/train.jsonl']):
+    # 逐条取出文本数据, 创建嵌入张量, 然后将张量数据插入Milvus
+    file_path1 = file_path[0]
+
+    count = 0
+    docs = []
+
+    with open(file_path1, 'r', encoding='utf-8') as f:
+        for line in f:
+            content = json.loads(line.strip())
+            prompt = content['query'] + "\n" + content['response']
+
+            temp_doc = Document(page_content=prompt, metadata={"doc_id": str(uuid.uuid4())})
+            docs.append(temp_doc)
+
+            count += 1
+
+    print(f"已加载 {count} 条数据!")
+
+    return docs
+
+
+# ============================================================
+# 数据预处理: 从 PDF 提取结果加载文档 (第二路)
+# ============================================================
+
+def prepare_pdf_document(file_path="./pdf_output/pdf_detailed_text.xlsx"):
+    df = pd.read_excel(file_path)
+
+    # 空行直接删除, 否则后续处理报错
+    df = df.dropna(subset=['text_content'])
+
+    # 将DataFrame转换为LangChain文档
+    documents = []
+    for _, row in df.iterrows():
+        # 确保 text_content 是字符串, 且不为 NaN
+        text_content = str(row['text_content']) if pd.notna(row['text_content']) else ""
+
+        doc = Document(
+            page_content=text_content.strip(),
+            metadata={"doc_id": str(uuid.uuid4())}
+        )
+        documents.append(doc)
+
+    print(f"成功加载 {len(documents)} 个文档")
+
+    return documents
+
+
+# ============================================================
+# 主入口: 执行数据入库流程
+# ============================================================
+
+if __name__ == "__main__":
+    '''
+    # 预处理即将插入 Milvus 的文档数据
+    docs = prepare_document()
+    print("预处理文档数据成功......")
+
+    # 创建 Milvus 连接
+    milvus_vectorstore = Milvus_vector()
+    print("创建Milvus连接成功......")
+
+    # 创建向量索引
+    vectorstore = milvus_vectorstore.create_vector_store(docs)
+    print("全部初始化完成, 可以开始问答了......")
+    '''
+
+    # 将 PDF 后处理文档中的数据, 封装成Document
+    docs = prepare_pdf_document()
+    print("预处理 PDF 文档数据成功......")
+    # print(docs[0])
+
+    pdf_vectorstore = Pdf_retriever()
+    print("创建 PDF Milvus 连接成功......")
+
+    retriever = pdf_vectorstore.create_pdf_vector_store(docs)
+    print("创建基于 Milvus 数据库的父子文档检索器成功......")
+    print(retriever)
+    print("全部初始化完成, 可以开始问答了......")