From d0bbd8aeea139ca4bf4ed37020c71c60a65f0a95 Mon Sep 17 00:00:00 2001 From: batsumaru <> Date: Fri, 11 Sep 2026 17:39:50 +0900 Subject: Added weather, file system, and memory --- .gitignore | 1 + agent.py | 20 ++++-- client.py | 14 +++- context.py | 40 ++++++++++- memory.py | 8 +++ tools.py | 233 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- 6 files changed, 305 insertions(+), 11 deletions(-) create mode 100644 memory.py diff --git a/.gitignore b/.gitignore index fe89a3a..0f92de7 100644 --- a/.gitignore +++ b/.gitignore @@ -175,3 +175,4 @@ pyrightconfig.json # End of https://www.toptal.com/developers/gitignore/api/python +data/ diff --git a/agent.py b/agent.py index 816f976..41c7a7b 100644 --- a/agent.py +++ b/agent.py @@ -4,6 +4,7 @@ from context import ContextBuilder from identity import AgentIdentity, UserProfile from llm import LLM from tools import ToolRegistry +from memory import WorkingMemory class Agent: @@ -22,14 +23,16 @@ class Agent: user_profile, ) - self.messages: list[dict[str, object]] = [ - { - "role": "system", - "content": self.context.build_system_prompt(), - } - ] + self.working_memory = WorkingMemory( + objective="Testing working memory", + facts=["The user is testing the agent's working memory system."], + notes=["This is temporary test data."], + ) + + self.messages: list[dict[str, object]] = [] def chat(self, user_input: str) -> str: + self.messages.append( { "role": "user", @@ -39,7 +42,10 @@ class Agent: for _ in range(5): response = self.llm.chat( - self.messages, + self.context.build_messages( + self.messages, + self.working_memory, + ), tools=self.tools.schemas(), ) diff --git a/client.py b/client.py index 8d6829d..0cbd91c 100644 --- a/client.py +++ b/client.py @@ -1,7 +1,14 @@ from agent import Agent from identity import load_agent_identity, load_user_profile from llm import LLM -from tools import CalculatorTool, HelloTool, ToolRegistry +from tools import ( + CalculatorTool, + FileSystemListTool, + FileSystemReadTool, + HelloTool, + ToolRegistry, + WeatherTool, +) llm = LLM( "http://localhost:8080/v1/chat/completions", @@ -13,6 +20,9 @@ user_profile = load_user_profile("user.toml") tools = ToolRegistry() tools.register(HelloTool()) tools.register(CalculatorTool()) +tools.register(WeatherTool()) +tools.register(FileSystemListTool()) +tools.register(FileSystemReadTool()) agent = Agent( llm, @@ -26,6 +36,6 @@ while True: if user_input.lower() in {"exit", "quit"}: break - + answer = agent.chat(user_input) print(f"Agent: {answer}") diff --git a/context.py b/context.py index 4a65f57..d702706 100644 --- a/context.py +++ b/context.py @@ -1,4 +1,5 @@ from identity import AgentIdentity, UserProfile +from memory import WorkingMemory class ContextBuilder: @@ -10,7 +11,10 @@ class ContextBuilder: self.agent_identity = agent_identity self.user_profile = user_profile - def build_system_prompt(self) -> str: + def build_system_prompt( + self, + memory: WorkingMemory, + ) -> str: agent = self.agent_identity user = self.user_profile @@ -22,4 +26,38 @@ class ContextBuilder: f"Favorite color: {agent.quirks['favorite_color']}.\n" f"\n" f"The user's name is {user.name}" + f"\n" + f"{self.build_working_memory(memory)}" ) + + def build_working_memory(self, memory: WorkingMemory) -> str: + lines = ["Current working memory:"] + + if memory.objective: + lines.append(f"Objective: {memory.objective}") + + if memory.facts: + lines.append("Facts:") + lines.extend(f"- {fact}" for fact in memory.facts) + + if memory.notes: + lines.append("Notes:") + lines.extend(f"- {note}" for note in memory.notes) + + if len(lines) == 1: + lines.append("(empty)") + + return "\n".join(lines) + + def build_messages( + self, + history: list[dict[str, object]], + memory: WorkingMemory, + ) -> list[dict[str, object]]: + return [ + { + "role": "system", + "content": self.build_system_prompt(memory), + }, + *history, + ] diff --git a/memory.py b/memory.py new file mode 100644 index 0000000..9269f94 --- /dev/null +++ b/memory.py @@ -0,0 +1,8 @@ +from dataclasses import dataclass, field + + +@dataclass +class WorkingMemory: + objective: str | None = None + facts: list[str] = field(default_factory=list) + notes: list[str] = field(default_factory=list) diff --git a/tools.py b/tools.py index 64238e7..2ae8857 100644 --- a/tools.py +++ b/tools.py @@ -1,3 +1,6 @@ +from datetime import datetime, timedelta, timezone, UTC +from pathlib import Path +import requests from abc import ABC, abstractmethod from dataclasses import dataclass @@ -77,11 +80,12 @@ class HelloTool(Tool): name = arguments.get("name", "stranger") return ToolResult(content=f"Hello, {name}!") + class CalculatorTool(Tool): @property def name(self) -> str: return "calculator" - + @property def description(self) -> str: return "Perform basic arithemtic" @@ -137,6 +141,233 @@ class CalculatorTool(Tool): return ToolResult(content=str(result)) + +class WeatherTool(Tool): + @property + def name(self) -> str: + return "weather" + + @property + def description(self) -> str: + return "Get current weather conditions for a location" + + @property + def schema(self) -> dict[str, object]: + return { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": ("The city or location to get weather for."), + }, + }, + "required": ["location"], + } + + def execute(self, arguments: dict[str, object]) -> ToolResult: + location = arguments.get("location") + + if not isinstance(location, str): + raise TypeError("location must be a string") + + geocoding_response = requests.get( + "https://geocoding-api.open-meteo.com/v1/search", + params={ + "name": location, + "count": 1, + "language": "en", + "format": "json", + }, + timeout=10, + ) + geocoding_response.raise_for_status() + + geocoding_data = geocoding_response.json() + results = geocoding_data.get("results") + + if not results: + raise ValueError(f"location not found: {location}") + + place = results[0] + + latitude = place["latitude"] + longitude = place["longitude"] + timezone_name = place["timezone"] + + forecast_response = requests.get( + "https://api.open-meteo.com/v1/forecast", + params={ + "latitude": latitude, + "longitude": longitude, + "hourly": ( + "temperature_2m," + "precipitation_probability," + "precipitation," + "weather_code," + "wind_speed_10m" + ), + "timezone": timezone_name, + "forecast_days": 2, + }, + timeout=10, + ) + forecast_response.raise_for_status() + + forecast = forecast_response.json() + hourly = forecast["hourly"] + + tomorrow = ( + (datetime.now(UTC) + timedelta(days=1)) + .astimezone(timezone(timedelta(seconds=forecast["utc_offset_seconds"]))) + .date() + ) + + temperatures = [] + precipitation_probabilities = [] + precipitation = [] + wind_speeds = [] + weather_codes = [] + + for index, timestamp in enumerate(hourly["time"]): + forecast_date = datetime.fromisoformat(timestamp).date() + + if forecast_date != tomorrow: + continue + + temperatures.append(hourly["temperature_2m"][index]) + precipitation_probabilities.append( + hourly["precipitation_probability"][index] + ) + precipitation.append(hourly["precipitation"][index]) + wind_speeds.append(hourly["wind_speed_10m"][index]) + weather_codes.append(hourly["weather_code"][index]) + + if not temperatures: + raise ValueError("no forecast data available for tomorrow") + return ToolResult( + content=( + f"Location: {place['name']}, {place['country']}\n" + f"Forecast date: {tomorrow}\n" + f"Temperature: {min(temperatures):.1f}–" + f"{max(temperatures):.1f} °C\n" + f"Maximum precipitation probability: " + f"{max(precipitation_probabilities)}%\n" + f"Total precipitation: {sum(precipitation):.1f} mm\n" + f"Maximum wind speed: {max(wind_speeds):.1f} km/h\n" + f"Weather codes: {sorted(set(weather_codes))}" + ) + ) + + +class FileSystemListTool(Tool): + @property + def name(self) -> str: + return "filesystem_list" + + @property + def description(self) -> str: + return "List files and directories inside the agent's sandbox." + + @property + def schema(self) -> dict[str, object]: + return { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": ( + "Relative path inside the filesystem sandbox. " + "Use '.' for the sandbox root." + ), + }, + }, + "required": ["path"], + } + + def execute(self, arguments: dict[str, object]) -> ToolResult: + path = arguments.get("path") + + if not isinstance(path, str): + raise TypeError("path must be a string") + + sandbox = Path.home() / "Projects" / "personal-agent" / "data" + target = (sandbox / path).resolve() + + if not target.is_relative_to(sandbox.resolve()): + raise ValueError("path is outside the filesystem sandbox") + + if not target.exists(): + raise ValueError(f"path does not exist: {path}") + + if not target.is_dir(): + raise ValueError(f"path is not a directory: {path}") + + entries = [] + + for entry in sorted(target.iterdir()): + kind = "directory" if entry.is_dir() else "file" + entries.append(f"{kind}: {entry.name}") + + if not entries: + return ToolResult(content="Directory is empty.") + + return ToolResult(content="\n".join(entries)) + + +class FileSystemReadTool(Tool): + MAX_FILE_SIZE = 1024 * 1024 # 1 MiB + + @property + def name(self) -> str: + return "filesystem_read" + + @property + def description(self) -> str: + return "Read the contents of a file inside the agent's sandbox." + + @property + def schema(self) -> dict[str, object]: + return { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": ("Relative path inside the filesystem sandbox."), + }, + }, + "required": ["path"], + } + + def execute(self, arguments: dict[str, object]) -> ToolResult: + path = arguments.get("path") + + if not isinstance(path, str): + raise TypeError("path must be a string") + + sandbox = Path.home() / "Projects" / "personal-agent" / "data" + sandbox = sandbox.resolve() + target = (sandbox / path).resolve() + + if not target.is_relative_to(sandbox): + raise ValueError("path is outside the filesystem sandbox") + + if not target.exists(): + raise ValueError(f"path does not exist: {path}") + + if not target.is_file(): + raise ValueError(f"path is not a file: {path}") + + if target.stat().st_size > self.MAX_FILE_SIZE: + raise ValueError("file is too large to read") + + try: + content = target.read_text(encoding="utf-8") + except UnicodeDecodeError as error: + raise ValueError("file is not valid UTF-8 text") from error + + return ToolResult(content=content) + + if __name__ == "__main__": registry = ToolRegistry() -- cgit v1.3