Level Up Your RAG Workflow: Building a GraphRAG-Powered KYC Agent

As developers, we often face the challenge of making sense of messy, scattered data — especially when it lives across dozens of PDFs, reports, and databases. Traditional RAG systems are a great start, but they often fall short when it comes to complex queries that span multiple documents or require a deeper understanding of context.

That’s where GraphRAG steps in.

GraphRAG brings structure to chaos by building a knowledge graph from raw text and using it as the backbone for LLM-powered responses. It doesn’t just retrieve text chunks — it understands how entities relate, clusters them into meaningful communities, and provides a graph-structured lens into your data. GraphRAG combines the structured clarity of knowledge graphs with the generative power of large language models (LLMs) to bring meaning and connection to fragmented information.

Unlike conventional RAG systems that rely on flat semantic search, GraphRAG extracts key entities, maps relationships, and organizes information into intuitive, navigable hierarchies. This makes it especially valuable in domains like compliance, risk management, and enterprise intelligence, where answers often lie at the intersection of multiple documents and hidden connections. By turning unstructured data into a structured web of knowledge, GraphRAG delivers more accurate, contextual, and actionable results — empowering users to make better decisions, faster.

In this article, we will explore how to build a prototype Know-Your-Customer (KYC) agent using OpenAI’s Agent SDK, illustrating how GraphRAG can streamline investigations and surface hidden risk signals.

Getting Started

Table of contents

• How RAG works
• What is a Knowledge Graph
• What is GraphRAG
• How GraphRAG works
• Where traditional RAG falls short
• Vector RAG and Graph RAG
• Experimenting with GraphRAG: Build a KYC agent with OpenAI, MCP, Ollama, and Neo4j
• Prerequisites
• Installing the dependencies
• Setting up the credentials
• Synthetic dataset preparation
• Creating schemas
• Building KYC agent
• Importing required libraries
• Setting up the neo4j connection
• Building tools
• Tool 1: Get customer and accounts
• Tool 2: Find customer rings
• Tool 3: Neo4j MCP server toolset
• Tool 4: Generate Cypher
• Tool 5: Create memory
• Main function
• Agent reasoning and execution flow
• Executing the agent
• References

How RAG works

RAG enhances large language models by allowing them to fetch relevant information from an external knowledge source (like documents or databases) before generating a response.

Instead of relying solely on the LLM’s internal training data, RAG introduces a two-step pipeline:

1. Retrieval Phase

• The input query is used to search a vector database (e.g., FAISS, Pinecone, Chroma).
• This database contains embedded representations (vectors) of your documents or chunks of data.
• Using semantic similarity, the system retrieves the top-K most relevant text chunks for the query.

2. Generation Phase

• These retrieved documents are fed into the LLM along with the original query.
• The model then generates an answer that is grounded in the retrieved context, improving factual accuracy and relevance.

What is a Knowledge Graph

A knowledge graph is a structured and visual way of organizing information that captures real-world entities, their attributes, and the relationships between them. It helps model complex data in a way that makes hidden connections more obvious and queryable — ideal for domains like customer intelligence, fraud detection, or KYC (Know Your Customer).

Let’s break it down using the example from your customer data graph:

Key Components:

• Entities (Nodes): These are the main objects represented in the graph. In your schema, entities include:
• Customer, Account, Transaction, Device, Address, Company, Payment_Method, IP_Address
• Attributes: These are properties of each entity, like a Customer having a name, ID, or email, though they’re not directly shown in the graph view. Attributes are usually stored as metadata inside each node.
• Relationships (Edges): These define how entities are connected. For example:
• A Customer OWNS an Account
• A Customer USES_DEVICE (e.g., a phone or laptop)
• A Customer LIVES_AT an Address
• A Customer HAS_METHOD linked to a Payment_Method
• An Account is TO or FROM a Transaction
• A Device is ASSOCIATED WITH an IP_Address
• A Customer is EMPLOYED_BY a Company

It might look something like this:

With this structure, it becomes easy to run intelligent queries like:

• “Which accounts are associated with customers living at the same address?”
• “Show all transactions made using devices linked to the same IP address.”
• “Which customers share payment methods or devices?”

A knowledge graph like this turns scattered data into a connected web of insights, making it a foundational component for systems like GraphRAG, where reasoning over relationships is key.

What is GraphRAG

Graph-based Retrieval-Augmented Generation (GraphRAG) is an advanced AI technique that enhances traditional RAG systems by integrating knowledge graphs into the information retrieval and generation process. While conventional RAG pipelines retrieve top-matching text chunks from unstructured documents and pass them to a language model for generation, GraphRAG introduces structure and semantics to this flow. It does this by first transforming raw textual data into a knowledge graph, where key entities (like people, organizations, or events) and their relationships are identified and connected.

This graph-based representation allows for deeper reasoning and more context-aware generation, as the model can now leverage structured relationships, not just surface-level text similarity. For instance, instead of just finding documents that mention a customer, GraphRAG can traverse the graph to understand which accounts they own, which devices they’ve used, or what addresses they’re associated with — all before generating a response.

