"""检索步骤"""

step_name: str

step_description: str

start_time: float

end_time: float

duration: float

status: str # "success", "error", "warning"

result_count: int

"""整合响应结果"""

answer: str

sources: Dict[str, Any]

metadata: Dict[str, Any]

"""HAG整合API - 使用LangChain Runnable模式"""

def __init__(self):

"""初始化整合API"""

try:

logger.info("开始初始化HAG整合API...")

# 加载配置

self.config = get_config()

# 初始化基础服务

self._init_services()

# 构建LangChain Runnable管道

self._build_runnable_chain()

logger.info("HAG整合API初始化完成")

except Exception as e:

logger.error(f"HAG整合API初始化失败: {e}")

raise

def _init_services(self):

"""初始化所有服务组件"""

try:

# 1. 初始化向量化服务

self.embedding_service = OllamaEmbeddingService()

# 2. 初始化向量存储服务

self.vector_store = WeaviateVectorStore()

# 3. 初始化Weaviate检索服务

self.retrieval_service = RetrievalService(

embedding_service=self.embedding_service,

vector_store=self.vector_store

)

# 4. 初始化向量化图谱检索服务

from src.services.vectorized_graph_retrieval_service import GraphSearchConfig

from src.knowledge.vectorized_graph_retrieval import KnowledgeChainConfig

# 创建向量化图谱存储

from src.knowledge.vectorized_graph_storage import VectorizedGraphStorage

vectorized_storage = VectorizedGraphStorage()

# 配置搜索参数

search_config = GraphSearchConfig(

max_nodes=20,

max_relations=30,

max_chains=10,

min_similarity=0.7,

enable_chain_building=True,

enable_async_search=True,

search_timeout=30.0

)

# 配置知识链路参数

chain_config = KnowledgeChainConfig(

max_chain_length=5,

min_similarity_threshold=0.7,

max_nodes_per_query=20,

max_relations_per_query=30,

chain_weight_decay=0.8,

relation_boost_factor=1.2

)

self.graph_service = VectorizedGraphRetrievalService(

storage=vectorized_storage,

config=search_config,

chain_config=chain_config

)

# 5. 初始化混合检索服务

self.hybrid_service = HybridRetrievalService(

document_retrieval_service=self.retrieval_service,

graph_retrieval_service=self.graph_service,

doc_weight=0.6,

graph_weight=0.4,

weight_strategy="adaptive",

enable_dynamic_weights=True,

weight_manager_config={

"learning_rate": 0.01,

"cache_size": 1000,

"enable_gnn": True,

"gnn_config": {

"hidden_dim": 128,

"num_layers": 2,

"dropout": 0.1,

"learning_rate": 0.001

}

},

enable_ab_testing=True,

ab_test_config={

"strategies": ["static", "adaptive", "gnn_driven"],

"traffic_split": [0.3, 0.4, 0.3],

"min_samples": 100

},

cache_config={

"cache_type": "hybrid",

"max_size": 2000,

"default_ttl": 3600,

"redis_config": {

"host": "localhost",

"port": 6379,

"db": 0,

"key_prefix": "hag_cache:"

},

"enable_compression": False,

"enable_async": True

},

enable_concurrent_queries=True,

max_workers=4,

connection_pool_config={

"neo4j": {

"uri": "bolt://localhost:7687",

"username": "neo4j",

"password": "password",

"database": "neo4j",

"min_connections": 2,

"max_connections": 10,

"max_connection_age": 3600,

"max_idle_time": 300

},

"weaviate": {

"url": "http://localhost:8080",

"api_key": None,

"additional_headers": {},

"min_connections": 2,

"max_connections": 8,

"max_connection_age": 3600,

"max_idle_time": 300

}

},

enable_performance_monitoring=True

)

# 6. 初始化LLM服务

self.llm_service = OllamaLLMService()

# 7. 初始化RAG管道

self.rag_pipeline = RAGPipeline(

hybrid_retrieval_service=self.hybrid_service,

llm_service=self.llm_service

)

logger.info("所有服务组件初始化完成")

except Exception as e:

logger.error(f"服务初始化失败: {e}")

raise

def _build_runnable_chain(self):

"""构建LangChain Runnable管道"""

try:

# 定义提示模板

prompt_template = ChatPromptTemplate.from_template("""

""")

# 构建Runnable链

self.runnable_chain = (

{

"question": RunnablePassthrough(),

"documents": RunnableLambda(self._retrieve_documents),

"entities": RunnableLambda(self._retrieve_entities),

"relationships": RunnableLambda(self._retrieve_relationships)

}

| prompt_template

| RunnableLambda(self._generate_answer)

| StrOutputParser()

)

