Introduction

GraphRAG (Graph-enhanced Retrieval-Augmented Generation) combines the structural power of knowledge graphs with the reasoning capabilities of Large Language Models like Claude. Instead of simply retrieving documents, GraphRAG extracts relevant subgraphs from your data, converts them to a text format Claude can understand, and uses that context to answer questions with graph-aware intelligence.

Why GraphRAG + Claude?

Key Concepts

Subgraph Extraction

Starting from an anchor node (found via vector search or specified directly), the system performs a Breadth-First Search (BFS) to collect nearby nodes and the edges connecting them. You control:

• hops — How many relationship steps to traverse (default: 2)
• max_nodes — Maximum nodes to include (default: 50)

Linearization

The extracted subgraph is converted to text that Claude can process. AstraeaDB supports four formats:

Token Budgeting

Claude has a context window limit. The pipeline estimates tokens (~4 chars per token) and automatically reduces the subgraph size if it would exceed your budget.

Prerequisites

1. AstraeaDB Server

# Build and start the server
cargo build --workspace
cargo run -p astraea-cli -- serve

2. Anthropic API Key

Get your API key from console.anthropic.com

# Set environment variable
export ANTHROPIC_API_KEY="sk-ant-api03-..."

3. Python Client (Optional)

pip install astraeadb anthropic

4. Go Client (Optional)

go get github.com/AstraeaDB/AstraeaDB-Official

5. Java Client (Optional)

// Add to your build.gradle.kts
dependencies {
    implementation("com.astraeadb:astraeadb-unified:0.1.0")
}

Build a Knowledge Graph

Before using GraphRAG, you need data in your graph. Let's create a simple knowledge graph about a software company.

Using Python

from astraeadb import JsonClient

with JsonClient() as client:
    # Create nodes with embeddings (from your embedding model)
    alice = client.create_node(
        labels=["Person", "Engineer"],
        properties={"name": "Alice", "role": "Senior Backend Engineer"},
        embedding=[0.1, 0.2, 0.3, ...]  # 128-dim vector
    )

    bob = client.create_node(
        labels=["Person", "Manager"],
        properties={"name": "Bob", "role": "Engineering Manager"},
        embedding=[0.15, 0.25, 0.35, ...]
    )

    project = client.create_node(
        labels=["Project"],
        properties={"name": "GraphDB", "status": "active"},
        embedding=[0.5, 0.1, 0.2, ...]
    )

    # Create relationships
    client.create_edge(alice, bob, "REPORTS_TO")
    client.create_edge(alice, project, "WORKS_ON", {"since": "2024-01"})
    client.create_edge(bob, project, "MANAGES")

Using JSON Protocol

echo '{"type":"CreateNode","labels":["Person"],"properties":{"name":"Alice"},"embedding":[0.1,0.2,0.3]}' | nc localhost 7687

Extract Subgraphs

The ExtractSubgraph operation retrieves a portion of the graph centered on a node.

JSON Request

{
  "type": "ExtractSubgraph",
  "center": 1,              // Anchor node ID
  "hops": 2,                // BFS depth
  "max_nodes": 50,          // Node limit
  "format": "structured"   // Linearization format
}

Response

{
  "status": "ok",
  "data": {
    "center": 1,
    "node_count": 5,
    "edge_count": 4,
    "text": "Node [Person: Alice] (role: \"Senior Backend Engineer\")\n  -[REPORTS_TO]-> [Person: Bob] ..."
  }
}

Linearization Formats

Choose the format that works best for your use case and Claude's understanding.

Structured (Default)

Indented tree format with arrows showing relationships. Best for general use.

Node [Person: Alice] (role: "Senior Backend Engineer")
  -[REPORTS_TO]-> [Person: Bob] (role: "Engineering Manager")
  -[WORKS_ON {since: "2024-01"}]-> [Project: GraphDB] (status: "active")
Node [Person: Bob] (role: "Engineering Manager")
  -[MANAGES]-> [Project: GraphDB]

Prose

Natural language sentences. Best when you want Claude to read it like text.

Alice is a Person and Engineer (role: "Senior Backend Engineer"). Alice REPORTS_TO Bob.
Alice WORKS_ON GraphDB (since: "2024-01"). Bob is a Person and Manager
(role: "Engineering Manager"). Bob MANAGES GraphDB. GraphDB is a Project (status: "active").

Triples

Subject-predicate-object format. Best for fact-based reasoning.