By combining the strengths of symbolic reasoning (via graphs) and generative AI (via LLMs), GraphRAG delivers more accurate, explainable, and scalable results. This makes it particularly useful in complex domains like fraud detection, compliance, legal discovery, and enterprise search, where understanding entity relationships is crucial for meaningful answers.

How GraphRAG works

GraphRAG operates in two main phases: Indexing (organizing information) and Querying (retrieving meaningful answers).

Indexing: Building the Knowledge Graph

• Chunking the Text: GraphRAG starts by breaking down your documents into smaller segments called TextUnits. These are manageable chunks that make analysis easier.
• Entity and Relationship Extraction: From each TextUnit, GraphRAG identifies key entities (like people, places, or organizations), claims, and how these elements are connected.
• Graph Construction: All this information is structured into a knowledge graph — a visual network of entities (nodes) and their relationships (edges). Important entities appear as larger nodes, and related ones are grouped into clusters.
• Community Summarization: For each cluster, GraphRAG generates a concise summary capturing the key themes and topics — offering a high-level view of your entire dataset.

Querying: Asking Questions and Getting Answers

Once the knowledge graph is built, GraphRAG uses it to answer questions in three ways:

• Global Search: Ideal for understanding overarching themes or trends. The system uses community-level summaries to provide broad insights.
• Local Search: Best for specific queries. If you ask about a person or company, GraphRAG explores their direct connections in the graph to provide a targeted, factual response.
• DRIFT Search: A hybrid approach that combines Local Search with community-level insights — useful when deeper, contextual understanding is needed.

Where traditional RAG falls short

Traditional Retrieval-Augmented Generation (RAG) excels at finding and returning semantically similar text snippets. But when data becomes more complex or scattered, traditional RAG begins to struggle — especially in real-world enterprise applications.

One major limitation is its inability to synthesize information spread across multiple sources. If a question requires connecting subtle, indirect relationships — such as tracing a customer’s activity across different devices, transactions, or locations — traditional RAG often fails to assemble those connections. It lacks an understanding of how different pieces of data relate within a larger context, leading to incomplete or inaccurate responses.

Another weakness lies in capturing broader context or summarizing nuanced datasets. Traditional RAG models aren’t built for higher-level semantic reasoning. So when faced with a query like “What are the main themes in the dataset?”, the system flounders — unless those themes are explicitly written out. That’s because this is not a simple retrieval problem; it’s a query-focused summarization task, which requires abstracting insights across the dataset — something traditional RAG isn’t inherently equipped to handle.

In short, while traditional RAG works well for localized lookups and direct Q&A, it falls short when dealing with interconnected, abstract, or multi-hop reasoning — the exact gaps that GraphRAG is designed to fill.

VectorRAG vs GraphRAG

While both VectorRAG and GraphRAG extend the power of language models by incorporating external knowledge, they approach the task very differently in how they retrieve, structure, and reason over information. VectorRAG excels at fast, precise look‑ups, while GraphRAG provides deeper, more explainable reasoning over structured relationships — ideal for complex investigative and analytical tasks.

Experimenting with GraphRAG: Build a KYC agent with OpenAI, MCP, Ollama, and Neo4j

Know-Your-Customer (KYC) processes involves navigating vast networks of entities — customers, accounts, devices, IPs, transactions — each intricately linked. Traditional RAG falls short in these scenarios where uncovering fraud demands tracing indirect, multi-hop relationships across data points. This is where GraphRAG shines. By grounding retrieval in a structured knowledge graph, it enables investigators to reason across complex connections and uncover hidden patterns — making it a powerful tool for detecting money laundering, sanctions violations, and other financial crimes.

In the next section, we will walk through how to build a prototype GraphRAG-powered KYC agent.

Prerequisites

This hands-on requires the following tools to be installed on your machine:

1. Ollama: Ollama is a platform for running large language models locally on your computer.

Download Ollama on Windows

To convert natural language questions into Cypher queries, we’ll use the Text-to-Cypher model provided by Neo4j.

Run the following command in your terminal to pull the model using Ollama:

ollama pull ed-neo4j/t2c-gemma3-4b-it-q8_0-35k

2. Neo4j: Neo4j is a graph database used to model and query relationships between entities like customers, accounts, transactions, and more.

You have two options to set up a free Neo4j database:

Option 1: Local Neo4j Docker Instance

• Download the Neo4j installer from the official site: https://neo4j.com/download
• Follow the installation instructions based on your operating system.
• Once installed, start the Neo4j Desktop application or run it via Docker to create and manage local databases.

Option 2: Neo4j AuraDB Free (Cloud-Based Managed Instance)

• Visit the Neo4j Aura Console: https://console.neo4j.io
• Sign in or create a Neo4j account.
• Click “Create Database” and select AuraDB Free.
• Once the database is created, download the connection credentials bundle. You’ll need these to connect to the database programmatically.

In this tutorial, we will proceed with Neo4j AuraDB Free as it is lightweight, cloud-hosted, and easily accessible from anywhere.

