Add GraphDatabase module with Neo4j + Redis caching

.gitignore

@@ -1 +1,8 @@
+__pycache__/
+*.pyc
 .env
+*.db
+*.db.lock
+.DS_Store
+pdf_output/
+data/

GraphDatabase/main.py

@@ -0,0 +1,193 @@
+import os
+from fastapi import FastAPI, HTTPException, Depends
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+from contextlib import asynccontextmanager
+from openai import OpenAI
+from dotenv import load_dotenv
+from models import NL2CypherRequest, CypherResponse, ValidationRequest, ValidationResponse
+from schemas import EXAMPLE_SCHEMA
+from prompts import create_system_prompt, create_validation_prompt
+from validators import CypherValidator, RuleBasedValidator
+
+# 加载环境变量
+load_dotenv()
+
+# 获取 OpenAI 的 api key
+openai_api_key = os.getenv("OPENAI_API_KEY")
+
+
+# 生命周期管理
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # 启动时初始化
+    neo4j_uri = os.getenv("NEO4J_URI")
+    neo4j_user = os.getenv("NEO4J_USER")
+    neo4j_password = os.getenv("NEO4J_PASSWORD")
+
+    if all([neo4j_uri, neo4j_user, neo4j_password]):
+        app.state.validator = CypherValidator(neo4j_uri, neo4j_user, neo4j_password)
+    else:
+        app.state.validator = RuleBasedValidator()
+
+    yield
+
+    # 关闭时清理
+    if hasattr(app.state.validator, 'close'):
+        app.state.validator.close()
+
+
+# 创建FastAPI应用
+app = FastAPI(title="NL2Cypher API", lifespan=lifespan)
+
+# 初始化 OpenAI 模型
+client = OpenAI(
+    api_key=openai_api_key,                       # 你的 OpenAI API 密钥
+    base_url="https://api.openai.com/v1",         # OpenAI 的 API 端点
+)
+
+# 添加CORS中间件
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+def clean_cypher_output(raw_output: str) -> str:
+    """清洗 LLM 返回的 Cypher 查询, 去掉多余的包装文本"""
+    import re
+    text = raw_output.strip()
+
+    # 去掉 markdown 代码块: ```cypher ... ``` 或 ``` ... ```
+    text = re.sub(r'```(?:cypher)?\s*', '', text)
+    text = text.strip('`')
+
+    # 去掉 Cypher: "..." 包装
+    match = re.match(r'^[Cc]ypher:\s*["\']?(.*?)["\']?\s*$', text, re.DOTALL)
+    if match:
+        text = match.group(1).strip()
+
+    # 去掉首尾引号
+    if (text.startswith('"') and text.endswith('"')) or \
+       (text.startswith("'") and text.endswith("'")):
+        text = text[1:-1].strip()
+
+    return text
+
+
+def generate_cypher_query(natural_language: str, query_type: str = None) -> str:
+    """使用 OpenAI 生成 Cypher 查询"""
+    system_prompt = create_system_prompt(str(EXAMPLE_SCHEMA.model_dump()))
+
+    user_prompt = natural_language
+    if query_type:
+        user_prompt = f"{query_type}查询: {natural_language}"
+
+    try:
+        response = client.chat.completions.create(
+            model="gpt-4o",
+            messages=[
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt}
+            ],
+            temperature=0.1,
+            max_tokens=2048,
+            stream=False
+        )
+        raw_output = response.choices[0].message.content.strip()
+        return clean_cypher_output(raw_output)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"OpenAI API错误: {str(e)}")
+
+
+def explain_cypher_query(cypher_query: str) -> str:
+    """解释Cypher查询"""
+    try:
+        response = client.chat.completions.create(
+            model="gpt-4o",
+            messages=[
+                {"role": "system", "content": "你是一个Neo4j专家, 请用简单明了的语言解释Cypher查询."},
+                {"role": "user", "content": f"请解释以下Cypher查询: {cypher_query}"}
+            ],
+            temperature=0.1,
+            max_tokens=1024,
+            stream=False
+        )
+        return response.choices[0].message.content.strip()
+    except Exception as e:
+        return f"无法生成解释: {str(e)}"
+
+
+@app.post("/generate", response_model=CypherResponse)
+async def generate_cypher(request: NL2CypherRequest):
+    """生成Cypher查询端点"""
+    # 利用 OpenAI 生成 Cypher 查询
+    cypher_query = generate_cypher_query(
+        request.natural_language_query,
+        request.query_type.value if request.query_type else None
+    )
+
+    # 利用 OpenAI 生成解释
+    explanation = explain_cypher_query(cypher_query)
+
+    # 验证查询
+    is_valid, errors = app.state.validator.validate_against_schema(cypher_query, EXAMPLE_SCHEMA)
+
+    # 计算置信度, 将基础置信度设置为0.9
+    confidence = 0.9
+
+    # 如果有潜在错误, 重新计算置信度 confidence
+    if errors:
+        confidence = max(0.3, confidence - len(errors) * 0.1)
+
+    return CypherResponse(
+        cypher_query=cypher_query,
+        explanation=explanation,
+        confidence=confidence,
+        validated=is_valid,
+        validation_errors=errors
+    )
+
+
+@app.post("/validate", response_model=ValidationResponse)
+async def validate_cypher(request: ValidationRequest):
+    """验证Cypher查询端点"""
+    is_valid, errors = app.state.validator.validate_against_schema(request.cypher_query, EXAMPLE_SCHEMA)
+
+    # 生成改进建议
+    suggestions = []
+    if errors:
+        try:
+            response = client.chat.completions.create(
+                model="gpt-4o",
+                messages=[
+                    {"role": "system", "content": "你是一个Neo4j专家, 请提供Cypher查询的改进建议."},
+                    {"role": "user", "content": create_validation_prompt(request.cypher_query)}
+                ],
+                temperature=0.1,
+                max_tokens=1024,
+                stream=False
+            )
+            suggestions = [response.choices[0].message.content.strip()]
+        except:
+            suggestions = ["无法生成建议"]
+
+    return ValidationResponse(
+        is_valid=is_valid,
+        errors=errors,
+        suggestions=suggestions
+    )
+
+
+@app.get("/schema")
+async def get_schema():
+    """获取图模式端点"""
+    return EXAMPLE_SCHEMA.model_dump()
+
+
+if __name__ == "__main__":
+    # 因为项目中的主服务Agent启动在8103端口, 所以这个neo4j的服务端口另选一个8101即可
+    uvicorn.run(app, host="0.0.0.0", port=8101)