logger.info("LangChain Runnable管道构建完成")

except Exception as e:

logger.error(f"Runnable管道构建失败: {e}")

raise

def _retrieve_documents(self, question: str) -> str:

"""检索相关文档 - Weaviate向量检索（余弦相似度+欧式距离）"""

try:

# 使用混合检索获取Top5文档

hybrid_result = self.retrieval_service.search_hybrid(question, limit=5)

documents = []

for result in hybrid_result.hybrid_results[:5]: # 确保只取Top5

documents.append(f"- {result.content[:200]}...")

return "\n".join(documents) if documents else "未找到相关文档"

except Exception as e:

logger.error(f"文档检索失败: {e}")

return "文档检索出错"

def _retrieve_entities(self, question: str) -> str:

"""检索相关实体 - Neo4j图谱检索"""

try:

# 将问题转换为向量，然后检索相关实体

query_vector = self.embedding_service.embed_text(question)

# 使用图谱服务检索实体

entities = self.graph_service.search_entities_by_name(question, limit=2)

entity_list = []

for entity in entities[:2]: # 确保只取2个节点

name = entity.get('name', '')

entity_type = entity.get('type', '')

description = entity.get('description', '')

entity_list.append(f"- {name} ({entity_type}): {description}")

return "\n".join(entity_list) if entity_list else "未找到相关实体"

except Exception as e:

logger.error(f"实体检索失败: {e}")

return "实体检索出错"

def _retrieve_relationships(self, question: str) -> str:

"""检索相关关系 - Neo4j图谱检索（正确逻辑：先找节点，再找关系）"""

try:

# 第一步：根据问题找到相关的实体节点

relevant_entities = self.graph_service.search_entities_by_name(question, limit=3)

if not relevant_entities:

return "未找到相关实体，无法检索关系"

# 第二步：基于找到的实体节点，查找它们的关系

all_relationships = []

for entity in relevant_entities:

entity_name = entity.get('name', '')

if entity_name:

# 获取该实体的关系网络

entity_rels = self.graph_service.get_entity_relationships(entity_name, limit=5)

relationships = entity_rels.get('relationships', [])

# 处理关系数据

for rel in relationships:

source = rel.get('entity', entity_name)

target = rel.get('related_entity', '')

rel_type = rel.get('relation_type', '')

description = rel.get('relation_description', '') or rel.get('related_description', '')

# 确保关系描述不为空

if not description:

description = f"{source} 与 {target} 之间的 {rel_type} 关系"

all_relationships.append({

'source': source,

'target': target,

'type': rel_type,

'description': description

})

# 第三步：去重并格式化输出

seen_relations = set()

unique_relationships = []

for rel in all_relationships:

rel_key = (rel['source'], rel['type'], rel['target'])

if rel_key not in seen_relations:

seen_relations.add(rel_key)

unique_relationships.append(rel)

# 限制输出数量并格式化

rel_list = []

for rel in unique_relationships[:10]: # 最多10个关系

source = rel['source']

target = rel['target']

rel_type = rel['type']

description = rel['description'][:100] + "..." if len(rel['description']) > 100 else rel['description']

rel_list.append(f"- {source} --[{rel_type}]--> {target}: {description}")

return "\n".join(rel_list) if rel_list else "未找到相关关系"

except Exception as e:

logger.error(f"关系检索失败: {e}")

return "关系检索出错"

def _generate_answer(self, prompt_data: Dict[str, Any]) -> str:

"""使用LLM生成答案"""

try:

# 构建完整的提示词

prompt_text = prompt_data.content if hasattr(prompt_data, 'content') else str(prompt_data)

# 使用LLM生成回答

answer = self.llm_service.generate_response(

prompt=prompt_text,

temperature=0.7,

max_tokens=1000

)

return answer

except Exception as e:

logger.error(f"答案生成失败: {e}")

return "抱歉，无法生成答案"

def query(self, question: str) -> IntegratedResponse:

