Update veryfinal.py

veryfinal.py

@@ -26,8 +26,8 @@ from langchain_text_splitters import RecursiveCharacterTextSplitter
 
 # Agno imports
 from agno.agent import Agent
-from agno.models.groq import Groq
-from agno.models.google import Gemini
+from agno.models.groq import GroqChat
+from agno.models.google import GeminiChat
 from agno.tools.tavily import TavilyTools
 from agno.memory.agent import AgentMemory
 from agno.storage.sqlite import SqliteStorage
@@ -144,19 +144,19 @@ def subtract(a: int, b: int) -> int:
 
 @tool
 def divide(a: int, b: int) -> float:
-    """Divide two numbers."""
+    """Divide two numbers; errors if divisor is zero."""
     if b == 0:
         raise ValueError("Cannot divide by zero.")
     return a / b
 
 @tool
 def modulus(a: int, b: int) -> int:
-    """Get the remainder of division."""
+    """Return the remainder of a divided by b."""
     return a % b
 
 @tool
 def optimized_web_search(query: str) -> str:
-    """Optimized Tavily web search."""
+    """Optimized Tavily web search for speed."""
     try:
         time.sleep(random.uniform(1, 2))
         docs = TavilySearchResults(max_results=2).invoke(query=query)
@@ -169,7 +169,7 @@ def optimized_web_search(query: str) -> str:
 
 @tool
 def optimized_wiki_search(query: str) -> str:
-    """Optimized Wikipedia search."""
+    """Optimized Wikipedia search for speed."""
     try:
         time.sleep(random.uniform(0.5, 1))
         docs = WikipediaLoader(query=query, load_max_docs=1).load()
@@ -184,10 +184,7 @@ def optimized_wiki_search(query: str) -> str:
 def setup_faiss():
     try:
         schema = """
-        {
-          page_content: .Question,
-          metadata: { task_id: .task_id, Final_answer: ."Final answer" }
-        }
+        { page_content: .Question, metadata: { task_id: .task_id, Final_answer: ."Final answer" } }
         """
         loader = JSONLoader(file_path="metadata.jsonl", jq_schema=schema, json_lines=True, text_content=False)
         docs = loader.load()
@@ -202,6 +199,7 @@ def setup_faiss():
         print(f"FAISS setup failed: {e}")
         return None
 
+# State definition
 class EnhancedAgentState(TypedDict):
     messages: Annotated[List[HumanMessage|AIMessage], operator.add]
     query: str
@@ -210,6 +208,7 @@ class EnhancedAgentState(TypedDict):
     perf: Dict[str,Any]
     agno_resp: str
 
+# Hybrid system combining LangGraph and Agno
 class HybridLangGraphAgnoSystem:
     def __init__(self):
         self.agno = create_agno_agents()
@@ -234,98 +233,87 @@ class HybridLangGraphAgnoSystem:
 
         def router(st: EnhancedAgentState) -> EnhancedAgentState:
             q = st["query"].lower()
-            if any(k in q for k in ["calculate","math"]):
-                t = "lg_math"
-            elif any(k in q for k in ["research","analyze"]):
-                t = "agno_research"
-            elif any(k in q for k in ["what is","who is"]):
-                t = "lg_retrieval"
-            else:
-                t = "agno_general"
+            if any(k in q for k in ["calculate","math"]):          t="lg_math"
+            elif any(k in q for k in ["research","analyze"]):      t="agno_research"
+            elif any(k in q for k in ["what is","who is"]):        t="lg_retrieval"
+            else:                                                  t="agno_general"
             return {**st, "agent_type": t}
 
         def lg_math(st: EnhancedAgentState) -> EnhancedAgentState:
             groq_limiter.wait_if_needed()
-            t0 = time.time()
-            llm = groq_llm.bind_tools([multiply, add, subtract, divide, modulus])
-            sys = SystemMessage(content="Fast calculator. FINAL ANSWER: [result]")
-            res = llm.invoke([sys, HumanMessage(content=st["query"])])
-            return {**st, "final_answer": res.content, "perf": {"time": time.time()-t0, "prov":"LG-Groq"}}
+            t0=time.time()
+            llm=groq_llm.bind_tools([multiply,add,subtract,divide,modulus])
+            sys=SystemMessage(content="Fast calculator. FINAL ANSWER: [result]")
+            res=llm.invoke([sys,HumanMessage(content=st["query"])])
+            return {**st, "final_answer":res.content, "perf":{"time":time.time()-t0,"prov":"LG-Groq"}}
 
         def agno_research(st: EnhancedAgentState) -> EnhancedAgentState:
             gemini_limiter.wait_if_needed()
-            t0 = time.time()
-            resp = self.agno["research"].run(st["query"], stream=False)
-            return {**st, "final_answer": resp, "perf": {"time": time.time()-t0, "prov":"Agno-Gemini"}}
+            t0=time.time()
+            resp=self.agno["research"].run(st["query"],stream=False)
+            return {**st, "final_answer":resp, "perf":{"time":time.time()-t0,"prov":"Agno-Gemini"}}
 
         def lg_retrieval(st: EnhancedAgentState) -> EnhancedAgentState:
             groq_limiter.wait_if_needed()
