mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-05-19 16:12:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(config): add ModelPresetConfig and runtime preset switching

Browse Source 
- Add `ModelPresetConfig` schema for named model presets
- Add `model_presets` dict to `Config` and `model_preset` field to `AgentDefaults`
- Add `resolve_preset()` to return effective model params from preset or defaults
- Add `@model_validator` to reject unknown preset names
- Update `_match_provider()` to use resolved preset model/provider
- Update `make_provider()` and `provider_signature()` to use `resolve_preset()`
- Add `model_preset` property to `AgentLoop` for atomic runtime switching
- Update `AgentLoop.from_config()` to inject a runtime `default` preset
- Wire self-tool to inspect/clear preset state
- Update CLI display strings to show active preset
This commit is contained in:
chengyongru 2026-05-09 15:30:47 +08:00 committed by Xubin Ren
parent 1175420339
commit 6f78267c82
8 changed files with 348 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
nanobot/agent/loop.py
View File

@ -34,7 +34,7 @@ from nanobot.agent.tools.self import MyTool
from nanobot.bus.events import InboundMessage, OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.command import CommandContext, CommandRouter, register_builtin_commands
from nanobot.config.schema import AgentDefaults
from nanobot.config.schema import AgentDefaults, ModelPresetConfig
from nanobot.providers.base import LLMProvider
from nanobot.providers.factory import ProviderSnapshot
from nanobot.session.manager import Session, SessionManager
@ -291,6 +291,8 @@ class AgentLoop:
image_generation_provider_configs: dict[str, ProviderConfig] | None = None,
provider_snapshot_loader: Callable[[], ProviderSnapshot] | None = None,
provider_signature: tuple[object, ...] | None = None,
model_presets: dict[str, ModelPresetConfig] | None = None,
model_preset: str | None = None,
):
from nanobot.config.schema import ToolsConfig
@ -395,6 +397,8 @@ class AgentLoop:
provider=provider,
model=self.model,
)
self.model_presets: dict[str, ModelPresetConfig] = model_presets or {}
self._active_preset: str | None = model_preset if model_presets and model_preset in model_presets else None
self._register_default_tools()
self._runtime_vars: dict[str, Any] = {}
self._current_iteration: int = 0
@ -420,8 +424,12 @@ class AgentLoop:
bus = MessageBus()
defaults = config.agents.defaults
provider = extra.pop("provider", None) or make_provider(config)
model = extra.pop("model", None) or defaults.model
context_window_tokens = extra.pop("context_window_tokens", None) or defaults.context_window_tokens
resolved = config.resolve_preset()
model = extra.pop("model", None) or resolved.model
context_window_tokens = extra.pop("context_window_tokens", None) or resolved.context_window_tokens
model_presets = dict(config.model_presets)
if "default" not in model_presets:
model_presets["default"] = resolved
return cls(
bus=bus,
provider=provider,
@ -443,6 +451,8 @@ class AgentLoop:
consolidation_ratio=defaults.consolidation_ratio,
max_messages=defaults.max_messages,
tools_config=config.tools,
model_presets=model_presets,
model_preset=defaults.model_preset,
**extra,
)
@ -480,6 +490,25 @@ class AgentLoop:
return
self._apply_provider_snapshot(snapshot)
# -- model_preset property --
@property
def model_preset(self) -> str | None:
return self._active_preset
@model_preset.setter
def model_preset(self, name: str | None) -> None:
"""Resolve a preset by name and apply all fields atomically."""
if not isinstance(name, str) or not name.strip():
raise ValueError("model_preset must be a non-empty string")
if name not in self.model_presets:
raise KeyError(f"model_preset {name!r} not found. Available: {', '.join(self.model_presets) or '(none)'}")
p = self.model_presets[name]
self.model = p.model
self.context_window_tokens = p.context_window_tokens
self.provider.generation = p.to_generation_settings()
self._active_preset = name
def _register_default_tools(self) -> None:
"""Register the default set of tools via plugin loader."""
from nanobot.agent.tools.context import ToolContext

5
nanobot/agent/tools/runtime_state.py
View File

@ -52,3 +52,8 @@ class RuntimeState(Protocol):
def _last_usage(self) -> Any: ...
def _sync_subagent_runtime_limits(self) -> None: ...
@property
def model_preset(self) -> str | None: ...
_active_preset: str | None

