Knowledge Graph from Text

What You'll Build

Building knowledge graphs from unstructured documents. By the end of this tutorial, you'll have a fully working implementation you can extend for production use.

Project Overview

Input → Processing Layer → AI Model → Output
  ↓           ↓               ↓         ↓
Validate   Transform       Inference  Format

Step 1: Project Setup

bash
Create project directory
mkdir knowledge-graph-from-text
cd knowledge-graph-from-text
Create virtual environment
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
Install dependencies
pip install openai anthropic python fastapi uvicorn python-dotenv pydantic
Create .env file
cat > .env << 'EOF'
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
EOF

Step 2: Core Implementation

python
main.py
import os
from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel
from typing import Optional
load_dotenv()
client = OpenAI()
class KnowledgeGraphFromTextConfig(BaseModel):
    model: str = "gpt-4o-mini"
    temperature: float = 0.3
    max_tokens: int = 2000
    system_prompt: str = "You are a helpful AI assistant specializing in knowledge-graph."
class KnowledgeGraphFromText:
    """Knowledge Graph from Text implementation.
    
    Building knowledge graphs from unstructured documents
    """
    
    def __init__(self, config: Optional[KnowledgeGraphFromTextConfig] = None):
        self.config = config or KnowledgeGraphFromTextConfig()
        self.client = client
        self.history = []
    
    def process(self, user_input: str) -> str:
        """Main processing pipeline for knowledge-graph."""
        
        # Add to conversation history
        self.history.append({"role": "user", "content": user_input})
        
        # Build messages
        messages = [
            {"role": "system", "content": self.config.system_prompt}
        ] + self.history[-10:]  # Keep last 10 turns
        
        # Call AI
        response = self.client.chat.completions.create(
            model=self.config.model,
            messages=messages,
            temperature=self.config.temperature,
            max_tokens=self.config.max_tokens
        )
        
        assistant_msg = response.choices[0].message.content
        
        # Add assistant response to history
        self.history.append({"role": "assistant", "content": assistant_msg})
        
        return assistant_msg
    
    def reset(self):
        """Clear conversation history."""
        self.history = []
        print("Conversation history cleared.")

Step 3: API Service

python
api.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
app = FastAPI(title="Knowledge Graph from Text API")
Initialize the system
system = KnowledgeGraphFromText()
class ProcessRequest(BaseModel):
    input: str
    session_id: str = "default"
class ProcessResponse(BaseModel):
    output: str
    session_id: str
@app.get("/health")
async def health():
    return {"status": "ok", "service": "Knowledge Graph from Text"}
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Step 4: CLI Interface

python
cli.py
import sys
def main():
    """Interactive CLI for Knowledge Graph from Text."""
    print(f"Knowledge Graph from Text")
    print(f"Type 'quit' to exit, 'reset' to clear history\n")
    
    system = KnowledgeGraphFromText()
    
    while True:
        try:
            user_input = input("You: ").strip()
            
            if not user_input:
                continue
            
            if user_input.lower() == 'quit':
                print("Goodbye!")
                sys.exit(0)
            
            if user_input.lower() == 'reset':
                system.reset()
                continue
            
            response = system.process(user_input)
            print(f"AI: {response}\n")
            
        except KeyboardInterrupt:
            print("\nGoodbye!")
            sys.exit(0)
if __name__ == "__main__":
    main()

Step 5: Testing

python
test_main.py
import pytest
from main import KnowledgeGraphFromText
@pytest.fixture
def system():
    return KnowledgeGraphFromText()
def test_basic_processing(system):
    result = system.process("Hello, please help me with knowledge-graph")
    assert isinstance(result, str)
    assert len(result) > 10
def test_history_tracking(system):
    system.process("First message")
    system.process("Second message")
    assert len(system.history) == 4  # 2 user + 2 assistant
def test_history_reset(system):
    system.process("Some message")
    system.reset()
    assert len(system.history) == 0

Running the Project

bash
Run API server
uvicorn api:app --reload
Test API
curl -X POST http://localhost:8000/process \
  -H "Content-Type: application/json" \
  -d '{"input": "Test knowledge-graph", "session_id": "test"}'
Run CLI
python cli.py
Run tests
pytest test_main.py -v

What's Next

Resources

OpenAI API: https://platform.openai.com/docs

FastAPI: https://fastapi.tiangolo.com

python documentation: https://github.com/python/python