"""

import time

retrieval_steps = []

try:

logger.info(f"开始处理查询: {question}")

# 步骤1: 问题分析和向量化

step1_start = time.time()

try:

query_vector = self.embedding_service.embed_text(question)

step1_end = time.time()

retrieval_steps.append(RetrievalStep(

step_name="问题向量化",

step_description="将用户问题转换为向量表示",

start_time=step1_start,

end_time=step1_end,

duration=step1_end - step1_start,

status="success",

result_count=1,

details={"vector_dimension": len(query_vector) if query_vector else 0}

))

except Exception as e:

step1_end = time.time()

retrieval_steps.append(RetrievalStep(

step_name="问题向量化",

step_description="将用户问题转换为向量表示",

start_time=step1_start,

end_time=step1_end,

duration=step1_end - step1_start,

status="error",

result_count=0,

details={"error": str(e)}

))

# 步骤2: 文档检索 (Weaviate)

step2_start = time.time()

try:

hybrid_result = self.retrieval_service.search_hybrid(question, limit=5)

documents = hybrid_result.hybrid_results[:5]

step2_end = time.time()

retrieval_steps.append(RetrievalStep(

step_name="文档检索",

step_description="从Weaviate向量数据库检索相关文档",

start_time=step2_start,

end_time=step2_end,

duration=step2_end - step2_start,

status="success",

result_count=len(documents),

details={

"search_method": "hybrid_cosine_euclidean",

"top_scores": [doc.score for doc in documents[:3]] if documents else []

}

))

except Exception as e:

step2_end = time.time()

documents = []

retrieval_steps.append(RetrievalStep(

step_name="文档检索",

step_description="从Weaviate向量数据库检索相关文档",

start_time=step2_start,

end_time=step2_end,

duration=step2_end - step2_start,

status="error",

result_count=0,

details={"error": str(e)}

))

# 步骤3: 实体检索 (Neo4j)

step3_start = time.time()

try:

entities = self.graph_service.search_entities_by_name(question, limit=2)

step3_end = time.time()

retrieval_steps.append(RetrievalStep(

step_name="实体检索",

step_description="从Neo4j图数据库检索相关实体",

start_time=step3_start,

end_time=step3_end,

duration=step3_end - step3_start,

status="success",

result_count=len(entities),

details={

"entity_types": [entity.get('type', '') for entity in entities] if entities else []

}

))

except Exception as e:

step3_end = time.time()

entities = []

retrieval_steps.append(RetrievalStep(

step_name="实体检索",

step_description="从Neo4j图数据库检索相关实体",

start_time=step3_start,

end_time=step3_end,

duration=step3_end - step3_start,

status="error",

result_count=0,

details={"error": str(e)}

))

# 步骤4: 关系检索 (Neo4j) - 使用智能关系搜索

step4_start = time.time()

try:

# 使用优化的关系搜索方法，基于查询内容和相关性评分

unique_relationships = self.graph_service.search_relationships_by_query(question, limit=10)

logger.info(f"智能关系搜索结果: {len(unique_relationships)} 个关系")

for rel in unique_relationships[:3]: # 记录前3个关系用于调试

logger.info(f" - {rel.get('source', '')} -> {rel.get('type', '')} -> {rel.get('target', '')} (评分: {rel.get('relevance_score', 0)})")

step4_end = time.time()

retrieval_steps.append(RetrievalStep(

step_name="关系检索",

step_description="从Neo4j图数据库检索实体间关系（基于查询相关性）",

start_time=step4_start,

end_time=step4_end,

duration=step4_end - step4_start,

status="success",

result_count=len(unique_relationships),

details={

"relation_types": list(set([rel.get('type', '') for rel in unique_relationships])) if unique_relationships else [],

"avg_relevance_score": sum([rel.get('relevance_score', 0) for rel in unique_relationships]) / len(unique_relationships) if unique_relationships else 0,

"top_relations": [f"{rel.get('source', '')} -> {rel.get('type', '')} -> {rel.get('target', '')}" for rel in unique_relationships[:3]]

}

))

except Exception as e:

step4_end = time.time()

unique_relationships = []

retrieval_steps.append(RetrievalStep(

step_name="关系检索",

step_description="从Neo4j图数据库检索实体间关系",

start_time=step4_start,

end_time=step4_end,

duration=step4_end - step4_start,

status="error",

result_count=0,

details={"error": str(e)}

))

# 步骤5: 混合检索结果整合

step5_start = time.time()

try:

hybrid_result = self.hybrid_service.search_hybrid(question, doc_top_k=5, graph_top_k=4)

step5_end = time.time()

retrieval_steps.append(RetrievalStep(

step_name="混合检索整合",

step_description="整合文档和图谱检索结果",

start_time=step5_start,

end_time=step5_end,

duration=step5_end - step5_start,

status="success",

result_count=len(hybrid_result.documents) + len(hybrid_result.entities) + len(hybrid_result.relationships),

details={

"doc_weight": 0.6,

"graph_weight": 0.4,

"total_documents": len(hybrid_result.documents),

"total_entities": len(hybrid_result.entities),

"total_relationships": len(hybrid_result.relationships)

}

))

except Exception as e:

step5_end = time.time()

hybrid_result = None

retrieval_steps.append(RetrievalStep(

step_name="混合检索整合",

step_description="整合文档和图谱检索结果",

start_time=step5_start,

end_time=step5_end,

duration=step5_end - step5_start,

status="error",

result_count=0,

details={"error": str(e)}

))

# 步骤6: LLM答案生成

step6_start = time.time()

try:

answer = self.runnable_chain.invoke(question)

step6_end = time.time()

retrieval_steps.append(RetrievalStep(

step_name="答案生成",

step_description="使用LLM基于检索结果生成答案",

start_time=step6_start,

end_time=step6_end,

duration=step6_end - step6_start,

status="success",

result_count=1,

details={

"model": "gemma3:4b",

"answer_length": len(answer),

"temperature": 0.7

}

))

except Exception as e:

step6_end = time.time()

answer = f"抱歉，生成答案时出现错误: {str(e)}"

retrieval_steps.append(RetrievalStep(

step_name="答案生成",

step_description="使用LLM基于检索结果生成答案",

start_time=step6_start,

end_time=step6_end,

duration=step6_end - step6_start,

status="error",

result_count=0,

details={"error": str(e)}

))

# 构建响应

sources = {"documents": [], "entities": [], "relationships": []}

if hybrid_result:

sources = {

"documents": [

{

"content": doc["content"][:200] + "...",

"score": doc["score"],

"metadata": doc.get("metadata", {})

}

for doc in hybrid_result.documents[:5]

],

"entities": [

{

"name": entity.get("name", ""),

"type": entity.get("type", ""),

"properties": entity.get("properties", {})

}

for entity in hybrid_result.entities[:2]

],

"relationships": [

{

"source": rel.get("source", ""),

"target": rel.get("target", ""),

"type": rel.get("type", ""),

"description": rel.get("description", "")[:100] + "..."

}

for rel in hybrid_result.relationships[:10]

]

}

response = IntegratedResponse(

answer=answer,

sources=sources,

metadata={

"question": question,

"retrieval_metadata": hybrid_result.metadata if hybrid_result else {},

"processing_method": "langchain_runnable_with_detailed_steps",

"total_processing_time": sum([step.duration for step in retrieval_steps])

},

retrieval_process=retrieval_steps

)

logger.info(f"查询处理完成: 答案长度={len(answer)}, 检索步骤={len(retrieval_steps)}")

return response

except Exception as e:

logger.error(f"查询处理失败: {e}")

return IntegratedResponse(

answer=f"抱歉，处理查询时出现错误: {str(e)}",

sources={"documents": [], "entities": [], "relationships": []},

metadata={"error": str(e)},

retrieval_process=retrieval_steps

)

def get_system_status(self) -> Dict[str, Any]:

"""获取系统状态"""

try:

return {

"status": "active",

"services": {

"weaviate": "connected",

"neo4j": "connected",

"ollama": "connected"

},

"retrieval_stats": self.retrieval_service.get_stats(),

"graph_stats": self.graph_service.get_stats()

}

except Exception as e:

"""获取API实例（单例模式）"""

global api

if api is None:

api = HAGIntegratedAPI()

"""

