diff --git a/main.py b/main.py index 82e3f70..7ef87e9 100644 --- a/main.py +++ b/main.py @@ -414,19 +414,52 @@ def create_langcode_agent(system_prompt: Optional[str] = None, use_simple_agent: """ Create and return an agent instance with configurable model. - For Ollama and other providers without full tool-calling support, - uses a simple chat agent instead of deepagents. + For Ollama uses custom OllamaToolAgent with tool calling support. + For other providers with full tool support uses deepagents. """ - # Detect if we need simple agent (for Ollama without tool support) - if not use_simple_agent and CONFIG_AVAILABLE: + # Try to get config info for Ollama detection + is_ollama = False + if CONFIG_AVAILABLE: active_name = config_manager.get_active_configuration() if active_name: config = config_manager.get_configuration(active_name) if config and config.get("provider") == "ollama": - # Ollama doesn't fully support bind_tools in deepagents - # Always use simple agent for Ollama - print("⚠️ Using simple chat agent (Ollama detected)") - use_simple_agent = True + is_ollama = True + + # Use OllamaToolAgent for Ollama with tool calling + if is_ollama and not use_simple_agent: + try: + from ollama_agent import create_ollama_tool_agent + model = load_model_from_config() + custom_tools = get_custom_tools() + + # Enhance with memory and planning tools if available + if ENHANCEMENTS_AVAILABLE: + try: + memory_tools = create_simple_memory_tools(simple_memory) + custom_tools.extend(memory_tools) + except Exception: + pass + try: + planning_tools = enhanced_planner.create_planning_tools() + custom_tools.extend(planning_tools) + except Exception: + pass + + print("🔧 Using OllamaToolAgent with tool calling") + return create_ollama_tool_agent( + model=model, + tools=custom_tools, + system_prompt=system_prompt, + include_builtin_tools=True + ) + except Exception as e: + print(f"⚠️ Could not load OllamaToolAgent: {e}") + print("⚠️ Falling back to simple agent") + use_simple_agent = True + elif is_ollama: + print("⚠️ Using simple chat agent (Ollama detected)") + use_simple_agent = True if use_simple_agent: return create_simple_agent(system_prompt) @@ -1099,9 +1132,17 @@ def main(): # Otherwise, treat it as a chat message try: result = agent.invoke({"messages": [HumanMessage(content=user_input)]}) - # Extract the AI's response - ai_message = result["messages"][-1] - print(f"\n🤖 LangCode: {ai_message.content}") + + # Extract the AI's response (works for both simple and tool agent) + if "response" in result: + # OllamaToolAgent returns a separate response field + response_text = result["response"] + else: + # Simple agent returns the last message + ai_message = result["messages"][-1] + response_text = ai_message.content if hasattr(ai_message, 'content') else str(ai_message) + + print(f"\n🤖 LangCode: {response_text}") except Exception as e: error_msg = str(e) if not error_msg: diff --git a/ollama_agent.py b/ollama_agent.py new file mode 100644 index 0000000..fdee0bb --- /dev/null +++ b/ollama_agent.py @@ -0,0 +1,410 @@ +""" +Ollama Agent with Tool Calling Support + +Questo modulo implementa un agente che supporta tool calling per Ollama +usando un approccio di parsing delle risposte invece di bind_tools. +""" +import json +import re +import uuid +import subprocess +from typing import List, Dict, Any, Optional, Callable +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path + +from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage +from langchain_core.tools import Tool + + +# ============================================================================= +# TOOL CALL PATTERNS +# ============================================================================= + +TOOL_CALL_PATTERN = r'\[TOOL_CALL\]\s*(.*?)\s*\[\/TOOL_CALL\]' +TOOL_RESULT_PATTERN = r'\[TOOL_RESULT\]\s*(.*?)\s*\[\/TOOL_RESULT\]' + + +# ============================================================================= +# BASIC BUILT-IN TOOLS +# ============================================================================= + +def get_builtin_tools() -> List[Tool]: + """ + Restituisce i tool built-in disponibili per l'agente. + """ + tools = [] + + # Tool: Esegui comando shell + def execute_command(command: str) -> str: + """ + Execute a shell command and return the output. + USE WITH CAUTION - only for safe commands. + """ + try: + result = subprocess.run( + command, + shell=True, + capture_output=True, + text=True, + timeout=30 + ) + output = [] + if result.stdout: + output.append(f"STDOUT:\n{result.stdout}") + if result.stderr: + output.append(f"STDERR:\n{result.stderr}") + if result.returncode != 0: + output.append(f"Exit code: {result.returncode}") + return "\n".join(output) if output else "Command executed with no output" + except subprocess.TimeoutExpired: + return "Error: Command timed out after 30 seconds" + except Exception as e: + return f"Error: {str(e)}" + + tools.append(Tool( + name="execute", + description="Execute a shell command. Use this for file operations, running scripts, git commands, etc. Input is the command string to execute.", + func=execute_command + )) + + # Tool: Leggi file + def read_file(path: str) -> str: + """ + Read the contents of a file. + """ + try: + file_path = Path(path).expanduser() + if not file_path.exists(): + return f"Error: File '{path}' not found" + if file_path.is_dir(): + return f"Error: '{path}' is a directory, not a file" + return file_path.read_text(encoding='utf-8') + except Exception as e: + return f"Error reading file: {str(e)}" + + tools.append(Tool( + name="read_file", + description="Read the contents of a file. Input is the file path (can use ~ for home directory).", + func=read_file + )) + + # Tool: Scrivi file + def write_file(path: str, content: str) -> str: + """ + Write content to a file. Creates the file if it doesn't exist. + WARNING: This will overwrite existing files! + """ + try: + file_path = Path(path).expanduser() + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text(content, encoding='utf-8') + return f"Successfully wrote to {path}" + except Exception as e: + return f"Error writing file: {str(e)}" + + tools.append(Tool( + name="write_file", + description="Write content to a file. Inputs: 'path' (file path) and 'content' (text to write). WARNING: overwrites existing files!", + func=write_file + )) + + # Tool: Lista file + def list_directory(path: str = ".") -> str: + """ + List files in a directory. + """ + try: + dir_path = Path(path).expanduser() + if not dir_path.exists(): + return f"Error: Directory '{path}' not found" + if not dir_path.is_dir(): + return f"Error: '{path}' is not a directory" + + items = [] + for item in sorted(dir_path.iterdir()): + item_type = "DIR" if item.is_dir() else "FILE" + items.append(f"{item_type}: {item.name}") + return "\n".join(items) if items else "Directory is empty" + except Exception as e: + return f"Error listing directory: {str(e)}" + + tools.append(Tool( + name="ls", + description="List files in a directory. Input is the directory path (default: current directory).", + func=list_directory + )) + + # Tool: Cerca file + def search_files(directory: str = ".", pattern: str = "*") -> str: + """ + Search for files matching a pattern in a directory. + """ + try: + from pathlib import Path + dir_path = Path(directory).expanduser() + if not dir_path.exists(): + return f"Error: Directory '{directory}' not found" + + matches = list(dir_path.rglob(pattern)) + if not matches: + return f"No files matching '{pattern}' found in {directory}" + + results = [] + for match in matches[:50]: # Limit to 50 results + results.append(str(match)) + + return f"Found {len(matches)} matches (showing first 50):\n" + "\n".join(results) + except Exception as e: + return f"Error searching files: {str(e)}" + + tools.append(Tool( + name="search", + description="Search for files by name pattern. Inputs: 'directory' (where to search) and 'pattern' (e.g., '*.py', '*.md').", + func=search_files + )) + + # Tool: Info sistema + def system_info() -> str: + """ + Get system information (OS, Python version, current directory). + """ + import platform + import os + info = [] + info.append(f"OS: {platform.system()} {platform.release()}") + info.append(f"Python: {platform.python_version()}") + info.append(f"Current dir: {os.getcwd()}") + info.append(f"User: {os.getenv('USER', 'unknown')}") + return "\n".join(info) + + tools.append(Tool( + name="system_info", + description="Get system information (OS, Python version, current directory). No input required.", + func=system_info + )) + + return tools + + +@dataclass +class ToolCall: + """Rappresenta una chiamata a tool""" + id: str + name: str + arguments: Dict[str, Any] + + +class OllamaToolAgent: + """ + Agente che supporta tool calling per Ollama e provider simili. + + Strategia: Il modello risponde con marcatori speciali per le chiamate tool. + Noi parsiamo, eseguiamo, e riportiamo i risultati. + """ + + def __init__( + self, + llm, + tools: List[Tool], + system_prompt: Optional[str] = None, + max_iterations: int = 10 + ): + self.llm = llm + self.tools = {tool.name: tool for tool in tools} + self.max_iterations = max_iterations + + # Costruisci il system prompt con la lista dei tool + if system_prompt is None: + system_prompt = self._default_system_prompt() + else: + system_prompt = system_prompt + "\n\n" + self._tool_instructions() + + self.messages: List[BaseMessage] = [SystemMessage(content=system_prompt)] + + # Tool per la memoria + self.short_term_memory: List[Dict] = [] + + def _default_system_prompt(self) -> str: + """System prompt di default""" + return """You are LangCode, a helpful AI assistant with access to tools. + +When you need to use a tool, you MUST respond with this EXACT format: + +[TOOL_CALL] +{"tool": "tool_name", "arguments": {"arg1": "value1", "arg2": "value2"}} +[/TOOL_CALL] + +After receiving the tool result, continue your response normally. + +Available tools:""" + + def _tool_instructions(self) -> str: + """Genera le istruzioni per i tool disponibili""" + instructions = [] + instructions.append("\n\n# AVAILABLE TOOLS\n") + instructions.append("You have access to the following tools:\n\n") + + for name, tool in self.tools.items(): + # Estrai la descrizione dal docstring del tool + desc = tool.description or "No description" + # Pulisci la descrizione da caratteri speciali + desc = desc.replace('"', '\\"').replace('\n', ' ') + instructions.append(f"## {name}\n") + instructions.append(f"Description: {desc}\n") + + # Parsa i parametri se presenti + if hasattr(tool, 'args_schema') and tool.args_schema: + instructions.append("Parameters:\n") + schema = tool.args_schema + # Handle both dict and BaseModel schemas + try: + model_json_schema = getattr(schema, 'model_json_schema', None) # type: ignore + if callable(model_json_schema): + schema = model_json_schema() + except Exception: + pass + if isinstance(schema, dict) and 'properties' in schema: + for param_name, param_info in schema['properties'].items(): + param_type = param_info.get('type', 'any') + param_desc = param_info.get('description', '') + required = param_name in schema.get('required', []) + req_marker = "(required)" if required else "(optional)" + instructions.append(f" - {param_name}: {param_type} {req_marker} - {param_desc}\n") + + instructions.append("\n") + + instructions.append("\n# IMPORTANT RULES:\n") + instructions.append("1. When you need to use a tool, respond ONLY with [TOOL_CALL] blocks\n") + instructions.append("2. After receiving the result, continue your response\n") + instructions.append("3. For simple questions, respond directly without tools\n") + instructions.append("4. Use valid JSON in tool calls\n") + instructions.append("5. If a tool fails, explain the error to the user\n") + + return ''.join(instructions) + + def _parse_tool_calls(self, text: str) -> List[ToolCall]: + """Parse le chiamate tool dalla risposta del modello""" + tool_calls = [] + matches = re.finditer(TOOL_CALL_PATTERN, text, re.DOTALL) + + for match in matches: + try: + data = json.loads(match.group(1)) + tool_calls.append(ToolCall( + id=str(uuid.uuid4())[:8], + name=data.get('tool'), + arguments=data.get('arguments', {}) + )) + except json.JSONDecodeError: + continue + + return tool_calls + + def _execute_tool(self, tool_call: ToolCall) -> Any: + """Esegui un singolo tool""" + if tool_call.name not in self.tools: + return {"error": f"Tool '{tool_call.name}' not found"} + + try: + tool = self.tools[tool_call.name] + result = tool.invoke(tool_call.arguments) + + # Converti result in stringa per il modello + if result is None: + return {"result": "Tool executed successfully"} + elif isinstance(result, (dict, list)): + return {"result": json.dumps(result, ensure_ascii=False, indent=2)} + else: + return {"result": str(result)} + + except Exception as e: + return {"error": str(e)} + + def _format_tool_result(self, tool_call: ToolCall, result: Any) -> str: + """Formatta il risultato di un tool per il modello""" + return f"\n[TOOL_RESULT]\nTool: {tool_call.name}\nResult: {json.dumps(result, ensure_ascii=False)}\n[/TOOL_RESULT]\n" + + def invoke(self, input_dict: Dict) -> Dict: + """ + Invoca l'agente con un input. + + L'agente può fare multiple iterazioni tool-call prima di rispondere. + """ + # Estrai il messaggio utente + user_message = input_dict.get("messages", [])[-1] + if hasattr(user_message, 'content'): + user_content = user_message.content + else: + user_content = str(user_message) + + # Aggiungi il messaggio utente alla cronologia + self.messages.append(HumanMessage(content=user_content)) + + # Loop principale: risposta -> parse tools -> esegui -> ripeti + iteration = 0 + final_response = None + + while iteration < self.max_iterations: + iteration += 1 + + # Genera risposta dal modello + response = self.llm.invoke(self.messages) + self.messages.append(response) + + response_text = response.content if hasattr(response, 'content') else str(response) + + # Parse le chiamate tool + tool_calls = self._parse_tool_calls(response_text) + + if not tool_calls: + # Nessuna chiamata tool, questa è la risposta finale + final_response = response_text + break + + # Esegui i tool e aggiungi i risultati + for tool_call in tool_calls: + result = self._execute_tool(tool_call) + tool_result_text = self._format_tool_result(tool_call, result) + self.messages.append(HumanMessage(content=tool_result_text)) + + # Se non abbiamo una risposta (troppe iterazioni) + if final_response is None: + final_response = "Ho raggiunto il numero massimo di iterazioni. Qualcosa potrebbe essere andato storto." + + return {"messages": self.messages, "response": final_response} + + def clear_history(self): + """Pulisci la cronologia ma mantieni il system prompt""" + system_msg = self.messages[0] if self.messages and isinstance(self.messages[0], SystemMessage) else None + self.messages = [system_msg] if system_msg else [] + + +def create_ollama_tool_agent( + model, + tools: Optional[List[Tool]] = None, + system_prompt: Optional[str] = None, + include_builtin_tools: bool = True +) -> OllamaToolAgent: + """ + Factory function per creare un OllamaToolAgent. + + Args: + model: Il modello LLM da usare + tools: Lista di tool personalizzati (opzionale) + system_prompt: System prompt personalizzato (opzionale) + include_builtin_tools: Se True, include i tool built-in (execute, read_file, etc.) + """ + if tools is None: + tools = [] + + # Aggiungi i tool built-in + if include_builtin_tools: + builtin = get_builtin_tools() + tools = builtin + tools + + return OllamaToolAgent( + llm=model, + tools=tools, + system_prompt=system_prompt + )