Neo4j Aura Graph Analytics 및 외부 데이터 소스를 사용하여 모듈식 그래프 분석 파이프라인 구축
Neo4j Aura Graph Analytics는 클라우드에서 그래프 알고리즘을 실행하기 위한 서버리스 제품이에요. 기존 Neo4j 데이터베이스에 연결할 수 있지만, 데이터 소스가 Neo4j가 아닌 경우 standalone 모드로도 사용할 수 있어요. Pandas DataFrames에서 직접 데이터를 로드하고, PageRank 또는 커뮤니티 감지와 같은 알고리즘을 실행하고, Neo4j 데이터베이스를 건드리지 않고도 결과를 다시 스트리밍할 수 있죠. 데이터가 관계형 데이터베이스, 데이터 웨어하우스 또는 CSV 파일에 있고 그래프 인프라를 설정하지 않고 그래프 알고리즘을 적용하려는 경우에 유용해요.
MCP 서버를 통해 LLM 에이전트에 이러한 기능을 제공하려면 무엇이 필요한지 확인하고 싶었고, 몇 가지 흥미로운 과제를 해결해야 한다는 사실을 알게 되었어요.
- 각 데이터 제공업체에 대해 별도의 통합을 구축 및 유지하지 않고 외부 소스에서 데이터를 가져오는 방법
- LLM 컨텍스트에서 실제 그래프 데이터를 유지하는 방법 (데이터가 너무 많아지면 LLM이 감당하기 힘들겠죠?)
- 각 알고리즘을 개별적으로 노출하는 대신 알고리즘을 더 높은 수준의 도구로 그룹화하는 것이 적합한 경우
이번 블로그 게시물에서는 이 MCP 서버를 구축하면서 배운 내용을 살펴볼게요. 코드는 에서 확인할 수 있어요.
참고: 이는 개념 증명 구현이며 공식적으로 지원되거나 프로덕션 용도로 사용되지 않아요.
외부 데이터 소스 통합
외부 데이터 소스를 지원하는 한 가지 접근 방식은 Snowflake용, BigQuery용 등 각 제공업체에 대한 전용 통합을 구축하는 것이에요. 하지만 이는 많은 코드를 작성하고 유지 관리해야 한다는 것을 의미하며, 새로운 데이터 소스가 추가될 때마다 추가 작업이 필요하죠. 대신, 저는 이러한 시스템과 통신하는 방법을 이미 알고 있는 기존 MCP 서버를 활용할 수 있는지 확인하고 싶었어요. 외부 데이터 제공업체의 기존 MCP 구현을 마운트하는 것이 좋은 생각인 것 같았죠.
이 실험에서는 외부 데이터 공급자로 Supabase를 사용했어요. Supabase에는 이미 SQL 쿼리 실행, 테이블 나열 및 기타 데이터베이스 작업을 위한 도구를 제공하는 MCP 서버가 있거든요. 이 모든 것을 다시 구현하는 대신 Neo4j Aura Graph Analytics MCP 서버는 Supabase API 키가 환경에 제공될 때 Supabase MCP 서버를 동적으로 마운트할 수 있도록 했어요.
# Initialize MCP server
mcp = FastMCP(
name="neo4j-gds",
)
if supabase_access_token:
logger.info(f"Using Supabase MCP")
# Create the stdio transport
supabase_transport = StdioTransport(
command="npx",
args=[
"-y",
"@supabase/mcp-server-supabase@latest",
"--access-token",
supabase_access_token
]
)
# Pass transport directly - ProxyClient is created automatically
supabase_proxy = FastMCP.as_proxy(
supabase_transport,
name="Supabase"
)
# Mount the proxy
mcp.mount(supabase_proxy, prefix="supabase")API 키도 없고 Supabase 도구도 없다구요? 걱정 마세요! 이 패턴은 훌륭하게 확장될 수 있어요. 동일한 방식으로 다른 공급자에 대한 지원을 추가하여 사용 가능한 자격 증명을 기반으로 조건에 따라 MCP 서버를 마운트할 수 있답니다.
두 번째 과제는 LLM을 통한 데이터 흐름을 피하는 것이었어요. 에이전트가 Supabase에서 행을 가져와 다른 도구에 전달하여 Graph Data Science에 로드하면 해당 데이터가 모두 컨텍스트 창에 표시되잖아요. 소규모 데이터 세트의 경우에는 괜찮을 수 있지만, 확장되지는 않죠. 해결책은 데이터 세트를 가져오고 Aura 그래프 투영을 구성하는 전체 파이프라인을 내부적으로 처리하는 단일 도구를 구현하는 것이었어요.
if supabase_access_token:
@mcp.tool(
name=namespace_prefix + "load_gds_graph_from_supabase",
annotations=ToolAnnotations(
title="Load GDS Graph from Supabase",
readOnlyHint=False,
destructiveHint=False,
idempotentHint=True,
openWorldHint=True,
),
)
async def load_gds_graph_from_supabase(
project_id: str = Field(
...,
description="Supabase project ID"
),
nodes_query: str = Field(
...,
description="SQL query to fetch nodes. Must return a 'nodeId' column (integer) and optional 'labels' column (string or array) and numeric properties. Use double quotes to preserve case for aliasing. Example: SELECT id as \"nodeId\", type as \"labels\" FROM nodes"
),
relationships_query: str = Field(
...,
description="SQL query to fetch relationships. Must return 'sourceNodeId' and 'targetNodeId' columns (integers), optional 'relationshipType' column (string), and numeric properties. Use double quotes to preserve case for aliasing. Example: SELECT source_id as \"sourceNodeId\", target_id as \"targetNodeId\", type as \"relationshipType\" FROM edges"
),
graph_name: str = Field(
default="graph",
description="Name for the graph in GDS. Default is 'graph'"
),
undirected: bool = Field(
default=False,
description="If True, all relationship types will be treated as undirected. Default is False (directed)."
),
session_name: Optional[str] = Field(
default=None,
description="Name for the GDS session. If not provided, a default name will be generated"
),
memory: Optional[str] = Field(
default=None,
description="Memory allocation for the session (e.g., '4GB'). If not provided, will be estimated based on graph size"
),
ttl_hours: int = Field(
default=5,
description="Time-to-live for the session in hours. Default is 5 hours"
)
) -> list[ToolResult]:
"""Load a graph into Neo4j GDS by fetching data from Supabase via SQL queries with support for node labels, relationship types, and undirected relationships."""
... code implementation ...load_gds_graph_from_supabase 툴은 `node`와 `relationship`에 대한 SQL `query`를 받아서 Supabase의 execute_sql 툴을 호출하고, 결과를 바로 그래프 데이터 과학 세션에 로드해요. 여기서 중요한 점은 LLM은 실제 데이터를 전혀 볼 수 없고, `query`와 그래프가 로드되었다는 확인만 볼 수 있다는 거죠.
execute_sql을 사용하는 또 다른 장점은 LLM이 SQL `query` 언어를 사용해서 필요한 데이터 변환을 수행할 수 있다는 점이에요. `node`와 `relationship` 투영을 위한 기본 툴이니까요.
그룹화 알고리즘
세 번째 디자인 결정은 그래프 데이터 과학 알고리즘을 LLM에 어떻게 노출할 것인가에 대한 고민이었어요. Neo4j의 그래프 데이터 과학 라이브러리에는 중심성, 커뮤니티 감지, 경로 찾기, 유사성 등 다양한 카테고리에 걸쳐 정말 많은 알고리즘이 있거든요. 각 툴을 별도의 툴로 노출할 수도 있지만, 그렇게 하면 LLM이 선택해야 하는 툴 카탈로그가 너무 커져서 툴 선택이 더 어려워지고 토큰 비용도 많이 들게 되죠.
그래서 대신 의미론적 목적에 따라 알고리즘을 그룹화하기로 했어요. 예를 들어, 네 개의 개별 툴인 run_pagerank, run_betweenness, run_degree, run_closeness 대신에, 알고리즘 이름을 파라미터로 사용하는 하나의 run_centrality_algorithm 툴을 사용하는 거예요. 이렇게 하면 LLM이 사용할 특정 알고리즘을 완전히 제어하면서도 툴의 수를 관리하기 쉽게 유지할 수 있죠.
@mcp.tool(
name=namespace_prefix + "run_centrality_algorithm",
annotations=ToolAnnotations(
title="Run Centrality Algorithm",
readOnlyHint=True,
destructiveHint=False,
idempotentHint=True,
openWorldHint=True,
),
)
async def run_centrality_algorithm(
algorithm_name: Literal["pagerank", "betweenness", "degree", "closeness"] = Field(
...,
description="Name of the centrality algorithm to run. Options: 'pagerank', 'betweenness', 'degree', 'closeness'"
),
session_name: str = Field(
...,
description="Name of the GDS session"
),
graph_name: str = Field(
...,
description="Name of the graph to run the algorithm on"
)
커뮤니티 감지에도 똑같은 패턴이 적용돼요. run_community_detection_algorithm은 Louvain, Leiden, Label Propagation, 그리고 WCC(Weakly Connected Components)를 지원하죠. LLM은 어떤 유형의 분석을 해야 할지("커뮤니티를 찾아야 한다" 또는 "영향력을 측정해야 한다") 추론하고, 작업에 가장 적합한 특정 알고리즘을 선택할 수 있어요.
테스트 실행
설정을 테스트하기 위해서 테이블 두 개가 있는 Supabase 프로젝트를 만들었어요. 하나는 사람용, 다른 하나는 우정용이죠. 그런 다음 다음 MCP 서버 자격 증명을 사용해서 Claude Desktop을 구성했어요.
{
"mcpServers": {
"neo4j-aura-graph-analytics": {
"command": "uv",
"args": [
"--directory",
"/path/to/mcp-aura-gds-managed/mcp-neo4j-aura-gds-standalone",
"run",
"mcp-neo4j-aura-gds-standalone",
"--transport",
"stdio",
"--namespace",
"dev"
],
"env": {
"CLIENT_ID": "",
"CLIENT_SECRET": "",
"SUPABASE_ACCESS_TOKEN": ""
}
}
}
}다음으로 Claude에게 우정 그래프를 Graph Data Science 세션에 로드하고, 거기에서 PageRank와 WCC를 실행하도록 요청했어요. Claude는 다음 매개변수를 사용해서 그래프 데이터 과학 그래프를 로드하기 전에 Supabase에서 데이터를 식별하기 위해 4단계를 수행했죠.
총 9단계를 거쳐 워크플로가 완료되었고, LLM이 결과를 제시했어요.
중심성 알고리즘의 경우 상위 10개 결과를 LLM으로 스트리밍하고, 커뮤니티 감지 알고리즘의 경우 집계 통계를 반환해요.
요약
이 개념 증명은 통합을 처음부터 다시 만드는 대신 기존 MCP 서버를 구성해서 모듈형 그래프 분석 파이프라인을 구축할 수 있다는 걸 보여줘요. Supabase와 같은 외부 제공자를 탑재하고, 원시 데이터를 LLM 컨텍스트에서 제외하고, 알고리즘을 의미론적으로 그룹화하면 LLM 에이전트가 외부 소스에서 데이터를 가져오고 그래프 알고리즘을 실행하며 전용 Graph Database를 건드리거나 데이터 행으로 컨텍스트 창을 압도하지 않고도 통찰력을 얻을 수 있는 시스템이 완성되는 거죠.
이 접근 방식은 다양한 MCP 서버를 동적으로 마운트해서 확장돼요. Supabase를 Snowflake 또는 BigQuery로 바꾸더라도 아키텍처는 똑같이 유지될 거예요. 데이터가 아직 그래프 형식이 아닌 탐색적 워크플로의 경우, 이 아키텍처는 그래프 알고리즘을 에이전트 워크플로로 가져오는 간단한 방법을 제공하죠.
코드는 다음에서 사용할 수 있어요. .
필수 GraphRAG
Knowledge Graph를 통해 RAG의 잠재력을 최대한 활용하세요. 한정된 기간 동안 Manning으로부터 최종 가이드를 무료로 받아보세요.
- 에이전트 AI
- mcp 서버
에이치시스템즈의 LogTree는 Neo4j 기반 GraphRAG 플랫폼으로, 데이터를 자동으로 지식그래프화하고 자연어 질의로 즉시 답을 제공합니다.
'Agent AI' 카테고리의 다른 글
| 실제 프로덕션에서 효과적인 AI 에이전트 활용 사례 연구 (0) | 2026.04.03 |
|---|---|
| Neo4j 에이전트 기반 멀티 유저 던전 만들기 (0) | 2026.04.03 |
| 에이전트 기반 AI 아키텍처란 무엇일까요? 일반적인 패턴과 활용 시점 (1) | 2026.04.02 |
| Java와 Neo4j로 구현하는 에이전트 기반 AI (0) | 2026.04.02 |
| 에이전트 AI vs. 생성 AI: 에이전트에게 기억, 맥락, 그리고 안전장치가 필요한 이유 (0) | 2026.04.02 |