向量数据库与图数据库协同实践：融合大模型搭建图数据知识检索与关联推理智能应用24.2

from sentence_transformers import SentenceTransformer
import numpy as np

# 加载开源嵌入模型
model = SentenceTransformer("all-MiniLM-L6-v2")

# 原始文档文本
docs = [
    "张三参与智慧城市建设项目",
    "智慧城市项目签订三号工程合同",
    "张三负责三号合同现场管理工作"
]
# 生成向量
embeddings = model.encode(docs)
print("向量维度：", embeddings.shape)
# 输出：向量维度： (3, 384)

# 创建实体与关系
CREATE (p:Person{name:"张三"})-[:JOIN]->(proj:Project{name:"智慧城市项目"})
-[:SIGN]->(contract:Contract{name:"三号工程合同"})

# 查询张三参与项目关联的合同
MATCH (p:Person{name:"张三"})-[:JOIN]->(proj:Project)-[:SIGN]->(c:Contract)
RETURN proj.name,c.name

pip install qdrant-client neo4j sentence-transformers

from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, VectorParams, Distance
from sentence_transformers import SentenceTransformer

# 初始化向量模型
embed_model = SentenceTransformer("all-MiniLM-L6-v2")
vector_dim = 384

# 连接Qdrant向量库
qdrant_client = QdrantClient(host="127.0.0.1", port=6333)
collection_name = "knowledge_docs"

# 创建集合
if not qdrant_client.collection_exists(collection_name):
    qdrant_client.create_collection(
        collection_name=collection_name,
        vectors_config=VectorParams(size=vector_dim, distance=Distance.COSINE)
    )

# 测试文档
texts = [
    "张三参与智慧城市建设项目",
    "智慧城市项目签订三号工程合同",
    "张三负责三号合同现场管理工作"
]
vectors = embed_model.encode(texts)

# 批量写入向量库
points = []
for idx, (text, vec) in enumerate(zip(texts, vectors)):
    points.append(PointStruct(id=idx, vector=vec, payload={"text": text}))

qdrant_client.upsert(collection_name=collection_name, points=points)

# 向量检索函数
def search_vector_db(query: str, topk=3):
    query_vec = embed_model.encode(query)
    res = qdrant_client.search(
        collection_name=collection_name,
        query_vector=query_vec,
        limit=topk
    )
    return [hit.payload["text"] for hit in res]

from neo4j import GraphDatabase

driver = GraphDatabase.driver("bolt://127.0.0.1:7687", auth=("neo4j", "password"))

def graph_query_person(person_name: str):
    cypher = """
    MATCH (p:Person{name:$name})-r->(target)
    RETURN p.name,type(r),target.name
    """
    with driver.session() as session:
        result = session.run(cypher, name=person_name)
        records = []
        for rec in result:
            records.append(f"{rec[0]} {rec[1]} {rec[2]}")
    return records

def hybrid_search(user_query: str, entity_name: str):
    # 1.向量检索文档
    doc_result = search_vector_db(user_query)
    # 2.图数据库查询关联关系
    graph_result = graph_query_person(entity_name)

    context = f"""
【相关文档素材】
{chr(10).join(doc_result)}

【实体关联关系】
{chr(10).join(graph_result)}
"""
    return context

# 调用测试
if __name__ == "__main__":
    ctx = hybrid_search("张三参与了哪些项目与合同", "张三")
    print(ctx)

import time
import uuid
import math
import numpy as np
from dataclasses import dataclass, field
from typing import List, Optional, Tuple

from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, VectorParams, Distance
from sentence_transformers import SentenceTransformer
from neo4j import GraphDatabase


# ============================================================
# 1. 记忆模块
# ============================================================

@dataclass
class MemoryItem:
    memory_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    content: str = ""
    embedding: Optional[List[float]] = None
    create_time: float = field(default_factory=time.time)
    last_access_time: float = field(default_factory=time.time)
    weight: float = 1.0
    tags: List[str] = field(default_factory=list)
    source: str = "chat"
    version: int = 1
    is_valid: bool = True


class BaseMemoryStore:
    def __init__(self):
        self.memory_pool: dict[str, MemoryItem] = {}

    def add_memory(self, item: MemoryItem):
        self.memory_pool[item.memory_id] = item

    def get_all(self) -> List[MemoryItem]:
        return list(self.memory_pool.values())


def cosine_similarity(vec1: List[float], vec2: List[float]) -> float:
    a = np.array(vec1)
    b = np.array(vec2)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))


def deduplicate_memory(new_mem: MemoryItem, store: BaseMemoryStore, threshold=0.85) -> bool:
    """返回 True 表示存在重复，不入库"""
    if not new_mem.embedding:
        return False
    for mem in store.get_all():
        if not mem.embedding or not mem.is_valid:
            continue
        if cosine_similarity(new_mem.embedding, mem.embedding) >= threshold:
            mem.last_access_time = time.time()
            return True
    return False


def compress_memory_content(raw_text: str, level: str = "light") -> str:
    if len(raw_text) < 120:
        return raw_text
    # 此处替换为真实 LLM 调用
    return f"【{level}压缩结果】{raw_text[:60]}..."


def update_memory(store: BaseMemoryStore, target_id: str, new_content: str):
    if target_id not in store.memory_pool:
        return None
    old_mem = store.memory_pool[target_id]
    old_mem.is_valid = False
    new_mem = MemoryItem(
        content=new_content,
        tags=old_mem.tags.copy(),
        source=old_mem.source,
        version=old_mem.version + 1
    )
    store.add_memory(new_mem)
    return new_mem