GraphDatabase/models.py

@@ -0,0 +1,77 @@
+import os
+from pydantic import BaseModel, Field
+from typing import Optional, List, Dict, Any
+from enum import Enum
+
+
+class QueryType(str, Enum):
+    MATCH = "MATCH"
+    CREATE = "CREATE"
+    MERGE = "MERGE"
+    DELETE = "DELETE"
+    SET = "SET"
+    REMOVE = "REMOVE"
+
+
+class NL2CypherRequest(BaseModel):
+    natural_language_query: str = Field(
+        description="自然语言描述的需求",
+        examples=["查找'心血管和血栓栓塞综合征'建议服用什么药物?"]
+    )
+    query_type: Optional[QueryType] = Field(
+        default=None,
+        description="指定查询类型,如果不指定则由模型推断"
+    )
+    limit: Optional[int] = Field(
+        default=10,
+        description="结果限制数量",
+        ge=1,
+        le=1000
+    )
+
+
+class CypherResponse(BaseModel):
+    cypher_query: str = Field(
+        ...,
+        description="生成的Cypher查询语句"
+    )
+    explanation: str = Field(
+        ...,
+        description="对生成的Cypher查询的解释"
+    )
+    confidence: float = Field(
+        ...,
+        description="模型对生成查询的信心度(0-1)",
+        ge=0,
+        le=1
+    )
+    validated: bool = Field(
+        default=False,
+        description="查询是否通过验证"
+    )
+    validation_errors: List[str] = Field(
+        default_factory=list,
+        description="验证过程中发现的错误"
+    )
+
+
+class ValidationRequest(BaseModel):
+    cypher_query: str = Field(
+        ...,
+        description="需要验证的Cypher查询"
+    )
+
+
+class ValidationResponse(BaseModel):
+    is_valid: bool = Field(
+        ...,
+        description="查询是否有效"
+    )
+    errors: List[str] = Field(
+        default_factory=list,
+        description="发现的错误列表"
+    )
+    suggestions: List[str] = Field(
+        default_factory=list,
+        description="改进建议"
+    )

