matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-05-21 15:14:45 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix: Override LLM brevity for mesh questions — give detailed data responses

Browse source 
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
K7ZVX 2026-05-05 05:29:09 +00:00
commit 45630f2cc6
2 changed files with 359 additions and 353 deletions
Download patch file Download diff file Expand all files Collapse all files

689
meshai/config.py
View file

@ -1,344 +1,345 @@
"""Configuration management for MeshAI.""" """Configuration management for MeshAI."""
import logging import logging
import os import os
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
from typing import Optional from typing import Optional
import yaml import yaml
_config_logger = logging.getLogger(__name__) _config_logger = logging.getLogger(__name__)
@dataclass @dataclass
class BotConfig: class BotConfig:
"""Bot identity and trigger settings.""" """Bot identity and trigger settings."""
name: str = "ai" name: str = "ai"
owner: str = "" owner: str = ""
respond_to_dms: bool = True respond_to_dms: bool = True
filter_bbs_protocols: bool = True filter_bbs_protocols: bool = True
@dataclass @dataclass
class ConnectionConfig: class ConnectionConfig:
"""Meshtastic connection settings.""" """Meshtastic connection settings."""
type: str = "serial" # serial or tcp type: str = "serial" # serial or tcp
serial_port: str = "/dev/ttyUSB0" serial_port: str = "/dev/ttyUSB0"
tcp_host: str = "192.168.1.100" tcp_host: str = "192.168.1.100"
tcp_port: int = 4403 tcp_port: int = 4403
@dataclass @dataclass
class ResponseConfig: class ResponseConfig:
"""Response behavior settings.""" """Response behavior settings."""
delay_min: float = 2.2 delay_min: float = 2.2
delay_max: float = 3.0 delay_max: float = 3.0
max_length: int = 150 max_length: int = 150
max_messages: int = 2 max_messages: int = 2
@dataclass @dataclass
class HistoryConfig: class HistoryConfig:
"""Conversation history settings.""" """Conversation history settings."""
database: str = "conversations.db" database: str = "conversations.db"
max_messages_per_user: int = 50 max_messages_per_user: int = 50
conversation_timeout: int = 86400 # 24 hours conversation_timeout: int = 86400 # 24 hours
# Cleanup settings # Cleanup settings
auto_cleanup: bool = True auto_cleanup: bool = True
cleanup_interval_hours: int = 24 cleanup_interval_hours: int = 24
max_age_days: int = 30 # Delete conversations older than this max_age_days: int = 30 # Delete conversations older than this
@dataclass @dataclass
class MemoryConfig: class MemoryConfig:
"""Rolling summary memory settings.""" """Rolling summary memory settings."""
enabled: bool = True # Enable memory optimization enabled: bool = True # Enable memory optimization
window_size: int = 4 # Recent message pairs to keep in full window_size: int = 4 # Recent message pairs to keep in full
summarize_threshold: int = 8 # Messages before re-summarizing summarize_threshold: int = 8 # Messages before re-summarizing
@dataclass @dataclass
class ContextConfig: class ContextConfig:
"""Passive mesh context settings.""" """Passive mesh context settings."""
enabled: bool = True enabled: bool = True
observe_channels: list[int] = field(default_factory=list) # Empty = all channels observe_channels: list[int] = field(default_factory=list) # Empty = all channels
ignore_nodes: list[str] = field(default_factory=list) # Node IDs to ignore ignore_nodes: list[str] = field(default_factory=list) # Node IDs to ignore
max_age: int = 2_592_000 # 30 days in seconds max_age: int = 2_592_000 # 30 days in seconds
max_context_items: int = 20 # Max observations injected into LLM context max_context_items: int = 20 # Max observations injected into LLM context
@dataclass @dataclass
class CommandsConfig: class CommandsConfig:
"""Command settings.""" """Command settings."""
enabled: bool = True enabled: bool = True
prefix: str = "!" prefix: str = "!"
disabled_commands: list[str] = field(default_factory=list) disabled_commands: list[str] = field(default_factory=list)
custom_commands: dict = field(default_factory=dict) custom_commands: dict = field(default_factory=dict)
@dataclass @dataclass
class LLMConfig: class LLMConfig:
"""LLM backend settings.""" """LLM backend settings."""
backend: str = "openai" # openai, anthropic, google backend: str = "openai" # openai, anthropic, google
api_key: str = "" api_key: str = ""
base_url: str = "https://api.openai.com/v1" base_url: str = "https://api.openai.com/v1"
model: str = "gpt-4o-mini" model: str = "gpt-4o-mini"
timeout: int = 30 timeout: int = 30
system_prompt: str = ( system_prompt: str = (
"YOUR COMMANDS (handled directly by you via DM):\n" "YOUR COMMANDS (handled directly by you via DM):\n"
"!help — List available commands.\n" "!help — List available commands.\n"
"!ping — Connectivity test, responds with pong.\n" "!ping — Connectivity test, responds with pong.\n"
"!status — Shows your version, uptime, user count, and message count.\n" "!status — Shows your version, uptime, user count, and message count.\n"
"!weather [location] — Weather lookup using Open-Meteo API.\n" "!weather [location] — Weather lookup using Open-Meteo API.\n"
"!reset — Clears conversation history and memory.\n" "!reset — Clears conversation history and memory.\n"
"!clear — Same as !reset.\n\n" "!clear — Same as !reset.\n\n"
"YOUR ARCHITECTURE: Modular Python — pluggable LLM backends (OpenAI, Anthropic, " "YOUR ARCHITECTURE: Modular Python — pluggable LLM backends (OpenAI, Anthropic, "
"Google, local), per-user SQLite conversation history, rolling summary memory, " "Google, local), per-user SQLite conversation history, rolling summary memory, "
"passive mesh context buffer (observes channel traffic), smart chunking for LoRa " "passive mesh context buffer (observes channel traffic), smart chunking for LoRa "
"message limits, prompt injection defense, advBBS filtering.\n\n" "message limits, prompt injection defense, advBBS filtering.\n\n"
"RESPONSE RULES:\n" "RESPONSE RULES:\n"
"- Keep responses very brief — 1-2 short sentences, under 300 characters. Only give longer answers if the user explicitly asks for detail or explanation.\n" "- For casual conversation, keep responses brief (1-2 sentences).\n"
"- Be concise but friendly. No markdown formatting.\n" "- For mesh health questions, give detailed data-driven responses.\n"
"- If asked about mesh activity and no recent traffic is shown, say you haven't " "- Be concise but friendly. No markdown formatting.\n"
"observed any yet.\n" "- If asked about mesh activity and no recent traffic is shown, say you haven't "
"- When asked about yourself or commands, answer conversationally. Don't dump lists.\n" "observed any yet.\n"
"- You are part of the freq51 mesh in the Twin Falls, Idaho area." "- When asked about yourself or commands, answer conversationally. Don't dump lists.\n"
) "- You are part of the freq51 mesh in the Twin Falls, Idaho area."
use_system_prompt: bool = True # Toggle to disable sending system prompt )
web_search: bool = False # Enable web search (Open WebUI feature) use_system_prompt: bool = True # Toggle to disable sending system prompt
google_grounding: bool = False # Enable Google Search grounding (Gemini only) web_search: bool = False # Enable web search (Open WebUI feature)
google_grounding: bool = False # Enable Google Search grounding (Gemini only)
@dataclass
class OpenMeteoConfig: @dataclass
"""Open-Meteo weather provider settings.""" class OpenMeteoConfig:
"""Open-Meteo weather provider settings."""
url: str = "https://api.open-meteo.com/v1"
url: str = "https://api.open-meteo.com/v1"
@dataclass
class WttrConfig: @dataclass
"""wttr.in weather provider settings.""" class WttrConfig:
"""wttr.in weather provider settings."""
url: str = "https://wttr.in"
url: str = "https://wttr.in"
@dataclass
class WeatherConfig: @dataclass
"""Weather command settings.""" class WeatherConfig:
"""Weather command settings."""
primary: str = "openmeteo" # openmeteo, wttr, llm
fallback: str = "llm" # openmeteo, wttr, llm, none primary: str = "openmeteo" # openmeteo, wttr, llm
default_location: str = "" fallback: str = "llm" # openmeteo, wttr, llm, none
openmeteo: OpenMeteoConfig = field(default_factory=OpenMeteoConfig) default_location: str = ""
wttr: WttrConfig = field(default_factory=WttrConfig) openmeteo: OpenMeteoConfig = field(default_factory=OpenMeteoConfig)
wttr: WttrConfig = field(default_factory=WttrConfig)
@dataclass
class MeshMonitorConfig: @dataclass
"""MeshMonitor trigger sync settings.""" class MeshMonitorConfig:
"""MeshMonitor trigger sync settings."""
enabled: bool = False
url: str = "" # e.g., http://100.64.0.11:3333 enabled: bool = False
inject_into_prompt: bool = True # Tell LLM about MeshMonitor commands url: str = "" # e.g., http://100.64.0.11:3333
refresh_interval: int = 300 # Seconds between refreshes inject_into_prompt: bool = True # Tell LLM about MeshMonitor commands
refresh_interval: int = 300 # Seconds between refreshes
@dataclass
class KnowledgeConfig: @dataclass
"""FTS5 knowledge base settings.""" class KnowledgeConfig:
"""FTS5 knowledge base settings."""
enabled: bool = False
db_path: str = "" enabled: bool = False
top_k: int = 5 db_path: str = ""
top_k: int = 5
@dataclass
class MeshSourceConfig: @dataclass
"""Configuration for a mesh data source.""" class MeshSourceConfig:
"""Configuration for a mesh data source."""
name: str = ""
type: str = "" # "meshview" or "meshmonitor" name: str = ""
url: str = "" type: str = "" # "meshview" or "meshmonitor"
api_token: str = "" # MeshMonitor only, supports ${ENV_VAR} url: str = ""
refresh_interval: int = 300 api_token: str = "" # MeshMonitor only, supports ${ENV_VAR}
enabled: bool = True refresh_interval: int = 300
enabled: bool = True
@dataclass
class RegionAnchor: @dataclass
"""A fixed region anchor point.""" class RegionAnchor:
"""A fixed region anchor point."""
name: str = ""
lat: float = 0.0 name: str = ""
lon: float = 0.0 lat: float = 0.0
lon: float = 0.0
@dataclass
class MeshIntelligenceConfig: @dataclass
"""Mesh intelligence and health scoring settings.""" class MeshIntelligenceConfig:
"""Mesh intelligence and health scoring settings."""
enabled: bool = False
regions: list[RegionAnchor] = field(default_factory=list) # Fixed region anchors enabled: bool = False
locality_radius_miles: float = 8.0 # Radius for locality clustering within regions regions: list[RegionAnchor] = field(default_factory=list) # Fixed region anchors
offline_threshold_hours: int = 24 # Hours before node considered offline locality_radius_miles: float = 8.0 # Radius for locality clustering within regions
packet_threshold: int = 500 # Non-text packets per 24h to flag offline_threshold_hours: int = 24 # Hours before node considered offline
battery_warning_percent: int = 20 # Battery level for warnings packet_threshold: int = 500 # Non-text packets per 24h to flag
battery_warning_percent: int = 20 # Battery level for warnings
@dataclass
class Config: @dataclass
"""Main configuration container.""" class Config:
"""Main configuration container."""
bot: BotConfig = field(default_factory=BotConfig)
connection: ConnectionConfig = field(default_factory=ConnectionConfig) bot: BotConfig = field(default_factory=BotConfig)
response: ResponseConfig = field(default_factory=ResponseConfig) connection: ConnectionConfig = field(default_factory=ConnectionConfig)
history: HistoryConfig = field(default_factory=HistoryConfig) response: ResponseConfig = field(default_factory=ResponseConfig)
memory: MemoryConfig = field(default_factory=MemoryConfig) history: HistoryConfig = field(default_factory=HistoryConfig)
context: ContextConfig = field(default_factory=ContextConfig) memory: MemoryConfig = field(default_factory=MemoryConfig)
commands: CommandsConfig = field(default_factory=CommandsConfig) context: ContextConfig = field(default_factory=ContextConfig)
llm: LLMConfig = field(default_factory=LLMConfig) commands: CommandsConfig = field(default_factory=CommandsConfig)
weather: WeatherConfig = field(default_factory=WeatherConfig) llm: LLMConfig = field(default_factory=LLMConfig)
meshmonitor: MeshMonitorConfig = field(default_factory=MeshMonitorConfig) weather: WeatherConfig = field(default_factory=WeatherConfig)
knowledge: KnowledgeConfig = field(default_factory=KnowledgeConfig) meshmonitor: MeshMonitorConfig = field(default_factory=MeshMonitorConfig)
mesh_sources: list[MeshSourceConfig] = field(default_factory=list) knowledge: KnowledgeConfig = field(default_factory=KnowledgeConfig)
mesh_intelligence: MeshIntelligenceConfig = field(default_factory=MeshIntelligenceConfig) mesh_sources: list[MeshSourceConfig] = field(default_factory=list)
mesh_intelligence: MeshIntelligenceConfig = field(default_factory=MeshIntelligenceConfig)
_config_path: Optional[Path] = field(default=None, repr=False)
_config_path: Optional[Path] = field(default=None, repr=False)
def resolve_api_key(self) -> str:
"""Resolve API key from config or environment.""" def resolve_api_key(self) -> str:
if self.llm.api_key: """Resolve API key from config or environment."""
# Check if it's an env var reference like ${LLM_API_KEY} if self.llm.api_key:
if self.llm.api_key.startswith("${") and self.llm.api_key.endswith("}"): # Check if it's an env var reference like ${LLM_API_KEY}
env_var = self.llm.api_key[2:-1] if self.llm.api_key.startswith("${") and self.llm.api_key.endswith("}"):
return os.environ.get(env_var, "") env_var = self.llm.api_key[2:-1]
return self.llm.api_key return os.environ.get(env_var, "")
# Fall back to common env vars return self.llm.api_key
for env_var in ["LLM_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY"]: # Fall back to common env vars
if value := os.environ.get(env_var): for env_var in ["LLM_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY"]:
return value if value := os.environ.get(env_var):
return "" return value
return ""
def _dict_to_dataclass(cls, data: dict):
"""Recursively convert dict to dataclass, handling nested structures.""" def _dict_to_dataclass(cls, data: dict):
if data is None: """Recursively convert dict to dataclass, handling nested structures."""
return cls() if data is None:
return cls()
field_types = {f.name: f.type for f in cls.__dataclass_fields__.values()}
kwargs = {} field_types = {f.name: f.type for f in cls.__dataclass_fields__.values()}
kwargs = {}
for key, value in data.items():
if key.startswith("_"): for key, value in data.items():
continue if key.startswith("_"):
if key not in field_types: continue
continue if key not in field_types:
continue
field_type = field_types[key]
field_type = field_types[key]
# Handle nested dataclasses
if hasattr(field_type, "__dataclass_fields__") and isinstance(value, dict): # Handle nested dataclasses
kwargs[key] = _dict_to_dataclass(field_type, value) if hasattr(field_type, "__dataclass_fields__") and isinstance(value, dict):
# Handle list of MeshSourceConfig kwargs[key] = _dict_to_dataclass(field_type, value)
elif key == "mesh_sources" and isinstance(value, list): # Handle list of MeshSourceConfig
kwargs[key] = [ elif key == "mesh_sources" and isinstance(value, list):
_dict_to_dataclass(MeshSourceConfig, item) kwargs[key] = [
if isinstance(item, dict) else item _dict_to_dataclass(MeshSourceConfig, item)
for item in value if isinstance(item, dict) else item
] for item in value
# Handle list of RegionAnchor ]
elif key == "regions" and isinstance(value, list): # Handle list of RegionAnchor
kwargs[key] = [ elif key == "regions" and isinstance(value, list):
_dict_to_dataclass(RegionAnchor, item) kwargs[key] = [
if isinstance(item, dict) else item _dict_to_dataclass(RegionAnchor, item)
for item in value if isinstance(item, dict) else item
] for item in value
else: ]
kwargs[key] = value else:
kwargs[key] = value
return cls(**kwargs)
return cls(**kwargs)
def _dataclass_to_dict(obj) -> dict:
"""Recursively convert dataclass to dict for YAML serialization.""" def _dataclass_to_dict(obj) -> dict:
if not hasattr(obj, "__dataclass_fields__"): """Recursively convert dataclass to dict for YAML serialization."""
return obj if not hasattr(obj, "__dataclass_fields__"):
return obj
result = {}
for field_name in obj.__dataclass_fields__: result = {}
if field_name.startswith("_"): for field_name in obj.__dataclass_fields__:
continue if field_name.startswith("_"):
value = getattr(obj, field_name) continue
if hasattr(value, "__dataclass_fields__"): value = getattr(obj, field_name)
result[field_name] = _dataclass_to_dict(value) if hasattr(value, "__dataclass_fields__"):
elif isinstance(value, list): result[field_name] = _dataclass_to_dict(value)
# Handle list of dataclasses (like mesh_sources) elif isinstance(value, list):
result[field_name] = [ # Handle list of dataclasses (like mesh_sources)
_dataclass_to_dict(item) if hasattr(item, "__dataclass_fields__") else item result[field_name] = [
for item in value _dataclass_to_dict(item) if hasattr(item, "__dataclass_fields__") else item
] for item in value
else: ]
result[field_name] = value else:
return result result[field_name] = value
return result
def load_config(config_path: Optional[Path] = None) -> Config:
"""Load configuration from YAML file. def load_config(config_path: Optional[Path] = None) -> Config:
"""Load configuration from YAML file.
Args:
config_path: Path to config file. Defaults to ./config.yaml Args:
config_path: Path to config file. Defaults to ./config.yaml
Returns:
Config object with loaded settings Returns:
""" Config object with loaded settings
if config_path is None: """
config_path = Path("config.yaml") if config_path is None:
config_path = Path("config.yaml")
config_path = Path(config_path)
config_path = Path(config_path)
if not config_path.exists():
# Return default config if file doesn't exist if not config_path.exists():
config = Config() # Return default config if file doesn't exist
config._config_path = config_path config = Config()
return config config._config_path = config_path
return config
with open(config_path, "r") as f:
data = yaml.safe_load(f) or {} with open(config_path, "r") as f:
data = yaml.safe_load(f) or {}
config = _dict_to_dataclass(Config, data)
config._config_path = config_path config = _dict_to_dataclass(Config, data)
return config config._config_path = config_path
return config
def save_config(config: Config, config_path: Optional[Path] = None) -> None:
"""Save configuration to YAML file. def save_config(config: Config, config_path: Optional[Path] = None) -> None:
"""Save configuration to YAML file.
Args:
config: Config object to save Args:
config_path: Path to save to. Uses config._config_path if not specified config: Config object to save
""" config_path: Path to save to. Uses config._config_path if not specified
if config_path is None: """
config_path = config._config_path or Path("config.yaml") if config_path is None:
config_path = config._config_path or Path("config.yaml")
config_path = Path(config_path)
config_path = Path(config_path)
data = _dataclass_to_dict(config)
data = _dataclass_to_dict(config)
# Add header comment
header = "# MeshAI Configuration\n# Generated by meshai --config\n\n" # Add header comment
header = "# MeshAI Configuration\n# Generated by meshai --config\n\n"
with open(config_path, "w") as f:
f.write(header) with open(config_path, "w") as f:
yaml.dump(data, f, default_flow_style=False, sort_keys=False, allow_unicode=True) f.write(header)
yaml.dump(data, f, default_flow_style=False, sort_keys=False, allow_unicode=True)

23
meshai/router.py
View file

@ -125,14 +125,19 @@ _CITY_TO_REGION = {
# Mesh awareness instruction for LLM # Mesh awareness instruction for LLM
_MESH_AWARENESS_PROMPT = """ _MESH_AWARENESS_PROMPT = """
When the user asks about mesh health, network status, or optimization: MESH DATA RESPONSE RULES (these OVERRIDE the brevity rules above for mesh questions):
- Use the LIVE MESH HEALTH DATA injected above to answer with real numbers - When answering about mesh health, nodes, coverage, or network status: give DETAILED responses
- Be specific: name nodes, cite utilization percentages, reference actual scores - Include actual numbers: scores, percentages, node names, packet counts, battery levels
- Give actionable recommendations based on the data - Use the data injected above — don't summarize it to one sentence
- If asked about a region or node you have detail for, use that detail - Structure your response with the key data points the user asked about
- If asked about something the data doesn't cover, say so - don't fabricate - For node questions: include hardware, region, battery, channel utilization, coverage, neighbors, packets
- Keep responses concise - these go over LoRa with limited message size - For region questions: include score, infrastructure status, coverage breakdown, flagged nodes, environment
- Users can run !health for a quick mesh summary or !region [name] for regional info - For mesh questions: include overall score by pillar, regional breakdown, top issues, coverage gaps
- For coverage questions: break down by region showing node counts, avg gateways, single-gateway nodes
- For "where do we need infrastructure": name specific regions with poor coverage, how many nodes are affected
- You CAN use 3-5 messages if needed — LoRa chunking handles splitting
- Be specific and data-driven, not vague summaries
- Still no markdown formatting — plain text only
""" """
@ -673,4 +678,4 @@ class MessageRouter:
connector=self.connector, connector=self.connector,
history=self.history, history=self.history,
) )