def decay_weight(mem: MemoryItem, decay_lambda=0.0001, min_weight=0.1):
    delta = time.time() - mem.last_access_time
    new_weight = mem.weight * math.exp(-decay_lambda * delta)
    mem.weight = max(new_weight, min_weight)
    mem.last_access_time = time.time()
    return mem.weight


def detect_memory_conflict(mem_list: List[MemoryItem]) -> List[Tuple[str, str]]:
    conflict_pairs = []
    for i in range(len(mem_list)):
        for j in range(i + 1, len(mem_list)):
            m1, m2 = mem_list[i], mem_list[j]
            # 此处调用 LLM 判断冲突
            # if llm.check_conflict(m1.content, m2.content) == "冲突":
            #     conflict_pairs.append((m1.memory_id, m2.memory_id))
    return conflict_pairs


# ============================================================
# 2. 向量检索模块（Qdrant）
# ============================================================

embed_model = SentenceTransformer("all-MiniLM-L6-v2")
vector_dim = 384

qdrant_client = QdrantClient(host="127.0.0.1", port=6333)
collection_name = "knowledge_docs"

if not qdrant_client.collection_exists(collection_name):
    qdrant_client.create_collection(
        collection_name=collection_name,
        vectors_config=VectorParams(size=vector_dim, distance=Distance.COSINE)
    )


def index_documents(texts: List[str]):
    vectors = embed_model.encode(texts)
    points = [
        PointStruct(id=idx, vector=vec.tolist(), payload={"text": text})
        for idx, (text, vec) in enumerate(zip(texts, vectors))
    ]
    qdrant_client.upsert(collection_name=collection_name, points=points)


def search_vector_db(query: str, topk=3):
    query_vec = embed_model.encode(query)
    res = qdrant_client.search(
        collection_name=collection_name,
        query_vector=query_vec.tolist(),
        limit=topk
    )
    return [hit.payload["text"] for hit in res]


# ============================================================
# 3. 图数据库模块（Neo4j）
# ============================================================

driver = GraphDatabase.driver("bolt://127.0.0.1:7687", auth=("neo4j", "password"))


def graph_query_person(person_name: str):
    cypher = """
    MATCH (p:Person{name:$name})-r->(target)
    RETURN p.name, type(r), target.name
    """
    with driver.session() as session:
        result = session.run(cypher, name=person_name)
        return [f"{rec[0]} {rec[1]} {rec[2]}" for rec in result]


# ============================================================
# 4. 混合检索 + 上下文组装
# ============================================================

def hybrid_search(user_query: str, entity_name: str):
    doc_result = search_vector_db(user_query)
    graph_result = graph_query_person(entity_name)
    context = f"""
【相关文档素材】
{chr(10).join(doc_result)}

【实体关联关系】
{chr(10).join(graph_result)}
"""
    return context


# ============================================================
# 5. 反馈闭环模块
# ============================================================

class FeedbackPipeline:
    def __init__(self):
        self.candidate_bad_cases = []
        self.labeled_samples = []

    def auto_filter_bad_case(self, log_item: dict) -> bool:
        if log_item.get("user_dislike"):
            return True
        if len(log_item.get("response", "")) < 10:
            return True
        return False

    def add_labeled_sample(self, sample):
        self.labeled_samples.append(sample)
        self.export_to_eval_set()

    def export_to_eval_set(self):
        # 此处替换为真实存储逻辑
        print(f"导出 {len(self.labeled_samples)} 条样本到评测集")


# ============================================================
# 6. 审批助手主流程
# ============================================================

class ApprovalAssistant:
    def __init__(self):
        self.memory_store = BaseMemoryStore()
        self.feedback = FeedbackPipeline()

    def chat(self, user_query: str, entity_name: str = "张三"):
        # 1. 混合检索，组装上下文
        context = hybrid_search(user_query, entity_name)

        # 2. 调用 LLM 生成回答（此处模拟）
        response = f"根据检索结果，{user_query} 的相关信息如下：\n{context}"

        # 3. 记忆写入（去重 + 压缩）
        new_mem = MemoryItem(content=response, source="approval_chat")
        if not deduplicate_memory(new_mem, self.memory_store):
            compressed = compress_memory_content(response)
            new_mem.content = compressed
            self.memory_store.add_memory(new_mem)

        # 4. 记忆衰减（对所有记忆执行）
        for mem in self.memory_store.get_all():
            decay_weight(mem)

        return response

    def handle_feedback(self, log_item: dict):
        if self.feedback.auto_filter_bad_case(log_item):
            self.feedback.candidate_bad_cases.append(log_item)
            print("⚠️ 已标记为疑似 bad case，待人工标注")


# ============================================================
# 7. 测试入口
# ============================================================

if __name__ == "__main__":
    # 初始化文档索引
    index_documents([
        "张三参与智慧城市建设项目",
        "智慧城市项目签订三号工程合同",
        "张三负责三号合同现场管理工作",
    ])

    # 启动助手
    assistant = ApprovalAssistant()

    # 用户提问
    answer = assistant.chat("张三参与了哪些项目与合同", entity_name="张三")
    print(answer)

    # 模拟用户负反馈
    assistant.handle_feedback({
        "query": "张三参与了哪些项目与合同",
        "response": answer,
        "user_dislike": True
    })