(Alice, REPORTS_TO, Bob)
(Alice, WORKS_ON, GraphDB)
(Bob, MANAGES, GraphDB)

JSON

Full JSON structure. Best when you need precise property access.

{
  "nodes": [
    {"id": 1, "labels": ["Person", "Engineer"], "properties": {"name": "Alice", "role": "Senior Backend Engineer"}},
    {"id": 2, "labels": ["Person", "Manager"], "properties": {"name": "Bob", "role": "Engineering Manager"}},
    {"id": 3, "labels": ["Project"], "properties": {"name": "GraphDB", "status": "active"}}
  ],
  "edges": [
    {"source": 1, "target": 2, "type": "REPORTS_TO"},
    {"source": 1, "target": 3, "type": "WORKS_ON", "properties": {"since": "2024-01"}},
    {"source": 2, "target": 3, "type": "MANAGES"}
  ]
}

Token Budgeting

Claude models have context window limits. The GraphRAG pipeline respects your token budget automatically.

Setting Token Budget (Rust)

let config = GraphRagConfig {
    hops: 2,
    max_context_nodes: 100,
    text_format: TextFormat::Structured,
    token_budget: 4000,  // Fits in smaller context windows
    system_prompt: None,
};

Configure Claude as LLM Provider

AstraeaDB's astraea-rag crate provides an AnthropicProvider that formats requests for Claude's Messages API.

Rust Configuration

use astraea_rag::{AnthropicProvider, LlmConfig, ProviderType};

let config = LlmConfig {
    provider: ProviderType::Anthropic,
    model: "claude-sonnet-4-20250514".into(),
    api_key: Some(std::env::var("ANTHROPIC_API_KEY").unwrap()),
    endpoint: "https://api.anthropic.com/v1".into(),
    temperature: 0.7,
    max_tokens: 1024,
};

let anthropic = AnthropicProvider::new(config);

HTTP Callback Pattern

The provider is HTTP-agnostic. You inject your HTTP client as a callback:

let anthropic = AnthropicProvider::new(config)
    .with_http_fn(|url, body| {
        // Use reqwest, hyper, or any HTTP client
        let client = reqwest::blocking::Client::new();
        let response = client
            .post(url)
            .header("x-api-key", std::env::var("ANTHROPIC_API_KEY").unwrap())
            .header("anthropic-version", "2023-06-01")
            .header("content-type", "application/json")
            .json(body)
            .send()?
            .text()?;
        Ok(response)
    });

Request Format Sent to Claude

The provider formats requests for Claude's Messages API:

{
  "model": "claude-sonnet-4-20250514",
  "max_tokens": 1024,
  "system": "You are a helpful assistant. Use the following knowledge graph context...",
  "messages": [
    {
      "role": "user",
      "content": "Based on the context above, answer: Who manages the GraphDB project?"
    }
  ],
  "temperature": 0.7
}

Rust API

Full Pipeline Example

use astraea_rag::{
    graph_rag_query_anchored,
    GraphRagConfig, GraphRagResult,
    AnthropicProvider, LlmConfig, ProviderType,
    TextFormat,
};
use astraea_core::types::NodeId;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 1. Set up the graph (your GraphOps implementation)
    let graph = MyGraph::load("data/")?;

    // 2. Configure Claude
    let llm_config = LlmConfig {
        provider: ProviderType::Anthropic,
        model: "claude-sonnet-4-20250514".into(),
        api_key: Some(std::env::var("ANTHROPIC_API_KEY")?),
        endpoint: "https://api.anthropic.com/v1".into(),
        temperature: 0.7,
        max_tokens: 2048,
    };

    let claude = AnthropicProvider::new(llm_config)
        .with_http_fn(http_callback);

    // 3. Configure GraphRAG
    let rag_config = GraphRagConfig {
        hops: 2,
        max_context_nodes: 50,
        text_format: TextFormat::Structured,
        token_budget: 8000,
        system_prompt: Some(
            "You are analyzing a knowledge graph about a software company. \
             Answer questions based on the graph context provided.".into()
        ),
    };

    // 4. Run the query
    let result: GraphRagResult = graph_rag_query_anchored(
        &graph,
        &claude,
        "Who is working on the GraphDB project and who do they report to?",
        NodeId(3),  // GraphDB project node
        &rag_config,
    )?;

    println!("Answer: {}", result.answer);
    println!("Context included {} nodes ({} tokens)",
             result.nodes_in_context, result.estimated_tokens);

    Ok(())
}

With Vector Search

