Streaming-KG is a real-time Knowledge Graph pipeline that ingests unstructured data (like news articles), extracts structured knowledge using LLMs, and stores it in a graph database (Neo4j). The system enables advanced search, reasoning, and visualization over continuously updated knowledge.
RAG is a modern approach to building AI systems that answer questions using both a knowledge base and a language model. Streaming-KG implements RAG as follows:
- Where:
app/database/neo4j_client.py - Function:
search_graph(query_text) - What it does: Instead of asking the LLM to guess, you query your Neo4j Graph Database. You look for nodes that match the user's keyword and "traverse" the graph to find connected neighbors (context).
- The Code:
# This is the "Retrieval" part cypher_query = """ MATCH (n:Entity) WHERE ... CONTAINS ... MATCH path = (n)-[*1..2]-(m) ... RETURN ... """
- Where:
app/main.py - Function:
/searchendpoint - What it does: You take the user's raw question (
q) and combine it with the data found in step 1 (context). You are "augmenting" the prompt with real facts. - The Code:
# 1. Get facts from DB context, graph_visuals = neo4j_client.search_graph(clean_query) # 2. Pass BOTH the question AND the facts to the LLM answer = generate_answer(q, context)
- Where:
app/llm/extractor.py - Function:
generate_answer(query, context) - What it does: The LLM receives a prompt that says: "Here is some context I found in the database. Use ONLY this context to answer the user's question." It generates the final natural language sentence.
- The Code:
prompt = f""" Context from the Graph: {context_str} User Question: {query} ... """ client.chat.completions.create(...) # Generates the answer
- Backend: Python 3.10+, FastAPI
- Streaming: Apache Kafka, Zookeeper
- Database: Neo4j (Graph Database)
- AI/LLM: Ollama (Local Llama 3) or HuggingFace
- Infrastructure: Docker & Docker Compose
- Automated Knowledge Extraction: Converts raw text into structured, queryable knowledge.
- Real-Time Updates: Ingests and processes data streams (e.g., news, research) as they arrive.
- Graph Representation: Uses Neo4j to model entities and relationships, enabling rich queries and visualizations.
- AI Integration: Leverages LLMs (e.g., Llama, Zephyr) for entity/relation extraction and natural language answers.
- Scalable & Modular: Built with Kafka for streaming, FastAPI for APIs, and modular Python code for easy extension.
[User/API/CSV] → [Kafka Producer] → [Kafka Topic] → [Kafka Consumer]
| |
| ↓
| [LLM Extraction]
| ↓
| [Neo4j Graph DB]
| ↓
| [FastAPI Search API]
| ↓
| [Frontend/UI]
- Ingestion:
- Data (news, research, etc.) is sent via API or CSV to a Kafka topic.
- Streaming:
- Kafka decouples producers and consumers, enabling scalable, real-time pipelines.
- Extraction:
- Kafka consumer reads messages and uses an LLM to extract entities and relationships.
- Graph Storage:
- Extracted knowledge is upserted into Neo4j as nodes (with dynamic labels) and relationships.
- Search & Reasoning:
- FastAPI provides endpoints for search and Q&A over the graph.
- Visualization:
- Frontend (or Neo4j Browser) visualizes the evolving knowledge graph.
- Docker Desktop (Required for Kafka & Neo4j)
- Ollama (Required for local LLM inference)
- Python 3.10+
git clone <repo-url>
cd streaming-kgCreate a virtual environment and install dependencies:
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
pip install -r requirements.txtCreate a .env file in the root directory:
# Kafka
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
KAFKA_TOPIC=news-stream
# Neo4j
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=password123
# LLM (Ollama)
OLLAMA_BASE_URL=http://localhost:11434/v1
MODEL_ID=llama3Run Docker Compose to spin up Kafka, Zookeeper, and Neo4j:
docker-compose up -d- Neo4j UI: http://localhost:7474 (User: neo4j, Pass: password123)
- Kafka UI: http://localhost:8080
Ensure Ollama is running and pull the model:
ollama pull llama3You need to run two terminals.
Terminal 1: The Worker (Consumer) This listens to Kafka and processes data.
python -m app.kafka.consumerTerminal 2: The API (Server) This handles user requests and data ingestion.
uvicorn app.main:app --reloadSend a POST request to inject a news item.
curl -X POST "http://127.0.0.1:8000/ingest" \
-H "Content-Type: application/json" \
-d '{"text": "SpaceX launched the Starship rocket from Texas yesterday."}'Check Terminal 1: You should see the extraction logs and Neo4j update.
Ask a natural language question.
# Open in Browser or use Curl
http://127.0.0.1:8000/search?q=What did SpaceX launch?Response: "SpaceX launched the Starship rocket from Texas."
streaming-kg/
├── docker-compose.yml # Kafka & Neo4j Services
├── requirements.txt # Python Dependencies
├── .env # Environment Variables
├── app/
│ ├── main.py # FastAPI Entry Point (Ingest & Search)
│ ├── kafka/
│ │ ├── consumer.py # The "Worker" (Kafka -> LLM -> Neo4j)
│ │ └── producer.py # (Optional) Bulk CSV Loader
│ ├── llm/
│ │ ├── extractor.py # LLM Logic (Ollama/OpenAI)
│ │ └── prompts.py # System Prompts for JSON Extraction
│ └── database/
│ └── neo4j_client.py # Graph Logic (Cypher Queries)
└── data/ # Sample CSVs
- Automates knowledge extraction and structuring from unstructured sources.
- Enables advanced search, reasoning, and analytics over dynamic data.
- Supports real-time updates and scalable ingestion.
- Flexible for many domains: news, research, finance, etc.
- Add more LLMs and prompt templates for better extraction.
- Integrate with more data sources (APIs, RSS, etc.).
- Enhance the frontend for interactive graph exploration.
- Add user authentication and access control.
- Support for temporal and event-based reasoning.
- Deploy as a cloud-native microservice.
- Fork the Project
- Create your Feature Branch (
git checkout -b feature/AmazingFeature) - Commit your Changes (
git commit -m 'Add some AmazingFeature') - Push to the Branch (
git push origin feature/AmazingFeature) - Open a Pull Request
MIT