GraphDatabase/prompts.py

@@ -0,0 +1,65 @@
+import os
+from schemas import EXAMPLE_SCHEMA
+
+
+def create_system_prompt(schema: str) -> str:
+    return f"""
+    你是一个专业的Neo4j Cypher查询生成器, 你的任务是将自然语言描述转换为准确, 高效的Cypher查询.
+
+    # 图数据库模式
+    {schema}
+
+    # 重要规则
+    1. 始终使用参数化查询风格, 对字符串值使用单引号
+    2. 确保节点标签和关系类型使用正确的大小写
+    3. 对于模糊查询, 使用 CONTAINS 或 STARTS WITH 而不是 "="
+    4. 对于可选模式, 使用 OPTIONAL MATCH
+    5. 始终考虑查询性能, 使用适当的索引和约束
+    6. 对于需要返回多个实体的查询, 使用 RETURN 子句明确指定要返回的内容
+    7. 避免使用可能导致性能问题的查询模式
+
+    # 示例如下
+    自然语言: "查找心血管和血栓栓塞综合征建议服用什么药物?"
+    Cypher: "match (p:Disease)-[r:recommand_drug]-(d:Drug) where p.name='心血管和血栓栓塞综合征' return d.name"
+
+    自然语言: "查找嗜铬细胞瘤这种疾病有哪些临床症状?"
+    Cypher: "match (p:Disease)-[r:has_symptom]-(s:Symptom) where p.name='嗜铬细胞瘤' return s.name"
+
+    自然语言: "查找小儿先天性巨结肠推荐哪些饮食有利康复?"
+    Cypher: "match (p:Disease)-[r:recommand_eat]-(f:Food) where p.name='小儿先天性巨结肠' return f.name"
+
+    自然语言: "查找糖尿病需要做哪些检查项目?"
+    Cypher: "match (p:Disease)-[r:need_check]-(c:Check) where p.name='糖尿病' return c.name"
+
+    自然语言: "查找高血压属于哪个科室?"
+    Cypher: "match (p:Disease)-[r:belongs_to]-(d:Department) where p.name='高血压' return d.name"
+
+    自然语言: "查找感冒的常用药物有哪些?"
+    Cypher: "match (p:Disease)-[r:common_drug]-(d:Drug) where p.name='感冒' return d.name"
+
+    自然语言: "查找肺炎患者不能吃什么食物?"
+    Cypher: "match (p:Disease)-[r:no_eat]-(f:Food) where p.name='肺炎' return f.name"
+
+    自然语言: "查找胃炎患者适合吃什么食物?"
+    Cypher: "match (p:Disease)-[r:do_eat]-(f:Food) where p.name='胃炎' return f.name"
+
+    自然语言: "查找冠心病容易并发哪些疾病?"
+    Cypher: "match (p:Disease)-[r:acompany_with]-(d:Disease) where p.name='冠心病' return d.name"
+
+    自然语言: "查找阿莫西林是哪个厂家生产的?"
+    Cypher: "match (p:Producer)-[r:drugs_of]-(d:Drug) where d.name='阿莫西林' return p.name"
+
+    现在请根据以下自然语言描述生成Cypher查询:
+    """
+
+
+def create_validation_prompt(cypher_query: str) -> str:
+    return f"""
+    请分析以下Cypher查询, 指出其中的任何错误或潜在问题, 并提供改进建议:
+
+    {cypher_query}
+
+    请按以下格式回答:
+    错误: [列出所有错误]
+    建议: [提供改进建议]
+    """

GraphDatabase/schemas.py