-            t0 = time.time()
-            llm = groq_llm.bind_tools(self.tools)
-            sys = SystemMessage(content="Retrieve. FINAL ANSWER: [answer]")
-            res = llm.invoke([sys, HumanMessage(content=st["query"])])
-            return {**st, "final_answer": res.content, "perf": {"time": time.time()-t0, "prov":"LG-Retrieval"}}
+            t0=time.time()
+            llm=groq_llm.bind_tools(self.tools)
+            sys=SystemMessage(content="Retrieve. FINAL ANSWER: [answer]")
+            res=llm.invoke([sys,HumanMessage(content=st["query"])])
+            return {**st, "final_answer":res.content, "perf":{"time":time.time()-t0,"prov":"LG-Retrieval"}}
 
         def agno_general(st: EnhancedAgentState) -> EnhancedAgentState:
             nvidia_limiter.wait_if_needed()
-            t0 = time.time()
+            t0=time.time()
             if any(k in st["query"].lower() for k in ["calculate","compute"]):
-                resp = self.agno["math"].run(st["query"], stream=False)
+                resp=self.agno["math"].run(st["query"],stream=False)
             else:
-                resp = self.agno["research"].run(st["query"], stream=False)
-            return {**st, "final_answer": resp, "perf": {"time": time.time()-t0, "prov":"Agno-General"}}
+                resp=self.agno["research"].run(st["query"],stream=False)
+            return {**st, "final_answer":resp, "perf":{"time":time.time()-t0,"prov":"Agno-General"}}
 
         def pick(st: EnhancedAgentState) -> str:
             return st["agent_type"]
 
-        g = StateGraph(EnhancedAgentState)
-        g.add_node("router", router)
-        g.add_node("lg_math", lg_math)
-        g.add_node("agno_research", agno_research)
-        g.add_node("lg_retrieval", lg_retrieval)
-        g.add_node("agno_general", agno_general)
+        g=StateGraph(EnhancedAgentState)
+        g.add_node("router",router)
+        g.add_node("lg_math",lg_math)
+        g.add_node("agno_research",agno_research)
+        g.add_node("lg_retrieval",lg_retrieval)
+        g.add_node("agno_general",agno_general)
         g.set_entry_point("router")
-        g.add_conditional_edges("router", pick, {
-            "lg_math":"lg_math",
-            "agno_research":"agno_research",
-            "lg_retrieval":"lg_retrieval",
-            "agno_general":"agno_general"
+        g.add_conditional_edges("router",pick,{
+            "lg_math":"lg_math","agno_research":"agno_research",
+            "lg_retrieval":"lg_retrieval","agno_general":"agno_general"
         })
         for n in ["lg_math","agno_research","lg_retrieval","agno_general"]:
-            g.add_edge(n, "END")
+            g.add_edge(n,"END")
         return g.compile(checkpointer=MemorySaver())
 
     def process_query(self, q: str) -> Dict[str,Any]:
-        state = {
+        state={
             "messages":[HumanMessage(content=q)],
-            "query":q, "agent_type":"", "final_answer":"", "perf":{}, "agno_resp":""
+            "query":q,"agent_type":"","final_answer":"",
+            "perf":{},"agno_resp":""
         }
-        cfg = {"configurable":{"thread_id":f"hyb_{hash(q)}"}}
+        cfg={"configurable":{"thread_id":f"hyb_{hash(q)}"}}
         try:
-            out = self.graph.invoke(state, cfg)
+            out=self.graph.invoke(state,cfg)
             return {
-                "answer": out["final_answer"],
-                "performance_metrics": out["perf"],
-                "provider_used": out["perf"].get("prov")
+                "answer":out["final_answer"],
+                "performance_metrics":out["perf"],
+                "provider_used":out["perf"].get("prov")
             }
         except Exception as e:
-            return {"answer":f"Error: {e}", "performance_metrics":{}, "provider_used":"Error"}
+            return {"answer":f"Error: {e}","performance_metrics":{},"provider_used":"Error"}
 
-def build_graph(provider: str = "hybrid"):
-    if provider == "hybrid":
+def build_graph(provider: str="hybrid"):
+    if provider=="hybrid":
         return HybridLangGraphAgnoSystem().graph
-    elif provider == "groq":
-        # return a Groq-only graph
-    elif provider == "google":
-        # return a Google-only graph
-    # etc.
-    else:
-        raise ValueError(f"Unsupported provider: {provider}")
-
-
-if __name__ == "__main__":
-    graph = build_graph()
-    msgs = [HumanMessage(content="What are the names of the US presidents who were assassinated?")]
-    res = graph.invoke({"messages":msgs},{"configurable":{"thread_id":"test"}})
+    raise ValueError("Only 'hybrid' supported")
+
+# Test
+if __name__=="__main__":
+    graph=build_graph()
+    msgs=[HumanMessage(content="What are the names of the US presidents who were assassinated?")]
+    res=graph.invoke({"messages":msgs},{"configurable":{"thread_id":"test"}})
     for m in res["messages"]:
         m.pretty_print()