Jake/devin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: ✨ Working tool calls

Browse Source 
o
	new file:   llm_interface.py
This commit is contained in:
Jake Pullen
2025-09-12 15:32:03 +01:00
parent fd2432df38
commit 30989e62ca
8 changed files with 6347 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.cache.sqlite
BIN
View File
Binary file not shown.
.gitignore
+1 -1
View File
@@ -1,6 +1,6 @@
.continue/*
.ropeproject/*
random_*
# Python-generated files
__pycache__/
app.log
+6096
View File
File diff suppressed because one or more lines are too long
config.py
+72
View File
@@ -0,0 +1,72 @@
# App config
LM_STUDIO_URL = "http://127.0.0.1:1234/v1/chat/completions"
EXIT_STRINGS = ['exit','goodbye','go away','fuck off', 'bye']
# LLM Config
SYSTEM_MESSAGE = '''You have the following tools available,
if you cant use a tool, you dont need to tell me, just answer normally.
if you are using a tool reply only with the exact JSON format shown in examples with NO SPACES and NO OTHER TEXT.
CRITICAL: When calling tools, use COMPACT JSON with NO SPACES:
- Correct: {"tool":"get_weather","parameters":{"city":"New York"}}
- Wrong: { "tool": "get_weather", "parameters": { "city": "New York" } }
{
"name": "get_weather",
"description": "Get current weather for a location",
"examples": [
{
"input": {"tool":"get_weather","parameters":{"city":"New York"}},
"output": {"temperature": 22, "condition": "partly cloudy", "humidity": 65}
},
{
"input": {"tool":"get_weather","parameters":{"city":"London"}},
"output": {"temperature": 18, "condition": "rainy", "humidity": 80}
}
]
},
{
"name": "find_folder",
"description": "Find any folder that matches the name provided on your machine or an optional directory",
"examples": [
{
"input": {"tool":"find_folder","parameters":{"folder_name":"devin"}},
},
{
"input": {"tool":"find_folder","parameters":{"folder_name":"winutils"}},
}
]
}'''
# Tool config
# WMO Weather interpretation codes mapping
WEATHER_CODE_MAP = {
0: "Clear sky",
1: "Mainly clear",
2: "Partly cloudy",
3: "Overcast",
45: "Fog",
48: "Depositing rime fog",
51: "Light drizzle",
53: "Moderate drizzle",
55: "Dense drizzle",
56: "Light freezing drizzle",
57: "Dense freezing drizzle",
61: "Slight rain",
63: "Moderate rain",
65: "Heavy rain",
66: "Light freezing rain",
67: "Heavy freezing rain",
71: "Slight snow",
73: "Moderate snow",
75: "Heavy snow",
77: "Snow grains",
80: "Slight rain showers",
81: "Moderate rain showers",
82: "Violent rain showers",
85: "Slight snow showers",
86: "Heavy snow showers",
95: "Thunderstorm",
96: "Thunderstorm with slight hail",
99: "Thunderstorm with heavy hail"
}
llm_interface.py
+73
View File
@@ -0,0 +1,73 @@
import config
from tools import ToolExecutor
import requests
from rich import print
from rich.prompt import Prompt
#TODO: context squish when it gets big
class llm():
def __init__(self, logger) -> None:
self.logger = logger
self.first_prompt = 1
self.context = {
"history": [],
"summary": None,
"metadata": {
"created_at": None,
"token_budget": 8000,
}
}
def summarise_context(self):
# add ai summary & created at time
pass
def add_to_history(self, history_object):
self.context['history'].append(history_object)
self.logger.debug(self.context['history'])
def ask_model(self, prompt: str, url=config.LM_STUDIO_URL) -> str:
# prompt = config.PRE_PROMPT + prompt
payload = {
"model": 'qwen/qwen3-coder-30b',
"messages": [],
"temperature": 0.7,
"max_tokens": 2048,
}
self.add_to_history({"role": "user", "content": prompt})
payload["messages"] = [
{"role": "system", "content": config.SYSTEM_MESSAGE}
] + self.context['history']
self.logger.debug(f'json payload: {payload}')
resp = requests.post(url, json=payload)
resp.raise_for_status()
self.logger.debug(resp.json())
self.add_to_history(resp.json()["choices"][0]["message"])
return resp.json()["choices"][0]["message"]["content"].strip()
def tool_response(self, prompt: str, url=config.LM_STUDIO_URL) -> str:
payload = {
"model": 'qwen/qwen3-coder-30b',
"messages": [],
"temperature": 0.7,
"max_tokens": 2048,
}
payload["messages"] = [{"role": "tool", "content": prompt}]
resp = requests.post(url, json=payload)
resp.raise_for_status()
self.logger.debug(resp.json())
return resp.json()["choices"][0]["message"]["content"].strip()
def get_llm_prompt(self):
while True:
llm_prompt = Prompt.ask("[bold cyan]Ask your local llm[/]")
self.logger.debug(f'prompt is: {llm_prompt}')
if llm_prompt == '':
self.logger.error("Cannot be empty!")
return llm_prompt
def handle_llm_reply(self, llm_reply:str):
architect = ToolExecutor(self.logger)
return architect.process_llm_response(llm_output=llm_reply)
main.py
+49 -73
View File
@@ -1,91 +1,51 @@
from rich import print
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, BarColumn, track
from rich.markdown import Markdown
from rich.syntax import Syntax
from rich.live import Live
from rich.prompt import Prompt
import requests
import sys
from tools import ToolExecutor
import random
import time
LM_STUDIO_URL = "http://127.0.0.1:1234/v1/chat/completions"
EXIT_STRINGS = ['exit','goodbye','go away','fuck off']
import sys
import time
import config
from llm_interface import llm
import logging
#TODO: add context for llm calls so we have history and can chain messages
console = Console()
PRE_PROMPT = '''You have the following tools available, only use the tools if you need to.
please reply with only the tool call if you need to
{
"name": "get_weather",
"description": "Get current weather for a location",
"examples": [
{
"input": {"tool": "get_weather", "parameters":{"city":"New York"}},
"output": {"temperature": 22, "condition": "partly cloudy", "humidity": 65}
},
{
"input": {"tool": "get_weather", "parameters":{"city":"London"}},
"output": {"temperature": 18, "condition": "rainy", "humidity": 80}
}
]
}'''
class RichConsoleHandler(logging.StreamHandler):
def emit(self, record):
if record.levelno >= logging.CRITICAL:
console.print(f"[bold magenta]CRITICAL:[/bold magenta] {record.getMessage()}")
elif record.levelno >= logging.ERROR:
console.print(f"[bold red]ERROR:[/bold red] {record.getMessage()}")
elif record.levelno >= logging.WARNING:
console.print(f"[bold yellow]WARNING:[/bold yellow] {record.getMessage()}")
# elif record.levelno >= logging.INFO:
# console.print(f"[bold blue]INFO:[/bold blue] {record.getMessage()}")
# elif record.levelno >= logging.DEBUG:
# console.print(f"[bold green]DEBUG:[/bold green] {record.getMessage()}")
def ask_model(prompt: str, url=LM_STUDIO_URL) -> str:
payload = {
"model": 'qwen/qwen3-coder-30b',
"messages": [{"role": "user", "content": PRE_PROMPT+prompt}],
"temperature": 0.7,
"max_tokens": 2048,
}
resp = requests.post(url, json=payload)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"].strip()
def ask_model_no_pre(prompt: str, url=LM_STUDIO_URL) -> str:
payload = {
#"model": 'qwen/qwen3-coder-30b',
"model": 'openai/gpt-oss-20b',
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"max_tokens": 2048,
}
resp = requests.post(url, json=payload)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"].strip()
def get_llm_prompt():
def main(logger):
logger.info('Application Started')
language_model = llm(logger)
while True:
llm_prompt = Prompt.ask("[bold cyan]Ask your local llm[/]")
if llm_prompt.strip():
return llm_prompt.strip()
console.print("[red]Cannot be empty![/]", style="bold")
return llm_prompt
def handle_llm_reply(llm_reply:str):
architect = ToolExecutor()
return architect.process_llm_response(llm_output=llm_reply)
def main():
while True:
llm_prompt = get_llm_prompt()
if llm_prompt.lower() in EXIT_STRINGS:
llm_prompt = language_model.get_llm_prompt()
if llm_prompt.lower() in config.EXIT_STRINGS:
sys.exit()
start = time.time()
llm_reply = ask_model(llm_prompt)
print(llm_reply)
handled_response = handle_llm_reply(llm_reply)
print(handled_response)
print(handled_response == llm_reply)
llm_reply = language_model.ask_model(llm_prompt)
logger.info(f"LLM Reply: {llm_reply}")
handled_response = language_model.handle_llm_reply(llm_reply)
logger.info(f"Handled Response: {handled_response}")
logger.info(f"Response equals original: {handled_response == llm_reply}")
if handled_response != llm_reply:
output = ask_model_no_pre(prompt=f'Make this lovely markdown, use fun emojis {handled_response}')
output = language_model.tool_response(prompt=f'Make this lovely markdown, use fun emojis {handled_response}')
elif handled_response == llm_reply:
output = handled_response
@@ -99,5 +59,21 @@ def main():
console.print(panel_narrow)
if __name__ == "__main__":
main()
# Setup logging with custom handler for warnings and errors
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('app.log')
]
)
logger = logging.getLogger(__name__)
logger.addHandler(RichConsoleHandler())
logger.info("Logging Instantiated")
main(logger)
random_bullshit.py
-46
View File
@@ -1,46 +0,0 @@
# def ask_name() -> str:
# """Ask for a name and keep asking until it's not empty."""
# while True:
# name = Prompt.ask("[bold cyan]What is your name?[/]")
# if name.strip():
# return name.strip()
# console.print("[red]Name cannot be empty![/]", style="bold")
# def ask_age() -> int:
# """Ask for age and validate it's a positive integer."""
# while True:
# age_str = Prompt.ask("[bold cyan]How old are you?[/]")
# try:
# age = int(age_str)
# if age <= 0:
# raise ValueError
# return age
# except ValueError:
# console.print("[red]Please enter a valid positive integer.[/]", style="bold")
pre_prompt = '''{
"name": "get_weather",
"description": "Get current weather for a location",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "City name"},
"units": {"type": "string", "enum": ["celsius", "fahrenheit"], "description": "Temperature units"}
},
"required": ["city"]
},
"examples": [
{
"input": {"city": "New York", "units": "celsius"},
"output": {"temperature": 22, "condition": "partly cloudy", "humidity": 65}
},
{
"input": {"city": "London"},
"output": {"temperature": 18, "condition": "rainy", "humidity": 80}
}
]
}'''
# def get_weather(city, units="celsius"):
# # Your logic here (API calls, calculations, etc.)
# return {"temperature": 25, "condition": "sunny"}
tools.py
+52 -45
View File
@@ -3,56 +3,28 @@ import openmeteo_requests
import requests_cache
from retry_requests import retry
import json
# WMO Weather interpretation codes mapping
WEATHER_CODE_MAP = {
0: "Clear sky",
1: "Mainly clear",
2: "Partly cloudy",
3: "Overcast",
45: "Fog",
48: "Depositing rime fog",
51: "Light drizzle",
53: "Moderate drizzle",
55: "Dense drizzle",
56: "Light freezing drizzle",
57: "Dense freezing drizzle",
61: "Slight rain",
63: "Moderate rain",
65: "Heavy rain",
66: "Light freezing rain",
67: "Heavy freezing rain",
71: "Slight snow",
73: "Moderate snow",
75: "Heavy snow",
77: "Snow grains",
80: "Slight rain showers",
81: "Moderate rain showers",
82: "Violent rain showers",
85: "Slight snow showers",
86: "Heavy snow showers",
95: "Thunderstorm",
96: "Thunderstorm with slight hail",
99: "Thunderstorm with heavy hail"
}
import config
import logging
import os
import subprocess
class ToolExecutor:
def __init__(self):
def __init__(self, logger):
self.tools = {
"get_weather": self.get_weather
"get_weather": self.get_weather,
"find_folder": self.find_folder
}
self.logger = logger
def process_llm_response(self, llm_output):
# Parse the tool call from LLM response
if llm_output.startswith('{"tool":'):
# Parse the tool call from LLM response
llm_output = json.loads(llm_output)
print(llm_output)
self.logger.info(f"Parsed LLM output: {llm_output}")
tool_name = llm_output.get("tool",'no tool')
parameters = llm_output.get("parameters", 'no parameters')
print(f'parsed tool: {tool_name}, parsed parameters: {parameters}')
self.logger.info(f'parsed tool: {tool_name}, parsed parameters: {parameters}')
# Execute the actual function
result = self.tools[tool_name](**parameters)
@@ -64,7 +36,7 @@ class ToolExecutor:
def get_weather(self, city: str, units: str = "celsius"):
"""Get current weather for a location using Open-Meteo API with openmeteo-requests library."""
print(f'get weather called, city = {city}')
self.logger.info(f'get weather called, city = {city}')
if not city:
raise ValueError("City parameter is required")
@@ -86,7 +58,7 @@ class ToolExecutor:
geo_response = requests.get(geocoding_url, params=geocoding_params, timeout=10)
geo_response.raise_for_status()
geo_data = geo_response.json()
print(f'Geo data for city: {geo_data}')
self.logger.info(f'Geo data for city: {geo_data}')
if not geo_data.get("results"):
raise Exception(f"City '{city}' not found")
@@ -119,11 +91,11 @@ class ToolExecutor:
responses = openmeteo.weather_api(url, params=params)
response = responses[0]
print(f'Weather API response: {response}')
self.logger.info(f'Weather API response: {response}')
# Get current weather data
current = response.Current()
print(f'Current weather data: {current}')
self.logger.info(f'Current weather data: {current}')
# Extract values using the library's methods
temperature = current.Variables(0).Value() # temperature_2m
@@ -132,7 +104,7 @@ class ToolExecutor:
pressure = current.Variables(3).Value() # surface_pressure
wind_speed = current.Variables(4).Value() # wind_speed_10m
condition = WEATHER_CODE_MAP.get(weather_code, "Unknown")
condition = config.WEATHER_CODE_MAP.get(weather_code, "Unknown")
json_reply = {
@@ -146,8 +118,43 @@ class ToolExecutor:
"country": country,
"coordinates": {"latitude": latitude, "longitude": longitude}
}
print(json_reply)
self.logger.info(f"Weather data result: {json_reply}")
return json_reply
except Exception as e:
raise Exception(f"Weather lookup failed: {str(e)}")
def find_folder(self, folder_name: str, search_path: str = "/home/"):
"""Search for folders with a specific name on the PC."""
self.logger.info(f'find_folder called, folder_name = {folder_name}, search_path = {search_path}')
if not folder_name:
raise ValueError("folder_name parameter is required")
try:
found_folders = []
# Use find command to search for directories
cmd = ['find', search_path, '-type', 'd', '-name', folder_name]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
self.logger.warning(result)
if result.stdout:
found_folders = [path.strip() for path in result.stdout.split('\n') if path.strip()]
# Limit results to first 50 to avoid overwhelming output
found_folders = found_folders[:50]
json_reply = {
"folder_name": folder_name,
"search_path": search_path,
"found_folders": found_folders,
"count": len(found_folders)
}
self.logger.info(f"Folder search result: {json_reply}")
return json_reply
except subprocess.TimeoutExpired:
raise Exception("Folder search timed out after 30 seconds")
except Exception as e:
raise Exception(f"Folder search failed: {str(e)}")