@@ -0,0 +1,114 @@
+import os
+from pydantic import BaseModel
+from typing import Dict, List, Optional
+
+
+class NodeSchema(BaseModel):
+    label: str
+    properties: Dict[str, str]  # 属性名: 类型
+
+
+class RelationshipSchema(BaseModel):
+    type: str
+    from_node: str              # 起始节点标签
+    to_node: str                # 目标节点标签
+    properties: Dict[str, str]  # 属性名: 类型
+
+
+class GraphSchema(BaseModel):
+    nodes: List[NodeSchema]
+    relationships: List[RelationshipSchema]
+
+
+# 示例图模式 (按照neo4j数据库中的定义schema来填充)
+EXAMPLE_SCHEMA = GraphSchema(
+    # 节点的名称一定要严格保持跟neo4j一致
+    nodes=[
+        # --- PDF 原有的 4 个节点 ---
+        NodeSchema(label="Disease", properties={"name": "string", "desc": "string", "cause": "string", "prevent": "string", "cure_lasttime": "string", "cure_department": "string", "cure_way": "string", "cure_prob": "string", "easy_get": "string"}),
+        NodeSchema(label="Drug", properties={"name": "string"}),
+        NodeSchema(label="Food", properties={"name": "string"}),
+        NodeSchema(label="Symptom", properties={"name": "string"}),
+        # --- 基于 Neo4j 截图新增的 3 个节点 ---
+        NodeSchema(label="Check", properties={"name": "string"}),
+        NodeSchema(label="Department", properties={"name": "string"}),
+        NodeSchema(label="Producer", properties={"name": "string"}),
+    ],
+    # 关系的相关字段一定要严格保持跟neo4j一致, 大小写都不能错
+    relationships=[
+        # --- PDF 原有的 3 个关系 ---
+        RelationshipSchema(
+            type="has_symptom",
+            from_node="Disease",
+            to_node="Symptom",
+            properties={}
+        ),
+        RelationshipSchema(
+            type="recommand_drug",
+            from_node="Disease",
+            to_node="Drug",
+            properties={}
+        ),
+        RelationshipSchema(
+            type="recommand_eat",
+            from_node="Disease",
+            to_node="Food",
+            properties={}
+        ),
+        # --- 基于 Neo4j 截图新增的关系 ---
+        # Disease 需要做的检查项目
+        RelationshipSchema(
+            type="need_check",
+            from_node="Disease",
+            to_node="Check",
+            properties={}
+        ),
+        # Disease 所属的科室
+        RelationshipSchema(
+            type="belongs_to",
+            from_node="Disease",
+            to_node="Department",
+            properties={}
+        ),
+        # Disease 的常用药物
+        RelationshipSchema(
+            type="common_drug",
+            from_node="Disease",
+            to_node="Drug",
+            properties={}
+        ),
+        # Disease 宜吃的食物
+        RelationshipSchema(
+            type="do_eat",
+            from_node="Disease",
+            to_node="Food",
+            properties={}
+        ),
+        # Disease 忌吃的食物
+        RelationshipSchema(
+            type="no_eat",
+            from_node="Disease",
+            to_node="Food",
+            properties={}
+        ),
+        # Disease 的并发症
+        RelationshipSchema(
+            type="acompany_with",
+            from_node="Disease",
+            to_node="Disease",
+            properties={}
+        ),
+        # Drug 的生产商
+        RelationshipSchema(
+            type="drugs_of",
+            from_node="Producer",
+            to_node="Drug",
+            properties={}
+        ),
+    ]
+)
+
+
+if __name__ == '__main__':
+    res = str(EXAMPLE_SCHEMA.model_dump())
+    print(res)

GraphDatabase/validators.py