3. Python: Python is the core language used in this hands-on for scripting and backend logic.

Download Python

4. uv (Micro virtualenv manager): uv is a fast and modern Python project manager, to set up and manage our environment. It simplifies tasks like handling dependencies, creating virtual environments, and running scripts.

To install uv, run this in your terminal:

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
set Path=C:\Users\Codem\.local\bin;%Path%

Refer to the official website for detailed installation instructions.

Installation | uv

Installing the dependencies

• Initialize a uv project by executing the following command.

uv init kyc_agent
cd kyc_agent

• Create and activate a virtual environment by executing the following command.

uv venv
source .venv/bin/activate # for linux
.venv\Scripts\activate    # for windows

• Install neo4j, numpy, ollama, openai-agents, python-dotenv libraries using pip.

uv add neo4j neo4j-rust-ext numpy ollama openai-agents python-dotenv

Setting up the credentials

• Create a file named .env. This file will store your environment variables, including the OpenAI key and Neo4j credentials.
• Open the .env file and add the following code to specify your OpenAI API key and Neo4j credentials. Copy the Neo4j credentials from the file you downloaded while setting up the Neo4j AuraDB instance.

OPENAI_API_KEY=sk-proj-C1K1hKug99wXxtj...
NEO4J_URI=neo4j+s://b3383662.databases.neo4j.io
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=fa_BBW1s6kjvOjSLnTvkKht...
NEO4J_DATABASE=neo4j
AURA_INSTANCEID=b3383662
AURA_INSTANCENAME=Free instance

Synthetic dataset preparation

For the purpose of this blog, we have generated a synthetic dataset comprising 8,000 customers along with their associated accounts, transactions, registered addresses, devices, and IP addresses.

Dataset generation script

Use the following script to generate the dataset.

import numpy as np
import os
import random
import uuid
import time
from datetime import datetime, timedelta
from neo4j import GraphDatabase
from dotenv import load_dotenv
load_dotenv()  

random.seed(42)

NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687")
NEO4J_USER = os.getenv("NEO4J_USERNAME", "neo4j")
NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "password")
NEO4J_DATABASE = os.getenv("NEO4J_DATABASE", "neo4j")

driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))

def get_session():
    return driver.session(database=NEO4J_DATABASE)

# Create uniqueness constraints (once)
with get_session() as sess:
    
    for label in ('Customer','Account','Company','Address',
                  'Device','IP_Address','Payment_Method','Transaction'):
        sess.execute_write(
            lambda tx, L=label: tx.run(
                f"CREATE CONSTRAINT IF NOT EXISTS FOR (n:{L}) REQUIRE n.id IS UNIQUE"
            )
        )

# Configuration & ID lists
random.seed(42)
np.random.seed(42)

n_customers = 8_000
mean_accounts_per_customer   = 1.5
mean_devices_per_customer    = 2
mean_addresses_per_customer  = 1.2
mean_payment_methods_per_customer = 1
mean_transactions_per_account     = 10
p_pep       = 0.01
p_watchlist = 0.02

customers   = [f"CUST_{i:05d}" for i in range(1, n_customers+1)]
n_companies = int(n_customers * 0.2)
companies   = [f"COMP_{i:05d}" for i in range(1, n_companies+1)]


# Prepare payloads
customer_rows = [
    {"id": cust,
     "pep": (random.random() < p_pep),
     "wl":  (random.random() < p_watchlist),
     "name": cust
    }
    for cust in customers
]

company_rows = [
    {"id": comp,
     "ind": random.choice(['Finance','Tech','Manufacturing','Retail']),
     "name": comp
     }
    for comp in companies
]

# 3. Push Customers & Companies
print(f"loading start...")
start_time = time.perf_counter()
batch_size=50

with get_session() as sess:
    # Customers in implicit transactions of 50 rows each
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (c:Customer {id: row.id})
          SET c.is_pep       = row.pep,
              c.on_watchlist = row.wl,
              c.name = row.name
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=customer_rows,
        batch_size=batch_size
    )

    # Companies in implicit transactions of 50 rows each
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (c:Company {id: row.id})
          SET c.industry = row.ind,
              c.name = row.name 
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=company_rows,
        batch_size=batch_size
    )

    
end_time = time.perf_counter()
elapsed = end_time - start_time
print(f"⌛ Loading Customers & Companies took {elapsed:.2f} seconds")


# 3. Push Accounts, Addresses, Devices, IP addresses, Payment Methods and Transactions

# Build payloads
acct_counter = addr_counter = dev_counter = ip_counter = pm_counter = txn_counter = 0

account_rows     = []
employed_rows    = []
address_rows     = []
device_rows      = []
ip_rows          = []
payment_rows     = []
transaction_rows = []


# 1.1 Accounts & OWNS
for cust in customers:
    for _ in range(np.random.poisson(mean_accounts_per_customer)):
        acct_counter += 1
        aid = f"ACCT_{acct_counter:05d}"
        account_rows.append({"cust": cust, "acct": aid,"name":aid})