9
nanobot/agent/tools/self.py
View File

@ -347,6 +347,7 @@ class MyTool(Tool, ContextAware):
# RESTRICTED keys
for k in self.RESTRICTED:
parts.append(self._format_value(getattr(state, k, None), k))
parts.append(self._format_value(state.model_preset, "model_preset"))
# Other useful top-level keys shown in description
for k in ("workspace", "provider_retry_mode", "max_tool_result_chars", "_current_iteration", "web_config", "exec_config", "subagents"):
if _has_real_attr(state, k):
@ -411,6 +412,8 @@ class MyTool(Tool, ContextAware):
if "min_len" in spec and len(str(value)) < spec["min_len"]:
return f"Error: '{key}' must be at least {spec['min_len']} characters"
setattr(self._runtime_state, key, value)
if key == "model":
self._runtime_state._active_preset = None
if key == "max_iterations" and hasattr(self._runtime_state, "_sync_subagent_runtime_limits"):
self._runtime_state._sync_subagent_runtime_limits()
self._audit("modify", f"{key}: {old!r} -> {value!r}")
@ -429,7 +432,11 @@ class MyTool(Tool, ContextAware):
f"REJECTED type mismatch {key}: expects {old_t.__name__}, got {new_t.__name__}",
)
return f"Error: '{key}' expects {old_t.__name__}, got {new_t.__name__}"
setattr(self._runtime_state, key, value)
try:
setattr(self._runtime_state, key, value)
except (ValueError, KeyError) as e:
self._audit("modify", f"REJECTED {key}: {e}")
return f"Error: {e}"
self._audit("modify", f"{key}: {old!r} -> {value!r}")
return f"Set {key} = {value!r} (was {old!r})"
if callable(value):

18
nanobot/cli/commands.py
View File

