In this hands-on guide, I will walk you through building a powerful AI agent that combines Neo4j knowledge graphs with large language models to enable structured, verifiable reasoning. By the end of this tutorial, you will have a working system that can query complex relationships, extract entities, and generate contextually accurate responses backed by your own data.
Why Combine Knowledge Graphs with LLMs?
Large language models are impressive generalists, but they suffer from a critical limitation: hallucination. When you need factual, auditable answers about your specific domain, pure LLM responses can be unreliable. Knowledge graphs solve this by providing a structured layer of truth that the LLM must respect.
According to recent benchmarks, agents with knowledge graph retrieval achieve 73% fewer factual errors compared to vanilla RAG approaches. The key insight is that graphs naturally represent relationships—the way humans actually think about data.
Understanding the Architecture
Our system consists of three core components working in concert:
- Neo4j Database — Stores entities and relationships as nodes and edges
- HolySheep AI API — Provides LLM capabilities with structured output support
- Agent Orchestration Layer — Coordinates queries, reasoning, and response generation
HolySheep AI offers significant advantages here: their unified API platform provides access to multiple models including GPT-4.1 at $8/MTok and DeepSeek V3.2 at just $0.42/MTok, with sub-50ms latency. Their pricing at ¥1=$1 exchange rate saves you 85%+ compared to domestic alternatives charging ¥7.3 per dollar.
Prerequisites and Environment Setup
Installing Neo4j Desktop
For beginners, the easiest way to start is Neo4j Desktop. Download it from neo4j.com and follow the installation wizard. Once installed, create a new database by clicking "New Graph" and selecting "Blank Graph." Set a password you will remember—you will need it for connections.
Screenshot hint: Neo4j Desktop main screen showing "New Graph" button in the left sidebar
After creating your database, click "Start" to get it running. Note the connection URI (usually bolt://localhost:7687) and your password.
Setting Up Python Environment
Create a new Python project and install the necessary libraries:
pip install neo4j langchain-openai python-dotenv requestsCreate a .env file to store your credentials:
# .env file
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=your_password_here
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEYPopulating Your Knowledge Graph
Before querying, we need data. Let me show you how to create a simple company knowledge graph step by step.
Creating Nodes and Relationships
The following script creates a knowledge graph about a fictional tech company with employees, projects, and skills:
from neo4j import GraphDatabase
from dotenv import load_dotenv
import os
load_dotenv()
class KnowledgeGraphSetup:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def create_company_graph(self):
"""Creates a sample company knowledge graph with employees, projects, and skills."""
with self.driver.session() as session:
# Clear existing data
session.run("MATCH (n) DETACH DELETE n")
# Create employees as nodes
employees = [
{"id": "E001", "name": "Alice Chen", "role": "Senior Engineer", "department": "Engineering"},
{"id": "E002", "name": "Bob Smith", "role": "Product Manager", "department": "Product"},
{"id": "E003", "name": "Carol Wang", "role": "Data Scientist", "department": "Data"},
{"id": "E004", "name": "David Kim", "role": "DevOps Engineer", "department": "Engineering"}
]
for emp in employees:
session.run("""
MERGE (e:Employee {id: $id})
SET e.name = $name, e.role = $role, e.department = $department
""", id=emp["id"], name=emp["name"], role=emp["role"], department=emp["department"])
# Create projects as nodes
projects = [
{"id": "P001", "name": "Atlas Platform", "status": "Active", "budget": 500000},
{"id": "P002", "name": "Mercury API", "status": "Completed", "budget": 120000},
{"id": "P003", "name": "Nexus Analytics", "status": "Planning", "budget": 300000}
]
for proj in projects:
session.run("""
MERGE (p:Project {id: $id})
SET p.name = $name, p.status = $status, p.budget = $budget
""", id=proj["id"], name=proj["name"], status=proj["status"], budget=proj["budget"])
# Create skills as nodes
skills = ["Python", "Machine Learning", "Kubernetes", "SQL", "GraphQL"]
for skill in skills:
session.run("MERGE (s:Skill {name: $name})", name=skill)
# Create relationships: Employee WORKS_ON Project
session.run("""
MATCH (e:Employee {id: 'E001'})
MATCH (p:Project {id: 'P001'})
MERGE (e)-[:WORKS_ON {role: 'Tech Lead'}]->(p)
""")
session.run("""
MATCH (e:Employee {id: 'E003'})
MATCH (p:Project {id: 'P003'})
MERGE (e)-[:WORKS_ON {role: 'Lead Data Scientist'}]->(p)
""")
# Create relationships: Employee HAS_SKILL
session.run("MATCH (e:Employee {id: 'E001'}), (s:Skill {name: 'Python'}) MERGE (e)-[:HAS_SKILL]->(s)")
session.run("MATCH (e:Employee {id: 'E001'}), (s:Skill {name: 'Kubernetes'}) MERGE (e)-[:HAS_SKILL]->(s)")
session.run("MATCH (e:Employee {id: 'E003'}), (s:Skill {name: 'Machine Learning'}) MERGE (e)-[:HAS_SKILL]->(s)")
session.run("MATCH (e:Employee {id: 'E003'}), (s:Skill {name: 'Python'}) MERGE (e)-[:HAS_SKILL]->(s)")
# Create relationships: Project BELONGS_TO Department
session.run("""
MATCH (p:Project), (e:Employee)-[:WORKS_ON]->(p)
WITH p, e.department as dept
LIMIT 1
MERGE (p)-[:BELONGS_TO]->(:Department {name: dept})
""")
return "Knowledge graph created successfully with 4 employees, 3 projects, and 5 skills"
def verify_graph(self):
"""Verify the graph structure by counting nodes and relationships."""
with self.driver.session() as session:
result = session.run("""
MATCH (n)
RETURN labels(n)[0] as type, count(n) as count
""")
print("Node counts by type:")
for record in result:
print(f" {record['type']}: {record['count']}")
rel_result = session.run("""
MATCH ()-[r]->()
RETURN type(r) as rel_type, count(r) as count
""")
print("\nRelationship counts by type:")
for record in rel_result:
print(f" {record['rel_type']}: {record['count']}")
Execute the setup
kg = KnowledgeGraphSetup(
uri=os.getenv("NEO4J_URI"),
user=os.getenv("NEO4J_USER"),
password=os.getenv("NEO4J_PASSWORD")
)
print(kg.create_company_graph())
kg.verify_graph()
kg.close()Run this script and you should see output confirming your graph was created. The script uses Cypher queries—Neo4j's query language—which are designed to read naturally like sentences describing relationships.
Building the LLM-Powered Agent
Now we connect our knowledge graph to HolySheep AI's LLM API for intelligent querying. The HolySheep platform supports structured output, which is essential for parsing query results into clean formats.
The HolySheep API Integration
import requests
import json
from dotenv import load_dotenv
import os
load_dotenv()
class HolySheepLLM:
"""Wrapper for HolySheep AI API with structured output support."""
def __init__(self, api_key, base_url="https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
def generate_structured(self, prompt, system_prompt=None, schema=None):
"""
Generate response using HolySheep AI with structured output.
Args:
prompt: User's question
system_prompt: System instructions for the model
schema: JSON schema for structured output (optional)
Returns:
Response text or structured data
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
payload = {
"model": "deepseek-v3.2",
"messages": messages,
"temperature": 0.3,
"max_tokens": 1000
}
if schema:
payload["response_format"] = {"type": "json_object", "schema": schema}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
result = response.json()
return result["choices"][0]["message"]["content"]
def query_with_context(self, question, context_data):
"""
Ask a question with retrieved context from the knowledge graph.
Uses structured prompting for accurate responses.
"""
system_prompt = """You are a helpful assistant that answers questions based on provided context.
If the answer is in the context, provide it. If not, say you don't know.
Always cite specific entities and relationships when available."""
prompt = f"""Context from knowledge graph:
{json.dumps(context_data, indent=2)}
Question: {question}
Answer the question based on the context above. Include specific names, relationships, and numbers when relevant."""
return self.generate_structured(prompt, system_prompt)
Initialize the LLM
llm = HolySheepLLM(api_key=os.getenv("HOLYSHEEP_API_KEY"))
print("HolySheep AI connection established successfully!")Querying the Knowledge Graph
The following class combines our knowledge graph with the LLM to create a powerful querying agent:
from neo4j import GraphDatabase
import json
class KnowledgeGraphAgent:
"""
An AI agent that queries Neo4j knowledge graphs and generates
natural language responses using HolySheep AI.
"""
def __init__(self, neo4j_uri, neo4j_user, neo4j_password, llm_client):
self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, password))
self.llm = llm_client
def close(self):
self.driver.close()
def extract_entities_from_question(self, question):
"""Use LLM to identify what entities and relationships the question asks about."""
schema = {
"type": "object",
"properties": {
"entities": {
"type": "array",
"items": {"type": "string"},
"description": "Entity types to look for (Employee, Project, Skill, Department)"
},
"relationships": {
"type": "array",
"items": {"type": "string"},
"description": "Relationships to traverse (WORKS_ON, HAS_SKILL, BELONGS_TO)"
},
"filters": {
"type": "object",
"description": "Any filter conditions mentioned (status, department, etc.)"
}
}
}
extraction_prompt = f"""Analyze this question and extract the knowledge graph components needed to answer it.
Question: {question}
Return a JSON object with:
- entities: What node types to look for
- relationships: What edges to traverse
- filters: Any conditions mentioned
Only include fields that are relevant to the question."""
result = self.llm.generate_structured(extraction_prompt, schema=schema)
return json.loads(result)
def build_cypher_query(self, entities, relationships, filters=None):
"""Automatically generate Cypher query from extracted components."""
# Simple query builder for demonstration
# In production, use more sophisticated query generation
if "Employee" in entities and "Project" in entities:
if "WORKS_ON" in relationships:
base_query = """
MATCH (e:Employee)-[r:WORKS_ON]->(p:Project)
"""
if filters:
conditions = []
if "status" in filters:
conditions.append(f"p.status = '{filters['status']}'")
if "department" in filters:
conditions.append(f"e.department = '{filters['department']}'")
if conditions:
base_query += " WHERE " + " AND ".join(conditions)
base_query += " RETURN e.name as employee, e.role as role, p.name as project, p.status as status, p.budget as budget"
return base_query
# Default query for employee information
if "Employee" in entities:
query = "MATCH (e:Employee) RETURN e.id as id, e.name as name, e.role as role, e.department as department"
if filters and "department" in filters:
query = f"MATCH (e:Employee {{department: '{filters['department']}'}}) RETURN e.id as id, e.name as name, e.role as role"
return query
return "MATCH (n) RETURN n LIMIT 25"
def execute_query(self, cypher_query):
"""Execute a Cypher query and return results."""
with self.driver.session() as session:
result = session.run(cypher_query)
return [dict(record) for record in result]
def ask(self, question):
"""
Main agent method: Takes a natural language question and returns an answer.
"""
print(f"Processing question: {question}")
# Step 1: Extract knowledge graph components
components = self.extract_entities_from_question(question)
print(f"Extracted components: {components}")
# Step 2: Build and execute Cypher query
cypher = self.build_cypher_query(
components.get("entities", []),
components.get("relationships", []),
components.get("filters")
)
print(f"Generated Cypher: {cypher}")
context = self.execute_query(cypher)
print(f"Retrieved {len(context)} results from knowledge graph")
# Step 3: Generate natural language response
answer = self.llm.query_with_context(question, context)
return {
"question": question,
"answer": answer,
"supporting_data": context,
"cypher_query": cypher
}
Initialize the agent
agent = KnowledgeGraphAgent(
neo4j_uri=os.getenv("NEO4J_URI"),
neo4j_user=os.getenv("NEO4J_USER"),
neo4j_password=os.getenv("NEO4J_PASSWORD"),
llm_client=llm
)
Test the agent with sample questions
questions = [
"Who is working on the Atlas Platform?",
"What skills does Alice Chen have?",
"Which employees have Python skills?"
]
for q in questions:
result = agent.ask(q)
print(f"\n{'='*60}")
print(f"Question: {result['question']}")
print(f"Answer: {result['answer']}")
print(f"Data: {result['supporting_data']}")Adding Structured Reasoning with Chain of Thought
For more complex queries, we can implement chain-of-thought reasoning to make the agent's logic transparent and auditable:
class ReasoningAgent:
"""
Advanced agent with structured reasoning capabilities.
Implements chain-of-thought prompting for transparent decision-making.
"""
def __init__(self, kg_agent):
self.kg_agent = kg_agent
def decompose_question(self, question):
"""Break down complex questions into simpler sub-questions."""
decomposition_prompt = f"""Break down this complex question into simpler sub-questions that can be answered by querying a knowledge graph.
Question: {question}
Return a JSON array of sub-questions, each focusing on one aspect of the main question."""
schema = {
"type": "object",
"properties": {
"sub_questions": {
"type": "array",
"items": {"type": "string"}
},
"reasoning_steps": {
"type": "array",
"items": {"type": "string"}
}
}
}
result = self.kg_agent.llm.generate_structured(decomposition_prompt, schema=schema)
return json.loads(result)
def answer_with_reasoning(self, question):
"""
Answer the question with full reasoning trace.
Returns both the answer and the reasoning steps.
"""
print(f"Analyzing complex question: {question}")
# Step 1: Decompose into sub-questions
decomposition = self.decompose_question(question)
sub_questions = decomposition.get("sub_questions", [question])
reasoning_steps = decomposition.get("reasoning_steps", [])
# Step 2: Execute each sub-question
sub_answers = []
for sq in sub_questions:
result = self.kg_agent.ask(sq)
sub_answers.append({
"sub_question": sq,
"answer": result["answer"],
"data": result["supporting_data"]
})
# Step 3: Synthesize final answer with reasoning trace
synthesis_prompt = f"""Based on the following sub-question answers, synthesize a comprehensive response to the original question.
Original Question: {question}
Sub-questions and Answers:
{json.dumps(sub_answers, indent=2)}
Provide a final answer that directly addresses the original question, citing specific data from the sub-answers."""
final_answer = self.kg_agent.llm.generate_structured(synthesis_prompt)
return {
"original_question": question,
"reasoning_steps": reasoning_steps,
"sub_questions": sub_answers,
"final_answer": final_answer
}
Test the reasoning agent
reasoning_agent = ReasoningAgent(agent)
complex_question = "What is the relationship between the Atlas Platform project and our team's Python skills?"
result = reasoning_agent.answer_with_reasoning(complex_question)
print("\n" + "="*70)
print("COMPLEX QUERY WITH REASONING TRACE")
print("="*70)
print(f"\nQuestion: {result['original_question']}")
print(f"\nReasoning Steps:")
for i, step in enumerate(result['reasoning_steps'], 1):
print(f" {i}. {step}")
print(f"\nSub-question Results:")
for sq in result['sub_questions']:
print(f" Q: {sq['sub_question']}")
print(f" A: {sq['answer']}\n")
print(f"Final Answer: {result['final_answer']}")Advanced: Dynamic Knowledge Graph Updates
A truly useful agent should also be able to update the knowledge graph when it encounters new information:
class DynamicKnowledgeAgent:
"""
Agent that can both read from and write to the knowledge graph.
Enables learning from new information.
"""
def __init__(self, kg_agent):
self.kg_agent = kg_agent
def extract_cypher_writes(self, text):
"""Extract knowledge graph operations from natural language."""
schema = {
"type": "object",
"properties": {
"operations": {
"type": "array",
"items": {
"type": "object",
"properties": {
"operation": {"type": "string", "enum": ["create_node", "create_relationship", "update_node"]},
"node_type": {"type": "string"},
"properties": {"type": "object"},
"source_node": {"type": "string"},
"target_node": {"type": "string"},
"relationship_type": {"type": "string"}
}
}
}
}
}
prompt = f"""Analyze this text and extract any knowledge graph operations needed to store the information.
Text: {text}
For each piece of information, determine:
- Should a new node be created?
- Should relationships be established?
- What properties should be stored?
Return an array of operations to execute."""
result = self.kg_agent.llm.generate_structured(prompt, schema=schema)
return json.loads(result)
def execute_operations(self, operations):
"""Execute write operations on the knowledge graph."""
results = []
with self.kg_agent.driver.session() as session:
for op in operations:
if op["operation"] == "create_node":
node_type = op["node_type"]
props = op["properties"]
query = f"""
MERGE (n:{node_type} {{id: $id}})
SET n += $props
RETURN n
"""
result = session.run(query, id=props.get("id", props.get("name")), props=props)
results.append({"success": True, "operation": "created_node", "type": node_type})
elif op["operation"] == "create_relationship":
query = f"""
MATCH (s {{name: $source}})
MATCH (t {{name: $target}})
MERGE (s)-[r:{op['relationship_type']}]->(t)
RETURN s, t
"""
session.run(query,
source=op["source_node"],
target=op["target_node"])
results.append({
"success": True,
"operation": "created_relationship",
"from": op["source_node"],
"to": op["target_node"],
"type": op["relationship_type"]
})
return results
def learn_from_text(self, text):
"""Learn new information and update the knowledge graph."""
print(f"Learning from text: {text[:100]}...")
operations = self.extract_cypher_writes(text)
results = self.execute_operations(operations.get("operations", []))
return {
"text": text,
"extracted_operations": operations.get("operations", []),
"executed_results": results
}
Test learning capability
dynamic_agent = DynamicKnowledgeAgent(agent)
new_info = "Sarah Johnson joined the Engineering department as a Junior Developer and will work on the Mercury API project. She has strong skills in Python and SQL."
result = dynamic_agent.learn_from_text(new_info)
print("Learning result:", result)
Query to verify the update
update_result = agent.ask("What do we know about Sarah Johnson?")
print(f"\nVerification query result: {update_result['answer']}")Deployment Considerations
When moving to production, consider these optimizations:
- Caching — Cache frequent query patterns to reduce API calls and improve latency
- Connection pooling — Use connection pools for Neo4j to handle concurrent requests
- Rate limiting — Implement client-side rate limiting to respect API quotas
- Structured logging — Log all queries and responses for debugging and compliance
HolySheep AI's infrastructure provides built-in rate limiting at 1000 requests per minute on standard plans, with sub-50ms p99 latency for most regions. Their support for multiple model families means you can optimize costs by using cheaper models (DeepSeek V3.2 at $0.42/MTok) for simple queries while reserving GPT-4.1 ($8/MTok) for complex reasoning tasks.
Common Errors and Fixes
Error 1: Connection Refused to Neo4j
Error message: neo4j.exceptions.ServiceUnavailable: Connection refused
Cause: Neo4j database is not running or credentials are incorrect.
# Fix: Ensure Neo4j is running and credentials match
1. Check Neo4j Desktop is started (green indicator)
2. Verify your .env file matches your actual password
3. Test connection with Neo4j Browser at bolt://localhost:7687
If using Neo4j Aura (cloud), update URI:
NEO4J_URI=neo4j+s://your-instance.databases.neo4j.io # Aura connection string
Also update credentials:
NEO4J_USER=neo4j
NEO4J_PASSWORD=your_actual_bolt_passwordError 2: HolySheep API Authentication Failure
Error message: 401 Client Error: Unauthorized
Cause: Invalid or missing API key.
# Fix: Verify your API key is correct and properly set
import os
from dotenv import load_dotenv
load_dotenv()
Check if key is loaded correctly
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
print("ERROR: Please set your HolySheep API key in .env file")
print("Get your key from: https://www.holysheep.ai/register")
else:
print(f"API key loaded: {api_key[:8]}...") # Show first 8 chars for verification
If key expired, generate new one from dashboard
HolySheep keys can be regenerated from the API Keys section
Error 3: Cypher Query Syntax Errors
Error message: neo4j.exceptions.CypherSyntaxError: Invalid syntax
Cause: Malformed Cypher query, often from dynamic query building.
# Fix: Always validate Cypher queries before execution
Use Neo4j Browser's autocomplete and error highlighting
Common mistakes and fixes:
1. Missing quotes in string comparison
Wrong: WHERE e.department = Engineering
Correct: WHERE e.department = 'Engineering'
2. Using equals instead of colons in labels
Wrong: MATCH (e:Employee {department: 'Data'})
Correct: MATCH (e:Employee {department: 'Data'})
3. Relationship direction issues
If query returns empty, try reversing direction:
MATCH (p)<-[:WORKS_ON]-(e) vs MATCH (p)-[:WORKS_ON]->(e)
Add query validation:
def safe_query(session, query, params=None):
try:
result = session.run(query, params or {})
return list(result)
except Exception as e:
print(f"Query error: {e}")
print(f"Query was: {query}")
return []Error 4: LLM Structured Output Parsing Failure
Error message: JSONDecodeError: Expecting value
Cause: Model output doesn't match expected JSON schema.
# Fix: Add error handling and fallback for parsing
import json
import re
def safe_parse_json(text):
"""Safely parse JSON with fallback to regex extraction."""
try:
return json.loads(text)
except json.JSONDecodeError:
# Try to extract JSON from markdown code blocks
match = re.search(r'``(?:json)?\s*([\s\S]*?)\s*``', text)
if match:
try:
return json.loads(match.group(1))
except:
pass
# Fallback: Try to find JSON-like object
match = re.search(r'\{[\s\S]*\}', text)
if match:
try:
return json.loads(match.group(0))
except:
pass
return {"error": "Could not parse response", "raw": text}
Use with LLM responses:
result = llm.generate_structured(prompt)
parsed = safe_parse_json(result)
print(f"Parsed result: {parsed}")Error 5: Missing Relationship Patterns
Error message: Empty results even though data exists
Cause: Query pattern doesn't match graph structure.
# Fix: Use graph exploration queries to understand structure
First, explore what relationships exist:
explore_query = """
MATCH (a)-[r]->(b)
WITH labels(a)[0] as source_type, type(r) as rel_type, labels(b)[0] as target_type
RETURN source_type, collect(DISTINCT rel_type) as relationships, collect(DISTINCT target_type) as targets
"""
Or find all node types and their properties:
node_types_query = """
CALL db.labels() YIELD label
MATCH (n:{label})
RETURN label, keys(n) as properties, count(n) as count
"""
Use these to debug your query patterns
Also helpful: Neo4j Browser's :schema command shows the graph schema
Conclusion and Next Steps
In this comprehensive guide, I walked you through building an AI agent that combines Neo4j knowledge graphs with HolySheep AI's LLM capabilities. We covered everything from initial setup to advanced features like chain-of-thought reasoning and dynamic knowledge updates.
The key takeaways are: knowledge graphs provide auditable, structured data that grounds LLM responses in facts; the agent architecture enables natural language queries over complex relationship data; and HolySheep AI's multi-model support lets you optimize for both cost and capability depending on query complexity.
I encourage you to experiment with different graph schemas for your specific domain—whether that's product catalogs, organizational structures, or scientific ontologies. The combination of graph traversal and LLM reasoning opens up powerful possibilities for building trustworthy AI systems.
To get started with zero upfront cost, HolySheep AI offers free credits on registration, with support for WeChat and Alipay payments for convenience. Their pricing model at ¥1=$1 exchange rate represents significant savings compared to alternatives charging ¥7.3 per dollar.
👉 Sign up for HolySheep AI — free credits on registration