# 1.2 EMPLOYED_BY
for cust in customers:
    if random.random() < 0.8:
        comp = random.choice(companies)
        employed_rows.append({"cust": cust, "co": comp})

# 1.3 Addresses & LIVES_AT
for cust in customers:
    for _ in range(max(1, np.random.poisson(mean_addresses_per_customer))):
        addr_counter += 1
        aid = f"ADDR_{addr_counter:05d}"
        city = random.choice(['London','Manchester','Birmingham','Leeds'])
        address_rows.append({"cust": cust, "addr": aid, "city": city,"name":aid})

# 1.4 Devices & USES_DEVICE → ASSOCIATED_WITH IP_Address
for cust in customers:
    for _ in range(np.random.poisson(mean_devices_per_customer)):
        dev_counter += 1
        did = f"DEV_{dev_counter:05d}"
        osys = random.choice(['Android','iOS','Windows','MacOS'])
        device_rows.append({"cust": cust, "dev": did, "os": osys,"name":did})

        ip_counter += 1
        iid = f"IP_{ip_counter:05d}"
        ip_rows.append({"dev": did, "ip": iid,"name":iid})

# 1.5 Payment Methods & HAS_METHOD
for cust in customers:
    for _ in range(np.random.poisson(mean_payment_methods_per_customer)):
        pm_counter += 1
        pid = f"PM_{pm_counter:05d}"
        ptype = random.choice(['Credit_Card','Debit_Card','EWallet'])
        cnum = ''.join(random.choice('0123456789') for _ in range(16)) \
               if ptype in ('Credit_Card','Debit_Card') \
               else uuid.uuid4().hex[:16]
        payment_rows.append({
            "cust": cust,
            "pid": pid,
            "ptype": ptype,
            "cnum": cnum,
            "name": pid
        })


# 1.6 Transactions & FROM/TO
all_accts = [r["acct"] for r in account_rows]
for src in all_accts:
    for _ in range(np.random.poisson(mean_transactions_per_account)):
        txn_counter += 1
        tid = f"TXN_{txn_counter:06d}"
        amt = round(np.random.lognormal(mean=3, sigma=1), 2)
        ts  = (datetime(2025,1,1) + timedelta(days=random.randint(0,120))).isoformat()
        dst = random.choice(all_accts)
        transaction_rows.append({
            "src": src, "tid": tid, "amt": amt, "ts": ts, "dst": dst, "name":tid
        })


# 2. Push in batches
start_time = time.perf_counter()
with get_session() as sess:
    # 2.1 Accounts
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (a:Account {id: row.acct})
          SET a.name = row.name
          WITH a, row
          MATCH (c:Customer {id: row.cust})
          MERGE (c)-[:OWNS]->(a)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=account_rows, batch_size=batch_size
    )

    # 2.2 Employed
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MATCH (c:Customer {id: row.cust})
          MATCH (co:Company  {id: row.co})
          MERGE (c)-[:EMPLOYED_BY]->(co)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=employed_rows, batch_size=batch_size
    )

    # 2.3 Addresses
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (a:Address {id: row.addr})
          SET a.city = row.city,
              a.name = row.name
          WITH a, row
          MATCH (c:Customer {id: row.cust})
          MERGE (c)-[:LIVES_AT]->(a)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=address_rows, batch_size=batch_size
    )

    # 2.4 Devices
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (d:Device {id: row.dev})
          SET d.os = row.os,
            d.name = row.name
          WITH d, row
          MATCH (c:Customer {id: row.cust})
          MERGE (c)-[:USES_DEVICE]->(d)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=device_rows, batch_size=batch_size
    )
    # 2.5 IPs
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (i:IP_Address {id: row.ip})
          SET i.name = row.name
          WITH i, row
          MATCH (d:Device {id: row.dev})
          MERGE (d)-[:ASSOCIATED_WITH]->(i)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=ip_rows, batch_size=batch_size
    )

    # 2.6 Payment Methods
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (p:Payment_Method {id: row.pid})
          SET p.pm_type     = row.ptype,
              p.card_number = row.cnum,
              p.name = row.name
          WITH p, row
          MATCH (c:Customer {id: row.cust})
          MERGE (c)-[:HAS_METHOD]->(p)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=payment_rows, batch_size=batch_size
    )

    # 2.7 Transactions
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (t:Transaction {id: row.tid})
          SET t.amount    = row.amt,
              t.timestamp = row.ts,
              t.name = row.name
          WITH t, row
          MATCH (a1:Account {id: row.src})
          MATCH (a2:Account {id: row.dst})
          MERGE (a1)-[:FROM]->(t)-[:TO]->(a2)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=transaction_rows, batch_size=batch_size
    )

end_time = time.perf_counter()
elapsed = end_time - start_time
print(f"⌛ Loading Account, Employed, Owns, Addresses, Devices, Payment Methods & Transactions took {elapsed:.2f} seconds")


# 5. Select anomalies
n_anomalies        = int(0.05 * len(customers))
anoms             = random.sample(customers, n_anomalies)
chunk             = n_anomalies // 5