@@ -0,0 +1,120 @@
+import re
+from typing import List, Tuple
+from neo4j import GraphDatabase
+import os
+
+
+class CypherValidator:
+    def __init__(self, neo4j_uri: str, neo4j_user: str, neo4j_password: str):
+        self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))
+
+    def validate_syntax(self, cypher_query: str) -> Tuple[bool, List[str]]:
+        """验证Cypher查询的语法"""
+        errors = []
+
+        # 基本语法检查
+        if not cypher_query.strip().upper().startswith(('MATCH', 'CREATE', 'MERGE', 'CALL')):
+            errors.append("查询必须以MATCH, CREATE, MERGE 或 CALL开头!!!")
+
+        # 检查是否有潜在的注入风险
+        if any(keyword in cypher_query.upper() for keyword in ['DROP', 'DELETE', 'DETACH', 'REMOVE']):
+            if not any(keyword in cypher_query.upper() for keyword in ['DELETE', 'DETACH']):
+                errors.append("查询包含可能危险的操作符")
+
+        # 检查RETURN语句是否存在 (对于MATCH查询)
+        if cypher_query.upper().startswith('MATCH') and 'RETURN' not in cypher_query.upper():
+            errors.append("MATCH查询必须包含RETURN语句!!!")
+
+        # 使用Neo4j解释计划验证查询
+        try:
+            with self.driver.session() as session:
+                result = session.run(f"EXPLAIN {cypher_query}")
+                # 如果解释成功, 语法基本正确
+                return True, errors
+        except Exception as e:
+            errors.append(f"语法错误: {str(e)}")
+            return False, errors
+
+    def validate_against_schema(self, cypher_query: str, schema) -> Tuple[bool, List[str]]:
+        """根据模式验证查询"""
+        errors = []
+
+        # 提取所有节点标签
+        node_labels = [node.label for node in schema.nodes]
+        node_pattern = r'\(([a-zA-Z0-9_]+)?:?([a-zA-Z0-9_]+)\)'
+        matches = re.findall(node_pattern, cypher_query)
+
+        for match in matches:
+            if match[1] and match[1] not in node_labels:
+                errors.append(f"使用了不存在的节点标签: {match[1]}")
+
+        # 提取所有关系类型
+        rel_types = [rel.type for rel in schema.relationships]
+        rel_pattern = r'\[([a-zA-Z0-9_]+)?:?([a-zA-Z0-9_]+)\]'
+        rel_matches = re.findall(rel_pattern, cypher_query)
+
+        for match in rel_matches:
+            if match[1] and match[1] not in rel_types:
+                errors.append(f"使用了不存在的关系类型: {match[1]}")
+
+        return len(errors) == 0, errors
+
+    def close(self):
+        self.driver.close()
+
+
+
+# 简单的基于规则的验证器 (当无法连接Neo4j时使用)
+class RuleBasedValidator:
+    def validate(self, cypher_query: str, schema) -> Tuple[bool, List[str]]:
+        errors = []
+
+        # 检查基本结构
+        if not cypher_query.strip():
+            errors.append("查询不能为空!!!")
+            return False, errors
+
+        # 检查是否包含潜在危险操作
+        dangerous_patterns = [
+            (r'(?i)drop\s+', "DROP操作可能危险"),
+            (r'(?i)delete\s+', "DELETE操作需要谨慎"),
+            (r'(?i)detach\s+delete', "DETACH DELETE操作非常危险!!"),
+            (r'(?i)remove\s+', "REMOVE操作需要谨慎"),
+        ]
+
+        for pattern, message in dangerous_patterns:
+            if re.search(pattern, cypher_query):
+                errors.append(message)
+
+        # 检查MATCH查询是否包含RETURN
+        if re.match(r'(?i)match', cypher_query) and not re.search(r'(?i)return', cypher_query):
+            errors.append("MATCH查询必须包含RETURN子句")
+
+        # 检查CREATE查询是否合理
+        if re.match(r'(?i)create', cypher_query) and not re.search(r'(?i)(node|relationship|label|index)', cypher_query):
+            errors.append("CREATE查询应该明确创建节点或关系")
+
+        return len(errors) == 0, errors
+
+    def validate_against_schema(self, cypher_query: str, schema) -> Tuple[bool, List[str]]:
+        """兼容CypherValidator的接口, 先做规则验证再做schema验证"""
+        is_valid, errors = self.validate(cypher_query, schema)
+
+        # 额外进行schema验证
+        node_labels = [node.label for node in schema.nodes]
+        node_pattern = r'\(([a-zA-Z0-9_]+)?:?([a-zA-Z0-9_]+)\)'
+        matches = re.findall(node_pattern, cypher_query)
+
+        for match in matches:
+            if match[1] and match[1] not in node_labels:
+                errors.append(f"使用了不存在的节点标签: {match[1]}")
+
+        rel_types = [rel.type for rel in schema.relationships]
+        rel_pattern = r'\[([a-zA-Z0-9_]+)?:?([a-zA-Z0-9_]+)\]'
+        rel_matches = re.findall(rel_pattern, cypher_query)
+
+        for match in rel_matches:
+            if match[1] and match[1] not in rel_types:
+                errors.append(f"使用了不存在的关系类型: {match[1]}")
+
+        return len(errors) == 0, errors

