File size: 2,109 Bytes
fe36046 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 |
"""
Updated LangGraph Agent Implementation
Implements the architecture from the system diagram with memory layer, agent routing, and verification.
"""
import os
import sys
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Import the new agent system
from src import run_agent_system, memory_manager
from src.tracing import flush_langfuse, shutdown_langfuse
def run_agent(query: str) -> str:
"""
Main entry point for the agent system.
Args:
query: The user question
Returns:
The formatted final answer
"""
try:
# Run the new agent system
result = run_agent_system(
query=query,
user_id=os.getenv("USER_ID", "default_user"),
session_id=os.getenv("SESSION_ID", "default_session")
)
# Flush tracing events
flush_langfuse()
return result
except Exception as e:
print(f"Agent Error: {e}")
return f"I apologize, but I encountered an error: {e}"
def clear_memory():
"""Clear the agent's session memory"""
memory_manager.clear_session_cache()
print("Agent memory cleared")
def cleanup():
"""Cleanup function for graceful shutdown"""
try:
flush_langfuse()
shutdown_langfuse()
memory_manager.close_checkpointer()
print("Agent cleanup completed")
except Exception as e:
print(f"Cleanup error: {e}")
if __name__ == "__main__":
# Test the agent system
test_queries = [
"What is the capital of France?",
"Calculate the factorial of 5",
"What are the benefits of renewable energy?"
]
print("Testing new LangGraph Agent System")
print("=" * 50)
for i, query in enumerate(test_queries, 1):
print(f"\nTest {i}: {query}")
print("-" * 30)
try:
result = run_agent(query)
print(f"Result: {result}")
except Exception as e:
print(f"Error: {e}")
# Cleanup
cleanup()
print("\nAll tests completed!") |