opencode
a309c7e27c
- Nuovo modulo ollama_agent.py con parsing delle risposte - Supporta chiamate tool per provider senza bind_tools - Tool built-in: execute, read_file, write_file, ls, search, system_info - Il modello risponde con [TOOL_CALL] che noi parsiamo ed eseguiamo - Integrato in main.py per uso automatico con Ollama
411 lines
15 KiB
Python
411 lines
15 KiB
Python
"""
|
|
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
|
|
)
|