import asyncio
import os
from dotenv import load_dotenv
from smartgraph.components import CompletionComponent
from smartgraph.tools.duck_memory_toolkit import DuckMemoryToolkit
from smartgraph.logging import SmartGraphLogger
# Load environment variables
load_dotenv()
# Set up logging
logger = SmartGraphLogger.get_logger()
logger.set_level("INFO")
async def main():
# Initialize DuckMemoryToolkit
memory_toolkit = DuckMemoryToolkit("assistant_memory.duckdb")
# Create a CompletionComponent with DuckMemoryToolkit
completion = CompletionComponent(
"AI_Completion",
model="claude-3-haiku-20240307",
temperature=0.7,
api_key=os.getenv("ANTHROPIC_API_KEY"),
toolkits=[memory_toolkit],
)
# Add some initial knowledge
await memory_toolkit.add_memory("python", {"language": "Python", "type": "Programming language", "creator": "Guido van Rossum"})
user_input = "Tell me about Python and its creator"
logger.info(f"Processing input: {user_input}")
try:
result = await completion.process({"message": user_input})
if "error" in result:
logger.error(f"Error during processing: {result['error']}")
print(f"\nAn error occurred: {result['error']}\n")
else:
logger.info(f"Received: {result}")
print(f"\nAI Response: {result.get('ai_response', 'No response')}\n")
# Store the interaction in memory
await memory_toolkit.add_memory(f"interaction_{user_input}", {"question": user_input, "answer": result.get('ai_response', 'No response')})
except Exception as e:
logger.error(f"Unexpected error during processing: {str(e)}", exc_info=True)
print(f"\nAn unexpected error occurred: {str(e)}\n")
if __name__ == "__main__":
asyncio.run(main())