agent3.py

@@ -0,0 +1,298 @@
+import os
+import uvicorn
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+import json
+import requests
+import datetime
+from openai import OpenAI
+from neo4j import GraphDatabase
+from langchain_milvus import Milvus, BM25BuiltInFunction
+from vector import OpenAIEmbeddings, get_redis_client, cache_set, cache_get
+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 连接成功......")
+
+# 获取 neo4j 图数据库的连接
+neo4j_uri = os.getenv("NEO4J_URI", "bolt://localhost:7687")
+neo4j_user = os.getenv("NEO4J_USER", "neo4j")
+neo4j_password = os.getenv("NEO4J_PASSWORD", "neo4j")
+driver = GraphDatabase.driver(uri=neo4j_uri, auth=(neo4j_user, neo4j_password), max_connection_lifetime=1000)
+print("创建 Neo4j 连接成功......")
+
+# 创建大语言模型, 采用 OpenAI
+client_llm = create_openai_client()
+print("创建 OpenAI LLM 成功......")
+
+# 获取 Redis 连接
+client_redis = get_redis_client()
+print("创建 Redis 连接成功......")
+
+
+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: 先查 Redis 缓存, 如果缓存命中, 直接返回结果
+    # ============================================================
+    response_redis = cache_get(client_redis, query)
+
+    if response_redis is not None:
+        # redis 返回的字符串是以十六进制显示的, 需要按 utf-8 解码
+        response = response_redis.decode('utf-8')
+
+        now = datetime.datetime.now()
+        time = now.strftime("%Y-%m-%d %H:%M:%S")
+        answer = {
+            "response": response,
+            "status": 200,
+            "time": time
+        }
+        print('REDIS HIT !!!')
+        return answer
+
+    # ============================================================
+    # 2: 向量数据库 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 = format_docs(recall_rerank_milvus)
+    else:
+        context = ""
+
+    # ============================================================
+    # 2.5: PDF 文档的 Milvus 召回 (父子文档检索器)
+    # ============================================================
+    pdf_res = ""
+    retrieved_docs = parent_retriever.invoke(query)
+
+    if retrieved_docs is not None and len(retrieved_docs) >= 1:
+        pdf_res = retrieved_docs[0].page_content
+        print("PDF res: ", pdf_res)
+
+    context = context + "\n" + pdf_res
+
+    # ============================================================
+    # 3: 图数据库 neo4j 精准召回
+    # ============================================================
+    # 访问 neo4j API 服务, 生成 Cypher 命令
+    neo4j_res = ""
+    data = {"natural_language_query": query}
+    data_json = json.dumps(data)
+
+    try:
+        cypher_response = requests.post("http://0.0.0.0:8101/generate", data_json)
+
+        if cypher_response.status_code == 200:
+            cypher_response_data = cypher_response.json()
+
+            cypher_query = cypher_response_data["cypher_query"]
+            confidence = cypher_response_data["confidence"]
+            is_valid = cypher_response_data["validated"]
+
+            if cypher_query is not None and float(confidence) >= 0.9 and is_valid == True:
+                print("neo4j Cypher 初步生成成功 !!!")
+
+                # 验证 neo4j 生成的 Cypher 命令完全正确
+                data = {"cypher_query": cypher_query}
+                data_json = json.dumps(data)
+                cypher_valid = requests.post("http://0.0.0.0:8101/validate", data_json)
+
+                if cypher_valid.status_code == 200:
+                    cypher_valid_data = cypher_valid.json()
+                    if cypher_valid_data["is_valid"] == True:
+                        with driver.session() as session:
+                            try:
+                                record = session.run(cypher_query)
+                                result = list(map(lambda x: x[0], record))
+                                neo4j_res = ','.join(result)
+                            except Exception as e:
+                                print(e)
+                                print("neo4j查询失败 !!")
+                                neo4j_res = ""
+        else:
+            print("生成Cypher查询失败 !!")
+    except Exception as e:
+        print(f"neo4j API 服务不可用: {e}")
+
+    # 合并 Milvus、PDF 和 neo4j 的召回结果, 共同作为 LLM 的输入 prompt
+    context = context + "\n" + neo4j_res
+
+    # ============================================================
+    # 4: 为LLM定义系统和用户提示
+    # ============================================================
+    SYSTEM_PROMPT = """
+        System: 你是一个非常得力的医学助手, 你可以通过从数据库中检索出的信息找到问题的答案.
+        """
+
+    USER_PROMPT = f"""
+        User: 利用介于<context>和</context>之间的从数据库中检索出的信息来回答问题, 具体的问题介于<question>和</question>之间. 如果提供的信息为空, 则按照你的经验知识来给出尽可能严谨准确的回答, 不知道的时候坦诚的承认不了解, 不要编造不真实的信息.
+        <context>
+        {context}
+        </context>
+
+        <question>
+        {query}
+        </question>
+        """
+
+    # ============================================================
+    # 5: 使用 OpenAI 最新版本模型, 根据提示生成回复
+    # ============================================================
+    response = generate_openai_answer(client_llm, SYSTEM_PROMPT + USER_PROMPT.format(context, query))
+
+    # ============================================================
+    # 6: 写入缓存
+    # ============================================================
+    cache_set(client_redis, query, response)
+
+    # ============================================================
+    # 7: 组装服务返回数据
+    # ============================================================
+    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)

