introduce LangGraph-based QA-workflow with CodeAgent answers generation

answering.py

@@ -0,0 +1,17 @@
+import os
+
+from smolagents import CodeAgent, InferenceClientModel, FinalAnswerTool, DuckDuckGoSearchTool, WikipediaSearchTool, PythonInterpreterTool
+
+from dotenv import load_dotenv
+load_dotenv()
+
+
+def gen_question_answer(question: str) -> str:
+    duck_duck_go_search_tool = DuckDuckGoSearchTool()
+    wikipedia_search_tool = WikipediaSearchTool()
+    final_answer_tool = FinalAnswerTool()
+    python_interpreter_tool = PythonInterpreterTool()    
+    agent = CodeAgent(tools=[duck_duck_go_search_tool, wikipedia_search_tool, python_interpreter_tool, final_answer_tool], model=InferenceClientModel(), add_base_tools=True, additional_authorized_imports=["pandas"]) 
+    
+    response = agent.run(question)  # Use run() instead of query()
+    return str(response)

workflow.py

@@ -0,0 +1,117 @@
+from typing import TypedDict, Annotated, Callable, Optional, Any
+from langgraph.graph import StateGraph, END
+from answering import gen_question_answer
+
+
+class AgentState(TypedDict):
+    question: str
+    answer: Annotated[str, lambda x, y: y]  # Overwrite with new value
+    formatted_answer: Annotated[str, lambda x, y: y]
+
+class GAIAAnsweringWorkflow:
+    def __init__(
+        self,
+        qa_function: Optional[Callable[[str], str]] = None,
+        formatter: Optional[Callable[[str], str]] = None
+    ):
+        """
+        Initialize the GAIA agent workflow
+        
+        Args:
+            qa_function: Core question answering function (gen_question_answer)
+            formatter: Answer formatting function (default: GAIA boxed format)
+        """
+        self.qa_function = gen_question_answer #qa_function or self.default_qa_function
+        self.formatter = formatter or self.default_formatter
+        self.workflow = self.build_workflow()
+        
+    @staticmethod
+    def default_qa_function(question: str) -> str:
+        """Placeholder QA function (override with your CodeAgent)"""
+        return "42"
+    
+    @staticmethod
+    def default_formatter(answer: str) -> str:
+        """Default GAIA formatting"""
+        return answer #f"\\boxed{{{answer}}}"
+    
+    def build_workflow(self) -> Any:
+        """Construct and compile the LangGraph workflow"""
+        # Create graph
+        workflow = StateGraph(AgentState)
+        
+        # Add nodes
+        workflow.add_node("generate_answer", self.generate_answer_node)
+        workflow.add_node("format_output", self.format_output_node)
+        
+        # Define edges
+        workflow.set_entry_point("generate_answer")
+        workflow.add_edge("generate_answer", "format_output")
+        workflow.add_edge("format_output", END)
+        
+        return workflow.compile()
+    
+    def generate_answer_node(self, state: AgentState) -> dict:
+        """Node that executes the question answering tool"""
+        try:
+            answer = self.qa_function(state["question"])
+            return {"answer": answer}
+        except Exception as e:
+            print(str(e))
+            return {"answer": f"Error: {str(e)}"}
+    
+    def format_output_node(self, state: AgentState) -> dict:
+        """Node that formats the answer for GAIA benchmark"""
+        try:
+            formatted = self.formatter(state["answer"])
+            return {"formatted_answer": formatted}
+        except Exception as e:
+            return {"formatted_answer": f"\\boxed{{\\text{{Formatting error: {str(e)}}}}}"}
+    
+    def __call__(self, question: str) -> str:
+        """
+        Execute the agent workflow for a given question
+        
+        Args:
+            question: Input question string
+            
+        Returns:
+            Formatted GAIA answer
+        """
+        # Initialize state
+        initial_state = {
+            "question": question,
+            "answer": "",
+            "formatted_answer": ""
+        }
+        
+        # Execute workflow
+        result = self.workflow.invoke(initial_state)
+        return result["formatted_answer"]
+
+# Example usage with custom QA function
+if __name__ == "__main__":
+    # Custom QA function (replace with your CodeAgent integration)
+    def custom_qa(question: str) -> str:
+        if "life" in question:
+            return "42"
+        elif "prime" in question:
+            return "101"
+        return "unknown"
+    
+    # Create agent instance
+    agent = GAIAAnsweringWorkflow(
+        qa_function=gen_question_answer,
+        formatter=lambda ans: f"ANSWER: \\boxed{{{ans}}}"  # Custom formatting
+    )
+    
+    # Test cases
+    questions = [
+        "What is the answer to life, the universe, and everything?",
+        "What is the smallest 3-digit prime number?",
+        "Unknown question type?"
+    ]
+    
+    for q in questions:
+        result = agent(q)
+        print(f"Question: {q}\nAnswer: {result}\n")