@ -448,6 +448,14 @@ def _onboard_plugins(config_path: Path) -> None:
json.dump(data, f, indent=2, ensure_ascii=False)
def _model_display(config: Config) -> tuple[str, str]:
"""Return (resolved_model_name, preset_tag) for display strings."""
resolved = config.resolve_preset()
name = config.agents.defaults.model_preset
tag = f" (preset: {name})" if name else ""
return resolved.model, tag
def _load_runtime_config(config: str | None = None, workspace: str | None = None) -> Config:
"""Load config and optionally override the active workspace."""
from nanobot.config.loader import load_config, resolve_config_env_vars, set_config_path
@ -556,10 +564,10 @@ def serve(
console.print(f"[red]Error: {exc}[/red]")
raise typer.Exit(1) from exc
model_name = runtime_config.agents.defaults.model
model_name, preset_tag = _model_display(runtime_config)
console.print(f"{__logo__} Starting OpenAI-compatible API server")
console.print(f" [cyan]Endpoint[/cyan] : http://{host}:{port}/v1/chat/completions")
console.print(f" [cyan]Model[/cyan] : {model_name}")
console.print(f" [cyan]Model[/cyan] : {model_name}{preset_tag}")
console.print(" [cyan]Session[/cyan] : api:default")
console.print(f" [cyan]Timeout[/cyan] : {timeout}s")
if host in {"0.0.0.0", "::"}:
@ -1086,7 +1094,8 @@ def agent(
# Interactive mode — route through bus like other channels
from nanobot.bus.events import InboundMessage
_init_prompt_session()
console.print(f"{__logo__} Interactive mode [bold blue]({config.agents.defaults.model})[/bold blue] — type [bold]exit[/bold] or [bold]Ctrl+C[/bold] to quit\n")
_model, _preset_tag = _model_display(config)
console.print(f"{__logo__} Interactive mode [bold blue]({_model})[/bold blue]{_preset_tag} — type [bold]exit[/bold] or [bold]Ctrl+C[/bold] to quit\n")
if ":" in session_id:
cli_channel, cli_chat_id = session_id.split(":", 1)
@ -1448,7 +1457,8 @@ def status():
if config_path.exists():
from nanobot.providers.registry import PROVIDERS
console.print(f"Model: {config.agents.defaults.model}")
_model, _preset_tag = _model_display(config)
console.print(f"Model: {_model}{_preset_tag}")
# Check API keys from registry
for spec in PROVIDERS:

47
nanobot/config/schema.py
View File

@ -4,7 +4,7 @@ from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal
from pydantic import AliasChoices, BaseModel, ConfigDict, Field
from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator
from pydantic.alias_generators import to_camel
from pydantic_settings import BaseSettings
@ -73,10 +73,30 @@ class DreamConfig(Base):
return f"every {hours}h"
class ModelPresetConfig(Base):
"""A named set of model + generation parameters for quick switching."""
model: str
provider: str = "auto"
max_tokens: int = 8192
context_window_tokens: int = 65_536
temperature: float = 0.1
reasoning_effort: str | None = None
def to_generation_settings(self) -> Any:
from nanobot.providers.base import GenerationSettings
return GenerationSettings(
temperature=self.temperature,
max_tokens=self.max_tokens,
reasoning_effort=self.reasoning_effort,
)
class AgentDefaults(Base):
"""Default agent configuration."""
workspace: str = "~/.nanobot/workspace"
model_preset: str | None = None # Active preset name — takes precedence over fields below
model: str = "anthropic/claude-opus-4-5"
provider: str = (
"auto" # Provider name (e.g. "anthropic", "openrouter") or "auto" for auto-detection
@ -254,6 +274,26 @@ class Config(BaseSettings):
api: ApiConfig = Field(default_factory=ApiConfig)
gateway: GatewayConfig = Field(default_factory=GatewayConfig)
tools: ToolsConfig = Field(default_factory=ToolsConfig)
model_presets: dict[str, ModelPresetConfig] = Field(default_factory=dict)
@model_validator(mode="after")
def _validate_model_preset(self) -> "Config":
name = self.agents.defaults.model_preset
if name and name not in self.model_presets:
raise ValueError(f"model_preset {name!r} not found in model_presets")
return self
def resolve_preset(self) -> ModelPresetConfig:
"""Return effective model params: from active preset, or individual defaults."""
name = self.agents.defaults.model_preset
if name:
return self.model_presets[name]
d = self.agents.defaults
return ModelPresetConfig(
model=d.model, provider=d.provider, max_tokens=d.max_tokens,
context_window_tokens=d.context_window_tokens,
temperature=d.temperature, reasoning_effort=d.reasoning_effort,
)
@property
def workspace_path(self) -> Path:
@ -266,7 +306,8 @@ class Config(BaseSettings):
"""Match provider config and its registry name. Returns (config, spec_name)."""
from nanobot.providers.registry import PROVIDERS, find_by_name
forced = self.agents.defaults.provider
resolved = self.resolve_preset()
forced = resolved.provider
if forced != "auto":
spec = find_by_name(forced)
if spec:
@ -274,7 +315,7 @@ class Config(BaseSettings):
return (p, spec.name) if p else (None, None)
return None, None
model_lower = (model or self.agents.defaults.model).lower()
model_lower = (model or resolved.model).lower()
model_normalized = model_lower.replace("-", "_")
model_prefix = model_lower.split("/", 1)[0] if "/" in model_lower else ""
normalized_prefix = model_prefix.replace("-", "_")

40
nanobot/providers/factory.py
View File

@ -6,7 +6,7 @@ from dataclasses import dataclass
from pathlib import Path
from nanobot.config.schema import Config
from nanobot.providers.base import GenerationSettings, LLMProvider
from nanobot.providers.base import LLMProvider
from nanobot.providers.registry import find_by_name
@ -20,7 +20,8 @@ class ProviderSnapshot:
def make_provider(config: Config) -> LLMProvider:
"""Create the LLM provider implied by config."""
model = config.agents.defaults.model
resolved = config.resolve_preset()
model = resolved.model
provider_name = config.get_provider_name(model)
p = config.get_provider(model)
spec = find_by_name(provider_name) if provider_name else None
@ -83,42 +84,37 @@ def make_provider(config: Config) -> LLMProvider:
extra_body=p.extra_body if p else None,
)
defaults = config.agents.defaults
provider.generation = GenerationSettings(
temperature=defaults.temperature,
max_tokens=defaults.max_tokens,
reasoning_effort=defaults.reasoning_effort,
)
provider.generation = resolved.to_generation_settings()
return provider
def provider_signature(config: Config) -> tuple[object, ...]:
"""Return the config fields that affect the primary LLM provider."""
model = config.agents.defaults.model
defaults = config.agents.defaults
p = config.get_provider(model)
resolved = config.resolve_preset()
p = config.get_provider(resolved.model)
return (
model,
defaults.provider,
config.get_provider_name(model),
config.get_api_key(model),
config.get_api_base(model),
resolved.model,
resolved.provider,
config.get_provider_name(resolved.model),
config.get_api_key(resolved.model),
config.get_api_base(resolved.model),
p.extra_headers if p else None,
p.extra_body if p else None,
getattr(p, "region", None) if p else None,
getattr(p, "profile", None) if p else None,
defaults.max_tokens,
defaults.temperature,
defaults.reasoning_effort,
defaults.context_window_tokens,
resolved.max_tokens,
resolved.temperature,
resolved.reasoning_effort,
resolved.context_window_tokens,
)
def build_provider_snapshot(config: Config) -> ProviderSnapshot:
resolved = config.resolve_preset()
return ProviderSnapshot(
provider=make_provider(config),
model=config.agents.defaults.model,
context_window_tokens=config.agents.defaults.context_window_tokens,
model=resolved.model,
context_window_tokens=resolved.context_window_tokens,
signature=provider_signature(config),
)

134
tests/agent/test_self_model_preset.py Normal file
View File

@ -0,0 +1,134 @@
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from nanobot.agent.loop import AgentLoop
from nanobot.agent.tools.self import MyTool
from nanobot.bus.queue import MessageBus
from nanobot.config.schema import ModelPresetConfig
def _provider(default_model: str, max_tokens: int = 123) -> MagicMock:
provider = MagicMock()
provider.get_default_model.return_value = default_model
provider.generation = SimpleNamespace(
max_tokens=max_tokens, temperature=0.1, reasoning_effort=None
)
return provider
def _make_loop(tmp_path, presets=None, active_preset=None):
provider = _provider("base-model")
return AgentLoop(
bus=MessageBus(),
provider=provider,
workspace=tmp_path,
model="base-model",
context_window_tokens=1000,
model_presets=presets or {},
model_preset=active_preset,
)
def test_model_preset_getter_none_when_not_set(tmp_path) -> None:
loop = _make_loop(tmp_path)
assert loop.model_preset is None
def test_model_preset_setter_updates_state(tmp_path) -> None:
presets = {
"fast": ModelPresetConfig(
model="openai/gpt-4.1",
provider="openai",
max_tokens=4096,
context_window_tokens=32_768,
temperature=0.5,
reasoning_effort="low",
)
}
loop = _make_loop(tmp_path, presets=presets)
loop.model_preset = "fast"
assert loop.model_preset == "fast"
assert loop.model == "openai/gpt-4.1"
assert loop.context_window_tokens == 32_768
assert loop.provider.generation.temperature == 0.5
assert loop.provider.generation.max_tokens == 4096
assert loop.provider.generation.reasoning_effort == "low"
def test_model_preset_setter_raises_on_unknown(tmp_path) -> None:
loop = _make_loop(tmp_path)
with pytest.raises(KeyError, match="model_preset 'missing' not found"):
loop.model_preset = "missing"
def test_model_preset_setter_raises_on_empty_string(tmp_path) -> None:
loop = _make_loop(tmp_path)
with pytest.raises(ValueError, match="model_preset must be a non-empty string"):
loop.model_preset = ""
def test_self_tool_inspect_shows_model_preset(tmp_path) -> None:
presets = {
"fast": ModelPresetConfig(model="openai/gpt-4.1"),
}
loop = _make_loop(tmp_path, presets=presets, active_preset="fast")
tool = MyTool(runtime_state=loop, modify_allowed=True)
output = tool._inspect_all()
assert "model_preset: 'fast'" in output
def test_self_tool_set_model_preset_via_modify(tmp_path) -> None:
presets = {
"fast": ModelPresetConfig(model="openai/gpt-4.1"),
}
loop = _make_loop(tmp_path, presets=presets)
tool = MyTool(runtime_state=loop, modify_allowed=True)
result = tool._modify("model_preset", "fast")
assert "Error" not in result
assert loop.model_preset == "fast"
assert loop.model == "openai/gpt-4.1"
def test_self_tool_set_model_clears_active_preset(tmp_path) -> None:
presets = {
"fast": ModelPresetConfig(model="openai/gpt-4.1"),
}
loop = _make_loop(tmp_path, presets=presets, active_preset="fast")
tool = MyTool(runtime_state=loop, modify_allowed=True)
result = tool._modify("model", "anthropic/claude-opus-4-5")
assert "Error" not in result
assert loop._active_preset is None
assert loop.model == "anthropic/claude-opus-4-5"
def test_from_config_injects_default_preset(tmp_path) -> None:
from unittest.mock import patch
from nanobot.config.schema import Config
config = Config.model_validate({
"agents": {"defaults": {"model": "openai/gpt-4.1", "workspace": str(tmp_path)}},
})
fake_provider = _provider("openai/gpt-4.1")
with patch("nanobot.providers.factory.make_provider", return_value=fake_provider):
loop = AgentLoop.from_config(config)
assert "default" in loop.model_presets
assert loop.model_presets["default"].model == "openai/gpt-4.1"
def test_from_config_preserves_existing_default_preset(tmp_path) -> None:
from unittest.mock import patch
from nanobot.config.schema import Config
config = Config.model_validate({
"agents": {"defaults": {"model": "openai/gpt-4.1", "workspace": str(tmp_path)}},
"model_presets": {
"default": {"model": "custom-model"}
},
})
fake_provider = _provider("openai/gpt-4.1")
with patch("nanobot.providers.factory.make_provider", return_value=fake_provider):
loop = AgentLoop.from_config(config)
assert loop.model_presets["default"].model == "custom-model"

93
tests/config/test_model_presets.py Normal file
View File

@ -0,0 +1,93 @@
from nanobot.config.schema import Config, ModelPresetConfig
def test_resolve_preset_returns_defaults_when_no_preset() -> None:
config = Config()
resolved = config.resolve_preset()
assert resolved.model == config.agents.defaults.model
assert resolved.provider == config.agents.defaults.provider
assert resolved.max_tokens == config.agents.defaults.max_tokens
assert resolved.context_window_tokens == config.agents.defaults.context_window_tokens
assert resolved.temperature == config.agents.defaults.temperature
assert resolved.reasoning_effort == config.agents.defaults.reasoning_effort
def test_resolve_preset_returns_active_preset() -> None:
config = Config.model_validate({
"model_presets": {
"fast": {
"model": "openai/gpt-4.1",
"provider": "openai",
"maxTokens": 4096,
"contextWindowTokens": 32_768,
"temperature": 0.5,
"reasoningEffort": "low",
}
},
"agents": {
"defaults": {
"modelPreset": "fast",
}
},
})
resolved = config.resolve_preset()
assert resolved.model == "openai/gpt-4.1"
assert resolved.provider == "openai"
assert resolved.max_tokens == 4096
assert resolved.context_window_tokens == 32_768
assert resolved.temperature == 0.5
assert resolved.reasoning_effort == "low"
def test_validator_rejects_unknown_preset() -> None:
import pytest
with pytest.raises(ValueError, match="model_preset 'unknown' not found in model_presets"):
Config.model_validate({
"agents": {
"defaults": {
"modelPreset": "unknown",
}
}
})
def test_match_provider_uses_preset_model() -> None:
config = Config.model_validate({
"providers": {
"openai": {"apiKey": "sk-test"},
},
"model_presets": {
"fast": {
"model": "openai/gpt-4.1",
"provider": "openai",
}
},
"agents": {
"defaults": {
"modelPreset": "fast",
}
},
})
name = config.get_provider_name()
assert name == "openai"
def test_match_provider_uses_preset_provider_when_forced() -> None:
config = Config.model_validate({
"providers": {
"anthropic": {"apiKey": "sk-test"},
},
"model_presets": {
"fast": {
"model": "anthropic/claude-opus-4-5",
"provider": "anthropic",
}
},
"agents": {
"defaults": {
"modelPreset": "fast",
}
},
})
name = config.get_provider_name()
assert name == "anthropic"