milvus_agent.db

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

test.py

@@ -3,8 +3,9 @@ import time
 import json
 
 url = "http://0.0.0.0:8103/"
-data = {"question": "平日里蜂蜜加白醋一起喝有什么疗效？"}
-#data = {"question": "听说用酸枣仁泡水喝能养生,是真的吗？"}
+#data = {"question": "平日里蜂蜜加白醋一起喝有什么疗效？"}
+data = {"question": "听说用酸枣仁泡水喝能养生,是真的吗？"}
+#data = {"question": "糖尿病有什么症状？"}
 
 start_time = time.time()
 

vector.py

@@ -4,6 +4,7 @@ from tqdm import tqdm
 import json
 import uuid
 import time
+import redis
 import pandas as pd
 from openai import OpenAI
 from langchain.embeddings.base import Embeddings
@@ -18,6 +19,36 @@ from dotenv import load_dotenv
 load_dotenv()
 
 
+# ============================================================
+# Redis 缓存处理模块
+# ============================================================
+
+def get_redis_client():
+    # 创建Redis连接, 使用连接池 (推荐用于生产环境)
+    pool = redis.ConnectionPool(host='0.0.0.0', port=6379, db=0, password=None, max_connections=10)
+    r = redis.StrictRedis(connection_pool=pool)
+
+    # 测试连接
+    try:
+        r.ping()
+        print("成功连接到 Redis !")
+    except redis.ConnectionError:
+        print("无法连接到 Redis !")
+
+    return r
+
+
+# 将 (question, answer) 问答对, 存入 redis
+def cache_set(r, question: str, answer: str):
+    r.hset("qa", question, answer)
+    r.expire("qa", 3600)
+
+
+# 通过 question, 读取存在 redis 中的 answer
+def cache_get(r, question: str):
+    return r.hget("qa", question)
+
+
 # ============================================================
 # 嵌入模型, 采用 OpenAI text-embedding-3-small
 # ============================================================
@@ -247,7 +278,7 @@ if __name__ == "__main__":
     vectorstore = milvus_vectorstore.create_vector_store(docs)
     print("全部初始化完成, 可以开始问答了......")
     '''
-
+    ''''
     # 将 PDF 后处理文档中的数据, 封装成Document
     docs = prepare_pdf_document()
     print("预处理 PDF 文档数据成功......")
@@ -259,4 +290,9 @@ if __name__ == "__main__":
     retriever = pdf_vectorstore.create_pdf_vector_store(docs)
     print("创建基于 Milvus 数据库的父子文档检索器成功......")
     print(retriever)
+    '''
+    r = get_redis_client()
+    print("创建Redis连接成功......")
+    print(r)
+
     print("全部初始化完成, 可以开始问答了......")