If you don't know the anchor node, use vector search to find it:

use astraea_rag::graph_rag_query;

// Encode question to embedding (using your embedding model)
let question = "Who manages the database project?";
let question_embedding = embedding_model.encode(question)?;

let result = graph_rag_query(
    &graph,
    &vector_index,  // Your VectorIndex implementation
    &claude,
    question,
    &question_embedding,
    &rag_config,
)?;

println!("Found anchor node: {}", result.anchor_node_id);
println!("Answer: {}", result.answer);

Python API

While the Python client doesn't have built-in GraphRAG methods yet, you can use raw JSON requests combined with the Anthropic Python SDK.

Complete Python Example

import json
import socket
import anthropic

class GraphRAGClient:
    """GraphRAG client combining AstraeaDB with Claude."""

    def __init__(self, host="127.0.0.1", port=7687):
        self.host = host
        self.port = port
        self.sock = None
        self.claude = anthropic.Anthropic()

    def connect(self):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((self.host, self.port))
        self.reader = self.sock.makefile('r')

    def close(self):
        if self.sock:
            self.sock.close()

    def _send(self, request: dict) -> dict:
        data = json.dumps(request) + "\n"
        self.sock.sendall(data.encode())
        response = self.reader.readline()
        return json.loads(response)

    def extract_subgraph(self, center: int, hops: int = 2,
                          max_nodes: int = 50, format: str = "structured") -> dict:
        """Extract a subgraph centered on a node."""
        resp = self._send({
            "type": "ExtractSubgraph",
            "center": center,
            "hops": hops,
            "max_nodes": max_nodes,
            "format": format
        })
        if resp.get("status") == "error":
            raise RuntimeError(resp.get("message"))
        return resp.get("data", {})

    def ask_claude(self, question: str, context: str,
                    model: str = "claude-sonnet-4-20250514") -> str:
        """Send a question with graph context to Claude."""
        message = self.claude.messages.create(
            model=model,
            max_tokens=2048,
            system=f"""You are a helpful assistant analyzing a knowledge graph.
Use the following graph context to answer questions:

{context}""",
            messages=[
                {"role": "user", "content": question}
            ]
        )
        return message.content[0].text

    def graphrag_query(self, question: str, anchor: int,
                        hops: int = 2, max_nodes: int = 50) -> str:
        """Full GraphRAG pipeline: extract subgraph + ask Claude."""
        # 1. Extract subgraph
        subgraph = self.extract_subgraph(anchor, hops, max_nodes)
        context = subgraph.get("text", "")

        # 2. Ask Claude
        answer = self.ask_claude(question, context)

        return answer


# Usage
if __name__ == "__main__":
    client = GraphRAGClient()
    client.connect()

    try:
        # Ask a question about node 1's neighborhood
        answer = client.graphrag_query(
            question="Who does Alice work with and what projects are they on?",
            anchor=1,  # Alice's node ID
            hops=2,
            max_nodes=50
        )
        print("Answer:", answer)
    finally:
        client.close()

Go API

The Go client provides native GraphRAG support via the unified client. All GraphRAG operations are routed through the JSON/TCP transport.

Extract Subgraph

import (
    "context"
    "fmt"
    "github.com/AstraeaDB/AstraeaDB-Official"
)

ctx := context.Background()
client := astraeadb.NewClient(astraeadb.WithAddress("127.0.0.1", 7687))
client.Connect(ctx)
defer client.Close()

// Extract a 2-hop subgraph around a node, linearized as prose
subgraph, err := client.ExtractSubgraph(ctx, aliceNodeID,
    astraeadb.WithHops(2),
    astraeadb.WithMaxNodes(50),
    astraeadb.WithFormat("prose"),
)
fmt.Printf("Context: %s\nNodes: %d, Edges: %d, ~%d tokens\n",
    subgraph.Text, subgraph.NodeCount, subgraph.EdgeCount,
    subgraph.EstimatedTokens)

Full GraphRAG Pipeline

// GraphRAG with explicit anchor node
rag, err := client.GraphRAG(ctx, "What is the relationship between Alice and the fraud ring?",
    astraeadb.WithAnchor(aliceNodeID),
    astraeadb.WithRAGHops(3),
    astraeadb.WithRAGMaxNodes(100),
    astraeadb.WithRAGFormat("structured"),
)
if err != nil {
    log.Fatal(err)
}