# Prepare payload lists
super_rows        = []
ring_acct_rows    = []
ring_txn_rows     = []
bridge_rows       = []
isolate_rows      = []
dense_addr_rows   = []
dense_pm_rows     = []


# Super-hubs: 50 new accounts per customer
for cust in anoms[0:chunk]:
    for _ in range(50):
        acct_counter += 1
        aid = f"ACCT_{acct_counter:05d}"
        super_rows.append({"cust": cust, "acct": aid,"name":aid})

#Upload
with get_session() as sess:
    # 4.1 Super-hubs
    start_time = time.perf_counter()
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (a:Account {id: row.acct})
          SET a.name = row.name
          WITH a, row
          MATCH (c:Customer {id: row.cust})
          MERGE (c)-[:OWNS]->(a)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=super_rows, batch_size=50
    )
    end_time = time.perf_counter()
    elapsed = end_time - start_time
    print(f"⌛ Loading Anomalies: Super Hubs took {elapsed:.2f} seconds")

# 2.2 Circular rings: 3-customer cycles
for i in range(chunk, 2*chunk, 3):
    trio = anoms[i : i+3]
    if len(trio) == 3:
        accts = []
        for c in trio:
            acct_counter += 1
            aid = f"ACCT_{acct_counter:05d}"
            ring_acct_rows.append({"cust": c, "acct": aid})
            accts.append(aid)
        for j in range(3):
            txn_counter += 1
            tid = f"TXN_{txn_counter:06d}"
            ring_txn_rows.append({
                "src":      accts[j],
                "dst":      accts[(j+1) % 3],
                "tid":      tid,
                "amount":   1000,
                "ts":       datetime(2025, 2, 1).isoformat()
            })

with get_session() as sess:
    # 4.2 Circular rings – ring transactions
    start_time = time.perf_counter()
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (t:Transaction {id: row.tid})
          SET t.amount = row.amount, t.timestamp = row.ts
          WITH t, row
          MATCH (a1:Account {id: row.src}), (a2:Account {id: row.dst})
          MERGE (a1)-[:FROM]->(t)-[:TO]->(a2)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=ring_txn_rows, batch_size=50
    )
    end_time = time.perf_counter()
    elapsed = end_time - start_time
    print(f"⌛ Loading Anomalies: Circular Rings took {elapsed:.2f} seconds")


# 2.3 Bridges: employed by two companies
for cust in anoms[2*chunk : 3*chunk]:
    c1, c2 = random.sample(companies, 2)
    bridge_rows.append({"cust": cust, "co": c1})
    bridge_rows.append({"cust": cust, "co": c2})
with get_session() as sess:
    # 4.3 Bridges
    start_time = time.perf_counter()
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MATCH (c:Customer {id: row.cust}), (co:Company {id: row.co})
          MERGE (c)-[:EMPLOYED_BY]->(co)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=bridge_rows, batch_size=50
    )
    end_time = time.perf_counter()
    elapsed = end_time - start_time
    print(f"⌛ Loading Anomalies: Bridges - Customers employeed by 2 companies took {elapsed:.2f} seconds")

#  Isolates: 5 device→IP pairs per customer, no link to customers
for cust in anoms[3*chunk : 4*chunk]:
    for _ in range(5):
        dev_counter += 1
        ip_counter  += 1
        isolate_rows.append({
            "dev": f"DEV_{dev_counter:05d}",
            "ip":  f"IP_{ip_counter:05d}"
        })

with get_session() as sess:
    # 4.4 Isolates
    start_time = time.perf_counter()
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MERGE (d:Device {id: row.dev})
          SET d.os = 'Unknown'
          MERGE (i:IP_Address {id: row.ip})
          MERGE (d)-[:ASSOCIATED_WITH]->(i)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=isolate_rows, batch_size=50
    )
    end_time = time.perf_counter()
    elapsed = end_time - start_time
    print(f"⌛ Loading Anomalies: Isolated Devices and IP Addresses with no Customers took {elapsed:.2f} seconds")

# Dense watchlist cluster: shared address & payment method
shared_addr = f"ADDR_{addr_counter+1:05d}"
shared_pm   = f"PM_{pm_counter+1:05d}"
dense_addr_rows = [{"cust": cust, "addr": shared_addr}
                   for cust in anoms[4*chunk : ]]
dense_pm_rows   = [{"cust": cust, "pm":   shared_pm}
                   for cust in anoms[4*chunk : ]]

# 3. Create the two shared nodes up front
with get_session() as sess:
    sess.run(
        "MERGE (a:Address {id:$addr}) SET a.city='London', a.name=$addr",
        addr=shared_addr
    )
    sess.run(
        "MERGE (p:Payment_Method {id:$pm}) SET p.pm_type='Credit_Card', p.name=$pm",
        pm=shared_pm
    )
