graphiti-core

import asyncio import os from datetime import datetime from graphiti_core import Graphiti from graphiti_core.nodes import EpisodeType # Ensure OPENAI_API_KEY and FALKORDB_URI are set in your environment # Example: export OPENAI_API_KEY="sk-..." # Example: docker run -p 6379:6379 -p 3000:3000 -it --rm falkordb/falkordb:latest async def main(): # Initialize Graphiti with FalkorDB driver (default to localhost) graphiti = Graphiti( uri=os.environ.get('FALKORDB_URI', "falkor://localhost:6379") ) # Build indices (run once during setup) print("Building indices and constraints...") await graphiti.build_indices_and_constraints() print("Indices built.") # Add an episode (information to be stored in the graph) episode_body = ( "Alice met Bob at the AI conference in San Francisco on March 15, 2024. " "They discussed the latest developments in graph databases and decided " "to collaborate on a new project using Graphiti and FalkorDB." ) print("Adding episode...") await graphiti.add_episode( name="Conference Meeting", episode_body=episode_body, episode_type=EpisodeType.text, reference_time=datetime(2024, 3, 15), source_description="Conference notes" ) print("Episode added.") # Search the knowledge graph print("Searching the graph...") search_results = await graphiti.search( query="What did Alice and Bob discuss?", num_results=3 ) print("\nSearch Results:") if search_results: for i, result in enumerate(search_results): print(f"- Result {i+1}: {result.node.summary}") else: print("No search results found.") # Close the connection await graphiti.close() print("Connection closed.") if __name__ == "__main__": # Set a dummy API key if not already set, for local testing without network if not os.environ.get('OPENAI_API_KEY'): print("Warning: OPENAI_API_KEY environment variable not set. Using a dummy key.") os.environ['OPENAI_API_KEY'] = 'dummy-key-for-test' asyncio.run(main())