"""查询接口"""

try:

question = request.get("question", "")

if not question:

return {"status": "error", "message": "问题不能为空"}

result = query_knowledge(question)

return {

"status": "success",

"data": {

"answer": result.answer,

"sources": result.sources,

"metadata": result.metadata,

"retrieval_process": [

{

"step_name": step.step_name,

"step_description": step.step_description,

"duration": step.duration,

"status": step.status,

"result_count": step.result_count,

"details": step.details

}

for step in result.retrieval_process

]

}

}

except Exception as e:

logger.error(f"API查询失败: {e}")

"""系统状态接口"""

try:

status = get_api().get_system_status()

return {"status": "success", "data": status}

except Exception as e:

logger.error(f"获取系统状态失败: {e}")

"""获取性能统计信息"""

try:

from src.services.performance_monitor import performance_monitor

stats = performance_monitor.get_current_stats()

return {"status": "success", "data": stats}

except Exception as e:

logger.error(f"获取性能统计失败: {e}")

"""获取最近的查询记录"""

try:

from src.services.performance_monitor import performance_monitor

queries = performance_monitor.get_recent_queries(limit)

return {"status": "success", "data": queries}

except Exception as e:

logger.error(f"获取最近查询记录失败: {e}")

"""获取性能洞察和建议"""

try:

from src.services.performance_monitor import performance_monitor

insights = performance_monitor.get_performance_insights()

return {"status": "success", "data": insights}

except Exception as e:

logger.error(f"获取性能洞察失败: {e}")

"""导出性能指标到文件"""

try:

from src.services.performance_monitor import performance_monitor

performance_monitor.export_metrics(filepath)

return {"status": "success", "message": f"性能指标已导出到: {filepath}"}

except Exception as e:

logger.error(f"导出性能指标失败: {e}")