with get_session() as sess:
    start_time = time.perf_counter()
    # 4.5 Dense cluster – shared address
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MATCH (c:Customer {id: row.cust}), (a:Address {id: row.addr})
          MERGE (c)-[:LIVES_AT]->(a)
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=dense_addr_rows, batch_size=50
    )
    # 4.5 Dense cluster – shared payment method + watchlist flag
    sess.run(
        """
        UNWIND $rows AS row
        CALL (row) {
          MATCH (c:Customer {id: row.cust}), (p:Payment_Method {id: row.pm})
          MERGE (c)-[:HAS_METHOD]->(p)
          SET c.on_watchlist = true
        } IN TRANSACTIONS OF $batch_size ROWS
        """,
        rows=dense_pm_rows, batch_size=50
    )
    end_time = time.perf_counter()
    elapsed = end_time - start_time
    print(f"⌛ Loading Anomalies: Dense clusters - Around shared address & payment method took {elapsed:.2f} seconds")

You can now navigate to the Neo4j Aura Console and click on the “Query” section. Connect to your current database instance, then execute the following query to view the schema:

CALL db.schema.visualization();

Creating schemas

Next, we will define schemas for each of the core entities in our graph model — Customer, Account, Transaction, etc.

# schemas.py
from pydantic import BaseModel
from typing import List, Optional

# Tool 1: Get Customer and Accounts
class CustomerAccountsInput(BaseModel):
    customer_id: str

class TransactionModel(BaseModel):
    id: Optional[str] = None
    amount: Optional[float] = None
    timestamp: Optional[str] = None

class AccountModel(BaseModel):
    id: str = None
    name: str = None
    transactions: List[TransactionModel] = []

class CustomerModel(BaseModel):
    id: Optional[str] = None
    name: Optional[str] = None
    on_watchlist: Optional[bool] = False
    is_pep: Optional[bool] = False

class CustomerAccountsOutput(BaseModel):
    customer: CustomerModel
    accounts: List[AccountModel]

# Tool 2: Identify watchlisted customers in suspicious rings
from typing import Dict, Any

class RingModel(BaseModel):
    ring_path: List[Dict[str, Any]]  # List of node dicts
    watched_customers: List[Dict[str, Any]]  # List of customer dicts
    watch_relationships: List[Dict[str, Any]]  # List of relationship dicts

class CustomerRingsInput(BaseModel):
    max_number_rings: int = 10
    customer_in_watchlist: Optional[bool] = True
    customer_is_pep: Optional[bool] = False

class CustomerRingsOutput(BaseModel):
    customer_rings: List[RingModel]


class GenerateCypherRequest(BaseModel):
    question: str
    database_schema: str

Building KYC agent

Let’s begin with the agent creation process.

Create a file named agent.py and add the following code to it.

Importing required libraries

import os
from agents import Agent, Runner, function_tool
from agents.mcp import MCPServerStdio
from neo4j import GraphDatabase
from schemas import CustomerAccountsInput, CustomerAccountsOutput, CustomerModel, AccountModel, TransactionModel, GenerateCypherRequest
import asyncio
from ollama import chat
from dotenv import load_dotenv
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logging.getLogger("httpx").setLevel(logging.ERROR)
logger = logging.getLogger("KYC_AGENT")

# Load environment variables
load_dotenv()

Setting up the neo4j connection

# Read Neo4j environment variables into variables
NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687")
NEO4J_USER = os.getenv("NEO4J_USERNAME", "neo4j")
NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "password")
NEO4J_DATABASE = os.getenv("NEO4J_DATABASE", "neo4j")

# Neo4j connection setup
def get_neo4j_driver():
    return GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))

# Neo4j driver
driver = get_neo4j_driver()

Building tools

An agent’s effectiveness depends on its tools. In this case, we are providing five key tools to the KYC agent, including optimized Cypher queries wrapped in Python functions using the @function_tool decorator from the OpenAI Agent SDK.

Tool 1: Get customer and accounts

This tool retrieves a customer’s profile, including their accounts and recent transactions — an essential part of any investigation. It uses a function that takes a customer ID and runs a simple Cypher query.

@function_tool
def get_customer_and_accounts(input: CustomerAccountsInput, tx_limit: int = 5) -> CustomerAccountsOutput:
    logger.info(f"TOOL: GET_CUSTOMER_AND_ACCOUNTS - {input.customer_id}")
    with driver.session() as session:
        result = session.run(
            """
            MATCH (c:Customer {id: $customer_id})-[o:OWNS]->(a:Account)
            WITH c, a
            CALL (c,a) {
                MATCH (a)-[b:TO|FROM]->(t:Transaction)
                ORDER BY t.timestamp DESC
                LIMIT $tx_limit
                RETURN collect(t) as transactions
            }
            RETURN c as customer, a as account, transactions
            """,
            customer_id=input.customer_id,
            tx_limit=tx_limit
        )
        # Get the records from the result
        records = result.data()
        # Initialize lists to store the customer, accounts, and transactions
        accounts = []
        for record in records:
            customer = dict(record["customer"])
            account = dict(record["account"])
            account["transactions"] = [dict(t) for t in record["transactions"]]
            accounts.append(account)

        return CustomerAccountsOutput(
            customer=CustomerModel(**customer),
            accounts=[AccountModel(**a) for a in accounts]
        )