fmt.Printf("Anchor: node %d\n", rag.AnchorNodeID)
fmt.Printf("Context: %d nodes, %d edges (~%d tokens)\n",
    rag.NodesInContext, rag.EdgesInContext, rag.EstimatedTokens)
fmt.Println("Context for LLM:", rag.Context)

// GraphRAG with vector search (find anchor automatically)
rag2, err := client.GraphRAG(ctx, "Who is involved in money laundering?",
    astraeadb.WithQuestionEmbedding([]float32{0.1, 0.8, 0.3}),
    astraeadb.WithRAGHops(2),
)

Combining GraphRAG with Claude (using Go HTTP client)

import (
    "bytes"
    "encoding/json"
    "net/http"
)

// Step 1: Get graph context from AstraeaDB
rag, _ := client.GraphRAG(ctx, question,
    astraeadb.WithAnchor(anchorID),
    astraeadb.WithRAGHops(3),
    astraeadb.WithRAGFormat("structured"),
)

// Step 2: Send to Claude API
prompt := fmt.Sprintf("Graph context:\n%s\n\nQuestion: %s", rag.Context, question)
body, _ := json.Marshal(map[string]any{
    "model":      "claude-sonnet-4-5-20250929",
    "max_tokens": 1024,
    "messages":   []map[string]string{{"role": "user", "content": prompt}},
})
req, _ := http.NewRequest("POST", "https://api.anthropic.com/v1/messages", bytes.NewReader(body))
req.Header.Set("x-api-key", apiKey)
req.Header.Set("anthropic-version", "2023-06-01")
req.Header.Set("content-type", "application/json")

resp, _ := http.DefaultClient.Do(req)
// ... parse response ...

Java API

The Java client provides native GraphRAG support via the unified client. Temporal, semantic, and GraphRAG operations are routed through the JSON/TCP transport, while CRUD and traversal use gRPC when available.

Subgraph Extraction

import com.astraeadb.unified.UnifiedClient;
import com.astraeadb.model.*;
import com.astraeadb.options.*;

try (var client = UnifiedClient.builder()
        .host("127.0.0.1").build()) {
    client.connect();

    // Extract a 2-hop subgraph around a node
    SubgraphResult subgraph = client.extractSubgraph(aliceId,
        SubgraphOptions.builder()
            .hops(2)
            .maxNodes(50)
            .format("structured")
            .build());

    System.out.println("Context: " + subgraph.text());
    System.out.println("Nodes: " + subgraph.nodeCount());
    System.out.println("Tokens: " + subgraph.estimatedTokens());
}

Full GraphRAG Pipeline

// GraphRAG with anchor node
RagResult rag = client.graphRag(
    "What is the relationship between Alice and the fraud ring?",
    RagOptions.builder()
        .anchor(aliceId)
        .hops(3)
        .maxNodes(100)
        .format("structured")
        .build());

System.out.println("Context: " + rag.context());
System.out.println("Nodes in context: " + rag.nodesInContext());

// Without anchor (server uses vector search to find relevant node)
RagResult rag2 = client.graphRag(
    "Who is involved in money laundering?",
    RagOptions.builder()
        .hops(2)
        .build());

Combining GraphRAG with Claude (using Java HTTP client)

// 1. Get graph context from AstraeaDB
RagResult rag = client.graphRag(question,
    RagOptions.builder().anchor(nodeId).hops(2).build());

// 2. Send to Claude via Anthropic API
var httpClient = HttpClient.newHttpClient();
String body = """
    {"model": "claude-sonnet-4-5-20250929",
     "max_tokens": 1024,
     "messages": [{"role": "user",
       "content": "Context:\\n%s\\n\\nQuestion: %s"}]}
    """.formatted(rag.context(), question);

var request = HttpRequest.newBuilder()
    .uri(URI.create("https://api.anthropic.com/v1/messages"))
    .header("x-api-key", apiKey)
    .header("anthropic-version", "2023-06-01")
    .POST(HttpRequest.BodyPublishers.ofString(body))
    .build();

var response = httpClient.send(request,
    HttpResponse.BodyHandlers.ofString());

JSON Protocol

You can use GraphRAG directly via the TCP/JSON protocol.

GraphRag Request

{
  "type": "GraphRag",
  "question": "Who compromised the server?",
  "anchor": 1,                    // Optional: specific node ID
  "question_embedding": [0.1, ...], // Optional: for vector search
  "hops": 2,
  "max_nodes": 50,
  "format": "structured"
}

Response

