How to Build an AI RAG Workflow using FalkorDB + N8N + Graphiti Agent Module

				
					import asyncio
import json
from datetime import datetime
from graphiti_core import Graphiti
from graphiti_core.driver.falkordb_driver import FalkorDriver

openapi_file = 'openapi.json'
with open(openapi_file, 'r') as f:
    openapi = json.load(f)

async def clear_database(driver):
    """Delete and recreate the documentation database"""
    print("Clearing database...")
    try:
        # Delete the graph if it exists
        await driver.client.execute_command('GRAPH.DELETE', 'documentation')
        print("✓ Database cleared")
    except Exception as e:
        print(f"⚠ Could not delete existing database: {e}")

async def create_indices(graphiti):
    """Build Graphiti indices and constraints"""
    print("Building indices...")
    try:
        await graphiti.build_indices_and_constraints()
        print("✓ Indices created successfully")
    except Exception as e:
        print(f"⚠ Index creation skipped: {e}")

async def extract_knowledge():
    # Initialize FalkorDB driver
    driver = FalkorDriver(
        host='localhost',
        port=6379,
        database='documentation'
    )
    
    # Clear database first
    await clear_database(driver)
    
    # Initialize Graphiti after cleanup
    graphiti = Graphiti(graph_driver=driver)
    
    # Create indices
    await create_indices(graphiti)
    
    print("\\nAdding episodes to FalkorDB...")
    episode_count = 0
    
    for path, path_item in openapi.get('paths', {}).items():
        for method, details in path_item.items():
            if method.startswith('x-'):
                continue
            
            summary = details.get('summary', '')
            description = details.get('description', '')
            
            # Build episode text with all details
            params_text = ""
            for param in details.get('parameters', []):
                param_type = param.get('schema', {}).get('type', 'string')
                required = "required" if param.get('required', False) else "optional"
                param_desc = param.get('description', '')
                params_text += f"\\n- {param['name']} ({param_type}, {required}): {param_desc}"
            
            body_text = ""
            if 'requestBody' in details:
                request_body = details['requestBody']
                content = request_body.get('content', {})
                for content_type, content_obj in content.items():
                    schema = content_obj.get('schema', {})
                    if schema.get('type') == 'object':
                        for prop_name, prop_schema in schema.get('properties', {}).items():
                            prop_type = prop_schema.get('type', 'string')
                            required = "required" if prop_name in schema.get('required', []) else "optional"
                            body_text += f"\\n- {prop_name} ({prop_type}, {required})"
            
            errors_text = ""
            for status_code, response in details.get('responses', {}).items():
                if status_code == 'default':
                    continue
                errors_text += f"\\n- {status_code}: {response.get('description', '')}"
            
            episode_body = f"""
API Endpoint: {method.upper()} {path}
Summary: {summary}
Description: {description}

Path & Query Parameters: {params_text}

Request Body Fields: {body_text}

Response Codes: {errors_text}
"""
            
            try:
                await graphiti.add_episode(
                    name=f"{method.upper()} {path}",
                    episode_body=episode_body,
                    reference_time=datetime.now(),
                    source_description="OpenAPI specification"
                )
                print(f"✓ {method.upper()} {path}")
                episode_count += 1
            except Exception as e:
                print(f"✗ {method.upper()} {path}: {e}")
    
    print(f"\\n✓ Extraction complete!")
    print(f"  - Database cleared and reinitialized")
    print(f"  - {episode_count} API endpoints extracted")
    print(f"  - Indices created for optimal search")
    print(f"  - Knowledge graph ready for N8N queries")
    
    await graphiti.close()

if __name__ == "__main__":
    asyncio.run(extract_knowledge())