开发者视角：玄晶引擎双知识库架构如何破解企业AI“伪智能”困局？

xuanjing-engine/
├── app/                  # 应用层：业务场景代码
│   ├── consultant.py     # AI咨询师模块
│   └── customer_service.py # 获客智能体模块
├── engine/               # 引擎层：双引擎核心
│   ├── crystalink.py     # 任务调度逻辑
│   └── lapiscore.py      # API封装与工作流
├── model/                # 模型层：多模型调用
│   ├── model_map.py      # 任务-模型匹配配置
│   └── client.py         # 模型调用客户端
└── knowledge/            # 知识层：双库操作
    ├── rag_db.py         # RAG库CRUD
    └── vector_db.py      # 向量库操作

from py2neo import Graph, Node, Relationship, Subgraph
import pandas as pd

# 1. 连接Neo4j（本地或阿里云图数据库）
# 生产环境建议用环境变量存储密码，这里为演示简化
graph = Graph(
    "bolt://localhost:7687",  # 阿里云图数据库地址为：bolt://xxx.aliyuncs.com:7687
    auth=("neo4j", "xuanjing@2025")
)

# 2. 导入结构化数据（社保政策示例）
policy_data = pd.read_excel("shanghai_social_security.xlsx")
# 数据格式：行业,业务类型,服务场景,知识类型,内容,更新时间
for _, row in policy_data.iterrows():
    # 创建标签节点
    industry = Node("Industry", name=row["行业"])
    business = Node("Business", name=row["业务类型"])
    scene = Node("Scene", name=row["服务场景"])
    knowledge = Node("Knowledge", 
                    name=row["知识类型"], 
                    content=row["内容"],
                    update_time=row["更新时间"])
    
    # 建立关联关系
    rel1 = Relationship(industry, "包含", business)
    rel2 = Relationship(business, "包含", scene)
    rel3 = Relationship(scene, "包含", knowledge)
    
    # 批量写入（提升效率）
    graph.create(Subgraph([industry, business, scene, knowledge], [rel1, rel2, rel3]))

# 3. 调试：验证知识关联是否正确
# 执行Cypher查询，查看制造业-招聘外包场景的知识
query = """
MATCH (i:Industry)-[:包含]->(b:Business)-[:包含]->(s:Scene)-[:包含]->(k:Knowledge)
WHERE i.name = '制造业' AND b.name = '招聘外包'
RETURN s.name, k.name LIMIT 5
"""
result = graph.run(query).data()
print(pd.DataFrame(result))  # 输出关联结果，验证标签树是否正确

def rag_retrieval(industry, business, query):
    """
    RAG库检索接口：带业务标签过滤
    :param industry: 行业
    :param business: 业务类型
    :param query: 用户查询关键词
    :return: 结构化知识结果
    """
    # 1. 先按业务标签过滤场景
    scene_query = f"""
    MATCH (i:Industry)-[:包含]->(b:Business)-[:包含]->(s:Scene)
    WHERE i.name = '{industry}' AND b.name = '{business}'
    RETURN s.name AS scene
    """
    scenes = [item["scene"] for item in graph.run(scene_query).data()]
    
    # 2. 在场景内匹配关键词
    knowledge_list = []
    for scene in scenes:
        knowledge_query = f"""
        MATCH (s:Scene)-[:包含]->(k:Knowledge)
        WHERE s.name = '{scene}' AND k.content CONTAINS '{query}'
        RETURN k.content AS content, k.update_time AS update_time
        ORDER BY k.update_time DESC  # 优先返回最新知识
        """
        knowledge = graph.run(knowledge_query).data()
        knowledge_list.extend(knowledge)
    
    # 3. 去重并格式化结果
    result = []
    seen = set()
    for item in knowledge_list:
        if item["content"] not in seen:
            seen.add(item["content"])
            result.append({
                "type": "structured",
                "content": item["content"],
                "update_time": item["update_time"]
            })
    return result