Tool 2: Find customer rings

This tool detects circular transaction patterns — commonly linked to money laundering — by identifying cycles in the KYC graph where funds return to their origin. It uses a find_customer_rings function to run a Cypher query that returns up to 10 potential rings, including the involved customers, accounts, and transactions.

@function_tool 
def find_customer_rings(max_number_rings: int = 10, customer_in_watchlist: bool = True, customer_is_pep: bool = False, customer_id: str = None):
    logger.info(f"TOOL: FIND_CUSTOMER_RINGS - {max_number_rings} - {customer_in_watchlist} - {customer_is_pep}")
    with driver.session() as session:
        result = session.run(
            f"""
            MATCH p=(a:Account)-[:FROM|TO*6]->(a:Account)
            WITH p, [n IN nodes(p) WHERE n:Account] AS accounts
            UNWIND accounts AS acct
            MATCH (cust:Customer)-[r:OWNS]->(acct)
            WHERE cust.on_watchlist = $customer_in_watchlist AND cust.is_pep = $customer_is_pep
            WITH 
              p, 
              COLLECT(DISTINCT cust)   AS watchedCustomers,
              COLLECT(DISTINCT r)      AS watchRels
            RETURN 
              p, 
              watchedCustomers,
              watchRels
            LIMIT $max_number_rings
            """,
            max_number_rings=max_number_rings,
            customer_in_watchlist=customer_in_watchlist,
            customer_is_pep=customer_is_pep
        )
        rings = []
        for record in result:
            # Convert path to a list of node dictionaries for easier consumption
            path_nodes = [dict(node) for node in record["p"].nodes]
            watched_customers = [dict(cust) for cust in record["watchedCustomers"]]
            watch_rels = [dict(rel) for rel in record["watchRels"]]
            rings.append({
                "ring_path": path_nodes,
                "watched_customers": watched_customers,
                
            })
        
        return {"customer_rings": rings}

Tool 3: Neo4j MCP server toolset

This section outlines a common architecture for enabling agents to interact with a knowledge graph. It combines a Text-to-Cypher model (Gemma3–4B) with the Neo4 MCP Server to translate natural language into Cypher and execute dynamic queries. Key tools include get-neo4j-schema, read-neo4j-cypher, and write-neo4j-cypher, allowing the agent to understand the graph structure and perform read/write operations.

neo4j_mcp_server = MCPServerStdio(
    params={
        "command": "uvx",
        "args": ["mcp-neo4j-cypher@0.2.1"],
        "env": {
            "NEO4J_URI": NEO4J_URI,
            "NEO4J_USERNAME": NEO4J_USER,
            "NEO4J_PASSWORD": NEO4J_PASSWORD,
            "NEO4J_DATABASE": NEO4J_DATABASE,
        },
    },
    cache_tools_list=True,
    name="Neo4j MCP Server",
    client_session_timeout_seconds=20
)

Tool 4: Generate Cypher

Translating natural language into Cypher queries relies on schema-aware LLMs fine-tuned for this task. Open-source models like neo4j/text-to-cypher-Gemma-3-4B-Instruct-2025.04.0 from Hugging Face enable accurate query generation. For example, given a question about shared addresses, the agent can dynamically generate the appropriate Cypher query without failure.

@function_tool
def generate_cypher(request: GenerateCypherRequest) -> str:
    USER_INSTRUCTION = """Generate a Cypher query for the Question below:
    Use the information about the nodes, relationships, and properties from the Schema section below to generate the best possible Cypher query. 
    Return only the Cypher query as your final output, without any additional text or explanation.
    ####Schema:
    {schema}
    ####Question:
    {question}"""

    logger.info(f"TOOL: GENERATE_CYPHER - INPUT - {request.question}")
    user_message = USER_INSTRUCTION.format(
        schema=request.database_schema, 
        question=request.question
    )
    # Generate Cypher query using the text2cypher model
    model: str = "ed-neo4j/t2c-gemma3-4b-it-q8_0-35k"
    response = chat(
        model=model,
        messages=[{"role": "user", "content": user_message}]
    )
    generated_cypher = response['message']['content']
    # Replace \n with new line
    generated_cypher = generated_cypher.replace("\\n", "\n")

    print(f"GENERATED CYPHER: - OUTPUT - {generated_cypher}")
    
    return generated_cypher

Tool 5: Create memory

While agents handle short-term memory through conversation history, complex tasks like financial investigations require persistent long-term memory. This memory acts as a dynamic knowledge base, tracking insights and context across sessions. The create_memory tool enables this by storing investigation summaries as nodes linked to relevant entities in the knowledge graph.

