import json from typing import List, Dict, Optional from litellm import completion from agent_functions import AVAILABLE_FUNCTIONS, dataset_handler from llm_client import LLMClient class AgriculturalAgent: """LLM Agent for answering questions about agricultural research using the CGIAR dataset.""" def __init__(self, api_key: str, model: str = "groq/llama-3.3-70b-versatile"): """ Initialize the agent. Args: api_key: Groq API key model: Model to use (default: groq/llama-3.3-70b-versatile) """ self.llm_client = LLMClient(model=model, api_key=api_key) self.model = model self.conversation_history: List[Dict] = [] print("Initializing dataset...") dataset_handler.load_dataset() print("Dataset ready!") def get_tools_schema(self) -> List[Dict]: """Get the tools schema for function calling.""" tools = [] for func_name, func_info in AVAILABLE_FUNCTIONS.items(): tools.append({ "type": "function", "function": { "name": func_name, "description": func_info["description"], "parameters": func_info["parameters"] } }) return tools def call_function(self, function_name: str, arguments: Dict) -> str: """ Call a function by name with the provided arguments. Args: function_name: Name of the function to call arguments: Arguments to pass to the function Returns: Function result as string """ if function_name not in AVAILABLE_FUNCTIONS: return f"Error: Function '{function_name}' not found." func = AVAILABLE_FUNCTIONS[function_name]["function"] try: result = func(**arguments) return result except Exception as e: return f"Error calling function {function_name}: {str(e)}" def chat(self, user_message: str, system_message: Optional[str] = None) -> str: """ Process a user message and return a response. Args: user_message: The user's message system_message: Optional system message to override default Returns: Agent's response """ print(f"[AGENT] Processing user message: {user_message[:100]}...") if system_message is None: system_message = """You are an AI assistant specialized in agricultural research. You have access to a comprehensive dataset of agricultural research publications from CGIAR. Your role is to: 1. Help users find relevant agricultural research documents 2. Answer questions about agricultural topics using information from the dataset 3. Provide insights based on the research papers available When a user asks a question: - If they ask about a specific topic, crop, or agricultural concept, use the search_agricultural_documents function - If they want to browse topics, use the browse_topics function - If they ask about a specific document, use get_document_details - If they ask about the dataset itself, use get_dataset_info Always be helpful, accurate, and cite the sources when providing information from the dataset. If you don't have enough information, suggest searching for more specific documents.""" self.conversation_history.append({ "role": "user", "content": user_message }) messages = [{"role": "system", "content": system_message}] messages.extend(self.conversation_history) tools = self.get_tools_schema() print(f"[AGENT] Calling LLM API with model: {self.model}") try: max_iterations = 5 iteration = 0 last_tool_calls = [] tool_messages = [] while iteration < max_iterations: iteration += 1 print(f"[AGENT] LLM API call iteration {iteration}...") response = completion( model=self.model, messages=messages, tools=tools, tool_choice="auto", temperature=0.1, max_tokens=2048 ) message = response.choices[0].message if hasattr(message, 'tool_calls') and message.tool_calls: print(f"[AGENT] LLM requested {len(message.tool_calls)} tool call(s)") current_tool_calls = [(tc.function.name if hasattr(tc, 'function') else tc.get('function', {}).get('name'), tc.function.arguments if hasattr(tc, 'function') else tc.get('function', {}).get('arguments', '{}')) for tc in message.tool_calls] if iteration > 1 and current_tool_calls == last_tool_calls: print(f"[AGENT] Warning: Same tool calls detected, breaking loop") if tool_messages: assistant_response = tool_messages[-1].get('content', '')[:500] + "..." else: assistant_response = "I encountered an issue processing your request. Please try rephrasing your question." break last_tool_calls = current_tool_calls tool_calls_data = [] for tc in message.tool_calls: tool_calls_data.append({ "id": tc.id if hasattr(tc, 'id') else str(hash(str(tc))), "type": tc.type if hasattr(tc, 'type') else "function", "function": { "name": tc.function.name if hasattr(tc, 'function') else tc.get('function', {}).get('name'), "arguments": tc.function.arguments if hasattr(tc, 'function') else tc.get('function', {}).get('arguments', '{}') } }) self.conversation_history.append({ "role": "assistant", "content": message.content or "", "tool_calls": tool_calls_data }) tool_messages = [] for tc in message.tool_calls: function_name = tc.function.name if hasattr(tc, 'function') else tc.get('function', {}).get('name') try: arguments_str = tc.function.arguments if hasattr(tc, 'function') else tc.get('function', {}).get('arguments', '{}') if arguments_str is None or arguments_str == '' or arguments_str == 'null': arguments = {} else: arguments = json.loads(arguments_str) except (json.JSONDecodeError, AttributeError, TypeError) as e: print(f"[AGENT] Warning: Failed to parse arguments: {e}, using empty dict") arguments = {} print(f"[AGENT] Calling function: {function_name} with args: {arguments}") function_result = self.call_function(function_name, arguments) print(f"[AGENT] Function {function_name} returned result (length: {len(function_result)} chars)") tool_call_id = tc.id if hasattr(tc, 'id') else str(hash(str(tc))) tool_messages.append({ "role": "tool", "tool_call_id": tool_call_id, "name": function_name, "content": function_result }) self.conversation_history.extend(tool_messages) messages = [{"role": "system", "content": system_message}] messages.extend(self.conversation_history) else: assistant_response = message.content or "I apologize, but I couldn't generate a response." print(f"[AGENT] Generated response (length: {len(assistant_response)} chars)") self.conversation_history.append({ "role": "assistant", "content": assistant_response }) return assistant_response if tool_messages: print(f"[AGENT] Max iterations reached, returning last tool result") assistant_response = tool_messages[-1].get('content', '') self.conversation_history.append({ "role": "assistant", "content": assistant_response }) return assistant_response return "I apologize, but I encountered an issue processing your request. Please try again." except Exception as e: error_msg = f"Error in chat: {str(e)}" print(f"[AGENT] ERROR: {error_msg}") self.conversation_history.append({ "role": "assistant", "content": error_msg }) return error_msg def reset_conversation(self): """Reset the conversation history.""" self.conversation_history = []