# 测试接口：查询制造业-招聘外包的社保政策
test_result = rag_retrieval("制造业", "招聘外包", "社保")
print(test_result)

from pymilvus import MilvusClient, DataType
import torch
from modelscope.pipelines import pipeline
from modelscope.utils.constant import Tasks

# 1. 初始化中文Embedding模型（通义千问V2，需申请APIKey）
embedding_pipeline = pipeline(
    task=Tasks.text_embedding,
    model="alibaba-pai/pai-text-embedding-general-v2",
    device="cuda" if torch.cuda.is_available() else "cpu"
)

# 2. 连接Milvus（阿里云向量数据库）
client = MilvusClient(
    uri="https://xxx-xxxx.milvus.aliyuncs.com:19530",
    token="your-token"  # 阿里云控制台获取
)

# 3. 创建集合（含业务标签字段）
collection_name = "hr_case_vector"
if not client.has_collection(collection_name):
    client.create_collection(
        collection_name=collection_name,
        schema=[
            {"name": "id", "type": DataType.INT64, "is_primary": True},
            {"name": "embedding", "type": DataType.FLOAT_VECTOR, "dims": 1024},
            {"name": "content", "type": DataType.VARCHAR, "max_length": 2000},
            {"name": "industry", "type": DataType.VARCHAR, "max_length": 50},  # 业务标签
            {"name": "scene", "type": DataType.VARCHAR, "max_length": 50}       # 业务标签
        ],
        index_params={
            "index_type": "IVF_FLAT",
            "metric_type": "COSINE",  # 语义匹配用余弦相似度
            "params": {"nlist": 1024}
        }
    )

# 4. 向量生成与插入（客户案例示例）
def insert_case(industry, scene, content):
    # 生成向量
    result = embedding_pipeline({"text": content})
    embedding = result["embedding"]
    
    # 插入数据
    client.insert(
        collection_name=collection_name,
        data=[{
            "id": len(client.list_ids(collection_name)),
            "embedding": embedding,
            "content": content,
            "industry": industry,
            "scene": scene
        }]
    )

# 插入测试案例：制造业-技工招聘成功案例
test_content = "制造业技工招聘方案：1. 与XX产业带劳务市场合作；2. 提供入职奖金500元；3. 老员工推荐奖励300元，最终到岗率85%"
insert_case("制造业", "技工招聘", test_content)

def hybrid_retrieval(industry, business, scene, query):
    """
    双库协同检索：先RAG后向量，互补生成结果
    :param industry: 行业
    :param business: 业务类型
    :param scene: 服务场景
    :param query: 用户查询
    :return: 融合后的知识结果
    """
    # 1. RAG库获取结构化知识（政策/流程）
    rag_result = rag_retrieval(industry, business, query)
    
    # 2. 向量库获取非结构化知识（案例/经验）
    # 2.1 生成查询向量
    query_embedding = embedding_pipeline({"text": query})["embedding"]
    # 2.2 带业务标签过滤的向量检索
    vector_result = client.search(
        collection_name=collection_name,
        data=[query_embedding],
        filter=f"industry == '{industry}' and scene == '{scene}'",  # 业务标签过滤
        limit=3,
        output_fields=["content", "industry"]
    )[0]
    
    # 3. 格式化向量结果
    formatted_vector_result = [
        {
            "type": "unstructured",
            "content": hit["entity"]["content"],
            "similarity": hit["distance"]
        }
        for hit in vector_result if hit["distance"] > 0.7  # 过滤低相似度结果
    ]
    
    # 4. 融合结果：结构化在前，非结构化在后
    return rag_result + formatted_vector_result

# 测试：制造业-招聘外包-技工招聘场景，查询“招聘方案”
final_result = hybrid_retrieval(
    industry="制造业",
    business="招聘外包",
    scene="技工招聘",
    query="招聘方案"
)

