mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-05-20 00:22:31 +00:00
67875d7a15
The config field was added but never passed from config to AgentLoop. The value was always falling back to the default (40) regardless of what was set in config.json. Now passes tool_hint_max_length through all AgentLoop() call sites: - nanobot/nanobot.py (main bot) - nanobot/cli/commands.py (CLI agent, dev, webui commands) Also adds documentation in docs/configuration.md.
132 lines
4.2 KiB
Python
132 lines
4.2 KiB
Python
"""High-level programmatic interface to nanobot."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from nanobot.agent.hook import AgentHook, SDKCaptureHook
|
|
from nanobot.agent.loop import AgentLoop
|
|
from nanobot.bus.queue import MessageBus
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RunResult:
|
|
"""Result of a single agent run."""
|
|
|
|
content: str
|
|
tools_used: list[str]
|
|
messages: list[dict[str, Any]]
|
|
|
|
|
|
class Nanobot:
|
|
"""Programmatic facade for running the nanobot agent.
|
|
|
|
Usage::
|
|
|
|
bot = Nanobot.from_config()
|
|
result = await bot.run("Summarize this repo", hooks=[MyHook()])
|
|
print(result.content)
|
|
"""
|
|
|
|
def __init__(self, loop: AgentLoop) -> None:
|
|
self._loop = loop
|
|
|
|
@classmethod
|
|
def from_config(
|
|
cls,
|
|
config_path: str | Path | None = None,
|
|
*,
|
|
workspace: str | Path | None = None,
|
|
) -> Nanobot:
|
|
"""Create a Nanobot instance from a config file.
|
|
|
|
Args:
|
|
config_path: Path to ``config.json``. Defaults to
|
|
``~/.nanobot/config.json``.
|
|
workspace: Override the workspace directory from config.
|
|
"""
|
|
from nanobot.config.loader import load_config, resolve_config_env_vars
|
|
from nanobot.config.schema import Config
|
|
|
|
resolved: Path | None = None
|
|
if config_path is not None:
|
|
resolved = Path(config_path).expanduser().resolve()
|
|
if not resolved.exists():
|
|
raise FileNotFoundError(f"Config not found: {resolved}")
|
|
|
|
config: Config = resolve_config_env_vars(load_config(resolved))
|
|
if workspace is not None:
|
|
config.agents.defaults.workspace = str(
|
|
Path(workspace).expanduser().resolve()
|
|
)
|
|
|
|
provider = _make_provider(config)
|
|
bus = MessageBus()
|
|
defaults = config.agents.defaults
|
|
|
|
loop = AgentLoop(
|
|
bus=bus,
|
|
provider=provider,
|
|
workspace=config.workspace_path,
|
|
model=defaults.model,
|
|
max_iterations=defaults.max_tool_iterations,
|
|
context_window_tokens=defaults.context_window_tokens,
|
|
context_block_limit=defaults.context_block_limit,
|
|
max_tool_result_chars=defaults.max_tool_result_chars,
|
|
provider_retry_mode=defaults.provider_retry_mode,
|
|
tool_hint_max_length=defaults.tool_hint_max_length,
|
|
web_config=config.tools.web,
|
|
exec_config=config.tools.exec,
|
|
restrict_to_workspace=config.tools.restrict_to_workspace,
|
|
mcp_servers=config.tools.mcp_servers,
|
|
timezone=defaults.timezone,
|
|
unified_session=defaults.unified_session,
|
|
disabled_skills=defaults.disabled_skills,
|
|
session_ttl_minutes=defaults.session_ttl_minutes,
|
|
consolidation_ratio=defaults.consolidation_ratio,
|
|
tools_config=config.tools,
|
|
)
|
|
return cls(loop)
|
|
|
|
async def run(
|
|
self,
|
|
message: str,
|
|
*,
|
|
session_key: str = "sdk:default",
|
|
hooks: list[AgentHook] | None = None,
|
|
) -> RunResult:
|
|
"""Run the agent once and return the result.
|
|
|
|
Args:
|
|
message: The user message to process.
|
|
session_key: Session identifier for conversation isolation.
|
|
Different keys get independent history.
|
|
hooks: Optional lifecycle hooks for this run.
|
|
"""
|
|
capture = SDKCaptureHook()
|
|
prev = self._loop._extra_hooks
|
|
base_hooks = list(hooks) if hooks is not None else list(prev or [])
|
|
self._loop._extra_hooks = [capture, *base_hooks]
|
|
try:
|
|
response = await self._loop.process_direct(
|
|
message, session_key=session_key,
|
|
)
|
|
finally:
|
|
self._loop._extra_hooks = prev
|
|
|
|
content = (response.content if response else None) or ""
|
|
return RunResult(
|
|
content=content,
|
|
tools_used=capture.tools_used,
|
|
messages=capture.messages,
|
|
)
|
|
|
|
|
|
def _make_provider(config: Any) -> Any:
|
|
"""Create the LLM provider from config (extracted from CLI)."""
|
|
from nanobot.providers.factory import make_provider
|
|
|
|
return make_provider(config)
|