使用说明
MCP KG查询器
用于知识图引擎集成的MCP客户端。
概述
mcp_kg_inquirer 是一个Python客户端，用于通过模型上下文协议（MCP）与知识图引擎交互。它提供：
- 搜索操作：矢量、text2cypher、Vector_cypher和混合搜索模式
- RAG查询：用于问答的检索增强生成
- 直接HTTP传输：带有重试逻辑的简单GraphQL请求
- 断路器：防止连锁故障的弹性模式
安装
pip install mcp_kg_inquirer
依赖项
- 派丹提克 （>=2.0.0）-数据验证
- python dotenv （>=1.0.0）-环境变量加载
快速开始
import asyncio
import logging
from mcp_kg_inquirer import MCPKGInquirer
async def main():
# Create client with API key
client = MCPKGInquirer(
logging.getLogger(__name__),
endpoint_id="my-endpoint",
part_id="my-partition",
api_key="my-api-key",
)
# Or create client with bearer token (preferred for OAuth/JWT)
client = MCPKGInquirer(
logging.getLogger(__name__),
endpoint_id="my-endpoint",
part_id="my-partition",
bearer_token="my-jwt-token",
)
# Search using vector similarity
results = await client.search_vector("machine learning", limit=10)
print(f"Found {results.total} results")
for result in results.results:
print(f" [{result.score:.2f}] {result.content[:100]}...")
# Search using hybrid mode
results = await client.search_hybrid("neural networks")
# RAG query
answer = await client.rag("What is machine learning?")
print(f"Answer: {answer.answer}")
print(f"Context: {len(answer.context)} documents")
# Close client
await client.close()
asyncio.run(main())
配置
认证
客户端支持两种身份验证方法：
选项1：API密钥
import logging
client = MCPKGInquirer(
logging.getLogger(__name__),
endpoint_id="my-endpoint",
part_id="my-partition",
api_key="my-api-key",
)
选项2：承载令牌（OAuth/JWT）
import logging
client = MCPKGInquirer(
logging.getLogger(__name__),
endpoint_id="my-endpoint",
part_id="my-partition",
bearer_token="my-jwt-token",
)
环境变量
# Required
export KGE_ENDPOINT_ID="my-endpoint"
export KGE_PART_ID="my-partition"
# Authentication (provide ONE)
export KGE_API_KEY="my-api-key" # Option 1: API key
export KGE_BEARER_TOKEN="my-jwt-token" # Option 2: Bearer token (preferred)
# Optional
export KGE_BASE_URL="http://localhost:8000"
export KGE_TIMEOUT="30.0"
export KGE_MAX_RETRIES="3"
来自环境
import logging
from mcp_kg_inquirer import MCPKGInquirer
client = MCPKGInquirer.from_env(logging.getLogger(__name__))
用法
搜索
# Basic search
results = await client.search("machine learning")
# Vector search (semantic similarity)
results = await client.search_vector("neural networks")
# Hybrid search (vector + fulltext)
results = await client.search_hybrid("deep learning")
# Search with filters
results = await client.search(
"AI",
data_type="article",
filters={"category": "technology"},
page=1,
limit=20,
)
# Text-to-Cypher search (LLM-generated Cypher)
results = await client.search(
"find products with price > 100",
search_mode="text2cypher",
)
# Vector-Cypher search (vector + custom traversal)
results = await client.search(
"machine learning algorithms",
search_mode="vector_cypher",
prompt="Focus on algorithmic complexity",
)
检索增强生成
# Basic RAG
answer = await client.rag("What is deep learning?")
# With custom prompt
answer = await client.rag(
"Explain neural networks",
prompt="You are a technical assistant. Be concise.",
)
# With use case (predefined prompts)
answer = await client.rag(
"Summarize AI",
use_case="summary", # Options: default, summary, explanation
)
异步上下文管理器
import logging
async with MCPKGInquirer.from_env(logging.getLogger(__name__)) as client:
results = await client.search("test")
# Client automatically closed
搜索模式
| 模式 | 描述 | 最适合 |
|---|---|---|
vector |
使用嵌入的语义相似性 | 自然语言查询 |
text2cypher |
LLM生成的Cypher查询 | 结构化查询 |
vector_cypher |
矢量+自定义遍历 | 复杂的图形模式 |
hybrid |
矢量+全文组合 | 混合搜索需求 |
API 参考
MCPKG查询器
class MCPKGInquirer:
def __init__(
self,
logger: logging.Logger,
**setting: dict[str, Any],
): ...
async def search(
self,
query_text: str,
data_type: str = None,
filters: dict = None,
similarity_search: bool = False,
relevance_search: bool = False,
page: int = 1,
limit: int = 10,
prompt: str = None,
search_mode: str = None,
) -> SearchResponse: ...
async def search_vector(
self,
query_text: str,
data_type: str = None,
filters: dict = None,
page: int = 1,
limit: int = 10,
) -> SearchResponse: ...
async def search_hybrid(
self,
query_text: str,
data_type: str = None,
filters: dict = None,
page: int = 1,
limit: int = 10,
) -> SearchResponse: ...
async def rag(
self,
query_text: str,
prompt: str = None,
use_case: str = None,
) -> RAGResponse: ...
async def close() -> None: ...
搜索响应
@dataclass
class SearchResponse:
results: List[SearchResult]
total: int
page: int
limit: int
@dataclass
class SearchResult:
content: str
score: float
metadata: dict
node_id: str
labels: List[str]
RAG响应
@dataclass
class RAGResponse:
answer: str
context: List[RAGContext]
model: str
latency_ms: float
@dataclass
class RAGContext:
content: str
score: float
metadata: dict
source: str
错误处理
from mcp_kg_inquirer.exceptions import (
MCPKGInquirerError,
ConnectionError,
AuthenticationError,
SearchError,
RAGError,
ValidationError,
TimeoutError,
RateLimitError,
)
try:
results = await client.search("test")
except ValidationError as e:
print(f"Invalid parameters: {e}")
except ConnectionError as e:
print(f"Connection failed: {e}")
except RateLimitError as e:
print(f"Rate limit exceeded: {e}")
except SearchError as e:
print(f"Search failed: {e}")
发展
安装开发依赖项
pip install -e ".[dev]"
运行测试
pytest
运行覆盖率测试
pytest --cov=mcp_kg_inquirer --cov-report=html
类型检查
mypy mcp_kg_inquirer
代码检查
ruff check mcp_kg_inquirer
建筑学
mcp_kg_inquirer/
├── __init__.py # Public API
├── client.py # Main client class
├── config.py # Configuration management
├── connection.py # HTTP connection pooling
├── exceptions.py # Custom exceptions
├── search.py # Search implementation
├── rag.py # RAG implementation
└── utils.py # Helper functions
后端集成
客户端与 knowledge_graph_engine 后端：
┌─────────────────┐ HTTP/GraphQL ┌─────────────────┐
│ mcp_kg_inquirer │ ───────────────────── │ knowledge_graph │
│ (Client) │ │ Engine │
└─────────────────┘ └─────────────────┘
│
│ Neo4j Driver
▼
┌─────────────────┐
│ Neo4j DB │
│ (Vector + Graph)│
└─────────────────┘
许可证
MIT 许可证