# 打印融合结果
for idx, item in enumerate(final_result, 1):
    print(f"{idx}. 类型：{item['type']}")
    print(f"内容：{item['content']}\n")

class CrystalinkEngine:
    def __init__(self):
        # 加载任务-模型-知识映射配置
        self.task_config = {
            "policy_query": {  # 政策查询任务
                "model": "deepseek",
                "knowledge": ["rag"]
            },
            "case_analysis": {  # 案例分析任务
                "model": "gemini",
                "knowledge": ["rag", "vector"]
            },
            "video_script": {  # 短视频脚本任务
                "model": "sora2",
                "knowledge": ["vector"]
            }
        }
    
    def parse_task(self, user_input, context):
        """
        任务解析：根据用户输入和上下文判断任务类型
        :param user_input: 用户输入
        :param context: 上下文（行业、业务类型等）
        :return: 任务信息
        """
        # 简单的意图识别（生产环境可改用通义千问的意图识别API）
        if "政策" in user_input or "社保" in user_input:
            task_type = "policy_query"
        elif "案例" in user_input or "方案" in user_input:
            task_type = "case_analysis"
        elif "视频" in user_input or "脚本" in user_input:
            task_type = "video_script"
        else:
            task_type = "general_query"
        
        # 获取任务配置
        task_info = self.task_config.get(task_type, {
            "model": "tongyi",
            "knowledge": ["rag", "vector"]
        })
        
        # 补充上下文信息
        task_info["context"] = context
        task_info["user_input"] = user_input
        return task_info
    
    def dispatch(self, task_info):
        """
        资源调度：根据任务信息调用知识和模型
        """
        # 1. 获取知识
        industry = task_info["context"]["industry"]
        business = task_info["context"]["business"]
        scene = task_info["context"]["scene"]
        query = task_info["user_input"]
        
        knowledge = []
        if "rag" in task_info["knowledge"]:
            knowledge.extend(rag_retrieval(industry, business, query))
        if "vector" in task_info["knowledge"]:
            # 向量库检索需场景信息
            vector_res = self._vector_retrieval(industry, scene, query)
            knowledge.extend(vector_res)
        
        # 2. 调用模型
        model_client = self._get_model_client(task_info["model"])
        prompt = self._build_prompt(knowledge, query)
        model_result = model_client.generate(prompt)
        
        return model_result
    
    # 辅助方法：获取模型客户端、构建提示词等（省略）

class LapisCoreEngine:
    def __init__(self):
        # 初始化业务系统API客户端
        self.wechat_client = self._init_wechat_api()  # 企业微信API
        self.crm_client = self._init_crm_api()        # CRM系统API
        self.crystalink = CrystalinkEngine()          # 关联中枢引擎
    
    def ai_consultant_reply(self, user_id, user_input):
        """
        AI咨询师回复接口：直接对接企业微信
        :param user_id: 企业微信用户ID
        :param user_input: 用户输入
        :return: 回复结果
        """
        # 1. 从CRM获取用户上下文（行业、业务类型等）
        context = self.crm_client.get_customer_info(user_id)
        # 格式：{"industry": "制造业", "business": "招聘外包", "scene": "技工招聘"}
        
        # 2. 调用中枢引擎处理任务
        task_info = self.crystalink.parse_task(user_input, context)
        model_result = self.crystalink.dispatch(task_info)
        
        # 3. 直接推送到企业微信
        self.wechat_client.send_message(
            user_id=user_id,
            content=model_result,
            msg_type="text"
        )
        
        # 4. 结果存入CRM（便于后续复盘）
        self.crm_client.log_interaction(
            user_id=user_id,
            user_input=user_input,
            ai_reply=model_result
        )
        
        return {"status": "success", "reply": model_result}

# 测试：企业微信用户咨询
test_reply = LapisCoreEngine().ai_consultant_reply(
    user_id="wx123456",
    user_input="制造业技工招聘有什么好方案？"
)
print(test_reply)