@function_tool
def create_memory(content: str, customer_ids: list[str] = [], account_ids: list[str] = [], transaction_ids: list[str] = []) -> str:
    logger.info(f"TOOL: CREATE_MEMORY - {content} - {customer_ids} - {account_ids} - {transaction_ids}")
    with driver.session() as session:
        result = session.run(
            """
            CREATE (m:Memory {content: $content, created_at: datetime()})
            WITH m
            UNWIND $customer_ids as cid
            MATCH (c:Customer {id: cid})
            MERGE (m)-[:FOR_CUSTOMER]->(c)
            WITH m
            UNWIND $account_ids as aid
            MATCH (a:Account {id: aid})
            MERGE (m)-[:FOR_ACCOUNT]->(a)
            WITH m
            UNWIND $transaction_ids as tid
            MATCH (t:Transaction {id: tid})
            MERGE (m)-[:FOR_TRANSACTION]->(t)
            RETURN m.content as content
            """,
            content=content,
            customer_ids=customer_ids,
            account_ids=account_ids,
            transaction_ids=transaction_ids
        )
       
        return f"Created memory: {str(result)}"

Main function

The main() function sets up and runs an interactive KYC agent using the Neo4j MCP server and OpenAI's Agent SDK. It connects to the MCP server, defines the agent's instructions and tools, and maintains a conversation history for context. In a loop, it accepts user queries, passes them to the agent for processing, and displays the results. If needed, the agent dynamically generates and executes Cypher queries. The function also ensures proper cleanup of resources when the session ends.

async def main():
    await neo4j_mcp_server.connect()  # Connect the MCP server before using it

    # Define the instructions for the agent
    instructions = """You are a KYC analyst with access to a knowledge graph. Use the tools to answer questions about customers, accounts, and suspicious patterns.
    You are also a Neo4j expert and can use the Neo4j MCP server to query the graph.
    If you get a question about the KYC database that you can not answer with GraphRAG tools, you should
    - use the Neo4j MCP server to get the schema of the graph (if needed)
    - use the generate_cypher tool to generate a Cypher query from question and the schema
    - use the Neo4j MCP server to query the graph to answer the question
    """

    kyc_agent = Agent(
        name="KYC Analyst",
        instructions=instructions,
        tools=[get_customer_and_accounts, find_customer_rings, create_memory, generate_cypher],
        mcp_servers=[neo4j_mcp_server]
    )
    
    # Initialize conversation history
    conversation_history = []
    
    while True:
        query = input("Enter your KYC query (or 'quit' to exit): ")
        if query.lower() == 'quit':
            break
            
        # Run the agent with conversation history
        result = await Runner.run(
            kyc_agent, 
            conversation_history + [{"role": "user", "content": query}]
        )
        
        # Add the new interaction to conversation history
        conversation_history.extend([
            {"role": "user", "content": query},
            {"role": "assistant", "content": result.final_output}
        ])
        
        print(result.final_output)

    # Clean up
    await neo4j_mcp_server.cleanup()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    finally:
        # Clean up any remaining resources
        driver.close() 

Agent reasoning and execution flow

When the agent receives a user query, it follows a structured decision-making process to determine the appropriate tools to use:

1. Schema Discovery
   Example Query: “Get me the schema of the database.”
   Action: The agent recognizes a schema request and uses the neo4j-mcp-server.get-neo4j-schema tool to retrieve the graph schema.
2. Use of Custom GraphRAG Tool
   Example Query: “Show me 5 watchlisted customers involved in suspicious transaction rings.”
   Action: This aligns with a predefined tool. The agent calls find_customer_rings with the parameter customer_in_watchlist=True.
3. Dynamic Cypher Generation & Execution
   Example Query: “For each of these customers, find their addresses and check if they’re shared with others.”
   Action: Since no GraphRAG tool directly addresses this, the agent:
   Uses the previously fetched schema.
   Passes the question and schema to generate_cypher to generate a Cypher query.
   Executes the query via neo4j-mcp-server.read-neo4j-cypher.
4. Entity-Specific Data Retrieval
   Example Query: “Get details for the customer whose address is shared.”
   Action: The agent identifies this as a profile lookup and calls get_customer_and_accounts with the customer ID.
5. Memory Creation for Long-Term Context
   Example Query: “Write a 300-word summary of this investigation and link it to all related accounts and transactions.”
   Action: The agent generates the summary using its internal LLM and then stores it using create_memory, linking it to the relevant entities.

Executing the agent

The entire workflow is now fully integrated and operational. We can proceed to execute the agent and test it using the sample queries outlined above to observe its behavior and capabilities in handling real-world KYC scenarios.

python agent.py

Thanks for reading this article !!

Thanks Gowri M Bhatt for reviewing the content.

If you enjoyed this article, please click on the clap button 👏 and share to help others find it!

The full source code for this tutorial can be found here,

GitHub - codemaker2015/kyc-agent: A smart, tool‑augmented AI agent for Know Your Customer (KYC)…

References

• https://youtu.be/bpiphxrYn9I
• GraphRAG in Action: A Simple Agent for Know-Your-Customer Investigations | Towards Data Science
• The Ultimate MCP Handbook: From Basics to Advanced LLM Integration | Medium
• OpenAI Agents SDK
• GraphRAG Developer Guide — Developer Guides
• Google Cloud & Neo4j: Teaming Up at the Intersection of Knowledge Graphs, Agents, MCP, and Natural Language Interfaces