""" Generative AI Agent with Tool Calling Capabilities This module provides an intelligent agent that can use multiple tools (Wikipedia, Tavily) to answer user queries with up-to-date and accurate information. """ import os import yaml import logging from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.tools import Tool from tools.wikipedia_tool import WikipediaTool from langchain_tavily import TavilySearch from langgraph.prebuilt import create_react_agent # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class GenerativeAIAgent: """ An intelligent agent that uses LangGraph's ReAct pattern to answer queries. The agent can dynamically select and use tools (Wikipedia, Tavily Search) based on the user's query to provide accurate and up-to-date information. Attributes: llm: The language model (ChatOpenAI) tools: List of available tools for the agent agent_executor: The LangGraph ReAct agent executor """ def __init__(self, config_path: str = "config.yaml"): """ Initialize the GenerativeAIAgent with configuration and tools. Args: config_path: Path to the YAML configuration file """ logger.info("Initializing GenerativeAIAgent...") # Load environment variables load_dotenv(dotenv_path="local/.env") # Load configuration self.config = self._load_config(config_path) # Initialize tools self.wikipedia_tool = WikipediaTool(config_path) self.tavily_search = TavilySearch(max_results=5) # Define available tools self.tools = self._initialize_tools() # Initialize language model self.llm = self._initialize_llm() # Create ReAct agent executor self.agent_executor = create_react_agent(self.llm, self.tools) logger.info("Agent initialized successfully with %d tools", len(self.tools)) def _load_config(self, config_path: str) -> dict: """Load configuration from YAML file.""" try: with open(config_path, "r") as file: config = yaml.safe_load(file) logger.info("Configuration loaded from %s", config_path) return config except FileNotFoundError: logger.error("Config file not found: %s", config_path) raise except yaml.YAMLError as e: logger.error("Error parsing config file: %s", e) raise def _initialize_tools(self) -> list: """Initialize and return the list of tools available to the agent.""" tools = [ Tool( name="Wikipedia", description=( "Search Wikipedia for factual, encyclopedic information. " "Best for: historical facts, scientific concepts, biographies, " "general knowledge. Input should be a clear search query." ), func=self.wikipedia_tool.search ), Tool( name="Tavily", description=( "Search the web for current information and latest news. " "Best for: recent events, breaking news, current trends, " "real-time data. Input should be a search query." ), func=self.tavily_search.invoke ) ] logger.info("Initialized tools: %s", [tool.name for tool in tools]) return tools def _initialize_llm(self) -> ChatOpenAI: """Initialize the language model with configuration.""" model_config = self.config.get("openai", {}) llm = ChatOpenAI( model=model_config.get("model", "gpt-5"), temperature=model_config.get("temperature", 0.7), max_tokens=model_config.get("max_tokens", 1000), api_key=os.getenv("OPENAI_API_KEY") ) logger.info("LLM initialized: %s", model_config.get("model")) return llm def generate_response(self, user_input: str) -> str: """ Generate a response to the user's input using the agent. The agent will automatically select and use appropriate tools based on the query, following the ReAct (Reasoning + Acting) pattern. Args: user_input: The user's question or query Returns: str: The agent's response """ if not user_input or not user_input.strip(): logger.warning("Empty input received") return "Please provide a valid question or query." try: logger.info("Processing query: %s", user_input[:50] + "..." if len(user_input) > 50 else user_input) # Get system prompt from config system_prompt = self.config.get("app", {}).get( "system_prompt", "You are a helpful AI assistant with access to Wikipedia and web search tools." ) # Prepare messages for the agent messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_input} ] # Invoke the agent executor response = self.agent_executor.invoke({"messages": messages}) # Extract the final answer from the response final_answer = self._extract_final_answer(response) logger.info("Response generated successfully") return final_answer except Exception as e: logger.error("Error generating response: %s", str(e), exc_info=True) return self._format_error_message(str(e)) def _extract_final_answer(self, response: dict) -> str: """ Extract the final answer from the agent's response. Args: response: The response dictionary from the agent executor Returns: str: The extracted final answer """ if isinstance(response, dict) and "messages" in response: # Iterate through messages in reverse to find the last AI message with content for msg in reversed(response["messages"]): if hasattr(msg, "content") and msg.content and msg.content.strip(): return msg.content.strip() # Fallback logger.warning("Could not extract proper answer from response") return "I apologize, but I couldn't generate a proper response. Please try rephrasing your question." def _format_error_message(self, error: str) -> str: """ Format error messages in a user-friendly way. Args: error: The error message Returns: str: A formatted error message """ if "rate limit" in error.lower(): return "⚠️ Rate limit reached. Please wait a moment and try again." elif "api key" in error.lower(): return "⚠️ API authentication error. Please check your API keys." elif "timeout" in error.lower(): return "⚠️ Request timed out. Please try again." else: return f"⚠️ An error occurred: {error}\n\nPlease try rephrasing your question or try again later." def get_available_tools(self) -> list: """ Get a list of available tools and their descriptions. Returns: list: List of dictionaries containing tool information """ return [ { "name": tool.name, "description": tool.description } for tool in self.tools ]