{
  "status": "ok",
  "data": {
    "answer": "Based on the graph, the server 10.0.0.5 was compromised by...",
    "anchor_node_id": 1,
    "nodes_in_context": 12,
    "estimated_tokens": 1847
  }
}

Example: Security Analysis

Let's walk through using GraphRAG for cybersecurity investigation.

Scenario

You have a knowledge graph of network events: servers, users, processes, and their relationships. An alert fired and you want to understand the attack path.

Step 1: Build the Threat Graph

# Create nodes for entities
attacker = client.create_node(["User", "Suspicious"], {
    "username": "unknown_user",
    "first_seen": "2024-01-15T03:22:00Z"
})

server = client.create_node(["Server"], {
    "hostname": "prod-db-01",
    "ip": "10.0.0.5",
    "role": "database"
})

process = client.create_node(["Process"], {
    "name": "reverse_shell.py",
    "pid": 31337,
    "started": "2024-01-15T03:25:00Z"
})

# Create attack path edges
client.create_edge(attacker, server, "SSH_LOGIN", {"timestamp": "2024-01-15T03:22:00Z"})
client.create_edge(attacker, process, "SPAWNED")
client.create_edge(process, server, "RUNNING_ON")

Step 2: Ask Claude to Analyze

answer = client.graphrag_query(
    question="""Analyze this security incident:
    1. What was the attack vector?
    2. What systems were compromised?
    3. What actions did the attacker take?
    4. Recommend immediate response steps.""",
    anchor=attacker,  # Start from the suspicious user
    hops=3            # Follow the full attack path
)

print(answer)

Claude's Response

## Security Incident Analysis

**Attack Vector:** SSH brute-force or credential theft
The attacker "unknown_user" gained initial access via SSH login to prod-db-01
(10.0.0.5) at 03:22 UTC.

**Compromised Systems:**
- prod-db-01 (10.0.0.5) - Production database server

**Attacker Actions:**
1. SSH login at 03:22 UTC
2. Spawned malicious process "reverse_shell.py" (PID 31337) at 03:25 UTC
3. Established persistence via reverse shell

**Immediate Response:**
1. Isolate prod-db-01 from network immediately
2. Kill process 31337 and check for persistence mechanisms
3. Reset all credentials that had access to this server
4. Review SSH logs for brute-force patterns
5. Check for lateral movement to other systems

Example: Research Assistant

Use GraphRAG to build a research assistant that understands relationships between papers, authors, and concepts.

Build the Research Graph

# Papers with embeddings from abstract
paper1 = client.create_node(["Paper"], {
    "title": "Attention Is All You Need",
    "year": 2017,
    "citations": 50000
}, embedding=embed("transformer attention mechanism..."))

paper2 = client.create_node(["Paper"], {
    "title": "BERT: Pre-training of Deep Bidirectional Transformers",
    "year": 2018
}, embedding=embed("bidirectional language model..."))

# Authors
vaswani = client.create_node(["Author"], {"name": "Ashish Vaswani"})

# Concepts
attention = client.create_node(["Concept"], {"name": "Self-Attention"})

# Relationships
client.create_edge(paper1, vaswani, "AUTHORED_BY")
client.create_edge(paper1, attention, "INTRODUCES")
client.create_edge(paper2, paper1, "CITES")
client.create_edge(paper2, attention, "USES")

Ask Research Questions

answer = client.graphrag_query(
    question="What papers introduced the self-attention mechanism and what subsequent work built on it?",
    anchor=attention_node_id,
    hops=2
)

# Claude responds with graph-aware understanding of paper lineage

Best Practices

1. Choose the Right Anchor

2. Tune Hop Depth

• 1 hop — Immediate neighbors only. Fast but limited context.
• 2 hops — Good default. Captures most relevant relationships.
• 3+ hops — Broader context but may include irrelevant nodes.

3. Match Format to Task

4. Use System Prompts

Guide Claude's behavior with domain-specific system prompts:

system_prompt = """You are a security analyst examining network threat graphs.
Focus on:
- Attack patterns and TTPs
- Lateral movement paths
- Indicators of compromise
Always cite specific nodes and relationships from the context."""

5. Handle Large Graphs

• Set appropriate max_nodes to avoid overwhelming context
• Use token_budget to ensure responses fit
• Consider multiple queries with different anchors for complex investigations

6. Validate and Iterate

Check the nodes_in_context and estimated_tokens in responses to understand what Claude saw. If answers seem incomplete, try:

• Different anchor nodes
• More hops
• Different linearization formats