wylab/nanobot
1
0
Fork 0
Code Issues 5 Pull Requests 6 Actions Packages Projects Releases Wiki Activity

Rebase onto upstream (a4d95fd) #12

Closed
wylab wants to merge 144 commits from rebase-onto-upstream into main
pull from: rebase-onto-upstream
merge into: :main
Conversation 0 Commits 144 Files Changed 81
+197 -102
6 changed files with 197 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 4c32ecf114 - Show all commits
nanobot/agent/loop.py
+97 -30
View File
@@ -29,7 +29,7 @@ from nanobot.session.manager import SessionManager
class AgentLoop:
"""
The agent loop is the core processing engine.
It:
1. Receives messages from the bus
2. Builds context with history, memory, skills
@@ -37,6 +37,22 @@ class AgentLoop:
4. Executes tool calls
5. Sends responses back
"""
# Server-side context management: Anthropic trims old tool results and preserves all
# thinking blocks (keep="all" maximises cache hits). Client keeps full history.
CONTEXT_MANAGEMENT = {
"edits": [
{
"type": "clear_thinking_20251015",
"keep": "all", # Preserve all thinking blocks for cache reuse
},
{
"type": "clear_tool_uses_20250919",
"trigger": {"type": "input_tokens", "value": 80000},
"keep": {"type": "tool_uses", "value": 5},
},
]
}
def __init__(
self,
@@ -275,10 +291,6 @@ class AgentLoop:
status = self._get_quota_status()
return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id, content=status)
# Consolidate memory before processing if session is too large
if len(session.messages) > self.memory_window:
await self._consolidate_memory(session)
# Update tool contexts
message_tool = self.tools.get("message")
if isinstance(message_tool, MessageTool):
@@ -325,6 +337,8 @@ class AgentLoop:
channel=msg.channel,
chat_id=msg.chat_id,
)
# Mark where the current turn starts so we can slice the tool chain for storage
turn_start = len(messages)
# Select model based on quota
selected_model = self._select_model_based_on_quota()
@@ -332,7 +346,7 @@ class AgentLoop:
# Agent loop
iteration = 0
final_content = None
tools_used: list[str] = []
final_reasoning = None
while iteration < self.max_iterations:
iteration += 1
@@ -342,9 +356,10 @@ class AgentLoop:
response = await self.provider.chat(
messages=messages,
tools=self.tools.get_definitions(),
model=selected_model
model=selected_model,
context_management=self.CONTEXT_MANAGEMENT,
)
# Handle tool calls
if response.has_tool_calls:
# Add assistant message with tool calls
@@ -363,10 +378,9 @@ class AgentLoop:
messages, response.content, tool_call_dicts,
reasoning_content=response.reasoning_content,
)
# Execute tools
for tool_call in response.tool_calls:
tools_used.append(tool_call.name)
args_str = json.dumps(tool_call.arguments, ensure_ascii=False)
logger.info(f"Tool call: {tool_call.name}({args_str[:200]})")
result = await self.tools.execute(tool_call.name, tool_call.arguments)
@@ -379,24 +393,31 @@ class AgentLoop:
else:
# No tool calls, we're done
final_content = response.content
final_reasoning = response.reasoning_content
break
if final_content is None:
if iteration >= self.max_iterations:
final_content = f"Reached {self.max_iterations} iterations without completion."
else:
final_content = "I've completed processing but have no response to give."
# Log response preview
preview = final_content[:120] + "..." if len(final_content) > 120 else final_content
logger.info(f"Response to {msg.channel}:{msg.sender_id}: {preview}")
# Save to session (include tool names so consolidation sees what happened)
# Append final assistant response to messages so it's captured in the tool chain slice
messages = self.context.add_assistant_message(
messages, final_content, None,
reasoning_content=final_reasoning,
)
# Save to session: user message + full tool chain (tool_use, tool_results, thinking, final reply)
# Store current_message (not msg.content) so the time prefix is preserved
# and cache keys match on subsequent turns
session.add_message("user", current_message)
session.add_message("assistant", final_content,
tools_used=tools_used if tools_used else None)
for chain_msg in messages[turn_start:]:
session.add_raw_message(chain_msg)
self.sessions.save(session)
return OutboundMessage(
@@ -449,10 +470,12 @@ class AgentLoop:
channel=origin_channel,
chat_id=origin_chat_id,
)
turn_start = len(messages)
# Agent loop (limited for announce handling)
iteration = 0
final_content = None
final_reasoning = None
# Select model based on quota
selected_model = self._select_model_based_on_quota()
@@ -463,9 +486,10 @@ class AgentLoop:
response = await self.provider.chat(
messages=messages,
tools=self.tools.get_definitions(),
model=selected_model
model=selected_model,
context_management=self.CONTEXT_MANAGEMENT,
)
if response.has_tool_calls:
tool_call_dicts = [
{
@@ -482,7 +506,7 @@ class AgentLoop:
messages, response.content, tool_call_dicts,
reasoning_content=response.reasoning_content,
)
for tool_call in response.tool_calls:
args_str = json.dumps(tool_call.arguments, ensure_ascii=False)
logger.info(f"Tool call: {tool_call.name}({args_str[:200]})")
@@ -495,14 +519,22 @@ class AgentLoop:
messages.append({"role": "user", "content": "Reflect on the results and decide next steps."})
else:
final_content = response.content
final_reasoning = response.reasoning_content
break
if final_content is None:
final_content = "Background task completed."
# Save to session (mark as system message in history)
# Append final assistant response to messages
messages = self.context.add_assistant_message(
messages, final_content, None,
reasoning_content=final_reasoning,
)
# Save to session: user message + full tool chain
session.add_message("user", f"[System: {msg.sender_id}] {msg.content}")
session.add_message("assistant", final_content)
for chain_msg in messages[turn_start:]:
session.add_raw_message(chain_msg)
self.sessions.save(session)
return OutboundMessage(
@@ -512,7 +544,11 @@ class AgentLoop:
)
async def _consolidate_memory(self, session, archive_all: bool = False) -> None:
"""Consolidate old messages into MEMORY.md + HISTORY.md, then trim session."""
"""Consolidate session into MEMORY.md + HISTORY.md.
Context window management is now handled server-side via context_management.
This only runs on /new to write long-term facts and searchable history.
"""
if not session.messages:
return
memory = MemoryStore(self.workspace)
@@ -520,19 +556,50 @@ class AgentLoop:
old_messages = session.messages
keep_count = 0
else:
# Only write truly old messages; keep the recent ones
keep_count = min(10, max(2, self.memory_window // 2))
old_messages = session.messages[:-keep_count]
if not old_messages:
return
logger.info(f"Memory consolidation started: {len(session.messages)} messages, archiving {len(old_messages)}, keeping {keep_count}")
logger.info(f"Memory consolidation: archiving {len(old_messages)} messages, keeping {keep_count}")
# Format messages for LLM (include tool names when available)
# Format messages for LLM — handle full tool chain format
lines = []
for m in old_messages:
if not m.get("content"):
role = m.get("role", "?")
content = m.get("content")
timestamp = m.get("timestamp", "?")[:16]
if role == "tool":
result = str(content or "")[:200]
lines.append(f"[{timestamp}] TOOL_RESULT({m.get('name', '?')}): {result}")
continue
tools = f" [tools: {', '.join(m['tools_used'])}]" if m.get("tools_used") else ""
lines.append(f"[{m.get('timestamp', '?')[:16]}] {m['role'].upper()}{tools}: {m['content']}")
# Skip internal reflect prompts
if role == "user" and content == "Reflect on the results and decide next steps.":
continue
# Extract text from content (may be list of blocks)
if isinstance(content, list):
text_parts = [b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text"]
content_str = " ".join(text_parts)
elif isinstance(content, str):
content_str = content
else:
content_str = ""
# Get tool names from tool_calls or legacy tools_used
tool_names = []
if m.get("tool_calls"):
tool_names = [tc.get("function", {}).get("name", "?") for tc in m["tool_calls"]]
elif m.get("tools_used"):
tool_names = m["tools_used"]
if not content_str and not tool_names:
continue
tools_str = f" [tools: {', '.join(tool_names)}]" if tool_names else ""
lines.append(f"[{timestamp}] {role.upper()}{tools_str}: {content_str}")
conversation = "\n".join(lines)
current_memory = memory.read_long_term()
nanobot/providers/anthropic_oauth.py
+25 -1
View File
@@ -226,6 +226,7 @@ class AnthropicOAuthProvider(LLMProvider):
temperature: float = 0.7,
tools: list[dict[str, Any]] | None = None,
thinking_budget_override: int | None = None,
context_management: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Make request to Anthropic API."""
client = await self._get_client()
@@ -271,11 +272,16 @@ class AnthropicOAuthProvider(LLMProvider):
cached_tools[-1] = {**cached_tools[-1], "cache_control": {"type": "ephemeral", "ttl": "1h"}}
payload["tools"] = cached_tools
if context_management:
payload["context_management"] = context_management
edit_types = [e.get("type") for e in (context_management or {}).get("edits", [])]
logger.info(
"Anthropic request: model={} max_tokens={} thinking={} tools={}",
"Anthropic request: model={} max_tokens={} thinking={} tools={} context_mgmt={}",
payload.get("model"), payload.get("max_tokens"),
payload.get("thinking", "disabled"),
len(payload.get("tools", [])),
edit_types or "none",
)
response = await client.post(
@@ -331,6 +337,7 @@ class AnthropicOAuthProvider(LLMProvider):
max_tokens: int = 4096,
temperature: float = 0.7,
thinking_budget: int | None = None,
context_management: dict[str, Any] | None = None,
) -> LLMResponse:
"""Send chat completion request to Anthropic API."""
model = model or self.default_model
@@ -357,6 +364,7 @@ class AnthropicOAuthProvider(LLMProvider):
temperature=temperature,
tools=anthropic_tools,
thinking_budget_override=effective_thinking,
context_management=context_management,
)
return self._parse_response(response)
except Exception as e:
@@ -410,6 +418,22 @@ class AnthropicOAuthProvider(LLMProvider):
cache_write, cache_read,
)
# Log context editing activity if any edits were applied
if applied_edits := response.get("context_management", {}).get("applied_edits"):
for edit in applied_edits:
edit_type = edit.get("type", "?")
cleared_tokens = edit.get("cleared_input_tokens", 0)
if edit_type == "clear_tool_uses_20250919":
logger.info(
"Context edit: cleared {} tool uses ({} tokens)",
edit.get("cleared_tool_uses", 0), cleared_tokens,
)
elif edit_type == "clear_thinking_20251015":
logger.info(
"Context edit: cleared {} thinking turns ({} tokens)",
edit.get("cleared_thinking_turns", 0), cleared_tokens,
)
return LLMResponse(
content=text_content or None,
tool_calls=tool_calls,
nanobot/providers/base.py
+1
View File
@@ -89,6 +89,7 @@ class LLMProvider(ABC):
max_tokens: int = 4096,
temperature: float = 0.7,
thinking_budget: int | None = None,
context_management: dict[str, Any] | None = None,
) -> LLMResponse:
"""
Send a chat completion request.
nanobot/providers/litellm_provider.py
+1
View File
@@ -179,6 +179,7 @@ class LiteLLMProvider(LLMProvider):
max_tokens: int = 4096,
temperature: float = 0.7,
thinking_budget: int | None = None,
context_management: dict[str, Any] | None = None, # Anthropic-only, ignored here
) -> LLMResponse:
"""
Send a chat completion request via LiteLLM.
nanobot/providers/oauth_utils.py
+1 -1
View File
@@ -28,7 +28,7 @@ def get_auth_headers(token: str, is_oauth: bool = False) -> dict[str, str]:
if is_oauth:
headers["Authorization"] = f"Bearer {token}"
# Required headers to mimic Claude Code client
headers["anthropic-beta"] = "claude-code-20250219,oauth-2025-04-20"
headers["anthropic-beta"] = "claude-code-20250219,oauth-2025-04-20,context-management-2025-06-27"
headers["anthropic-dangerous-direct-browser-access"] = "true"
headers["user-agent"] = "claude-cli/2.1.2 (external, cli)"
headers["x-app"] = "cli"
nanobot/session/manager.py
+72 -70
View File
@@ -1,7 +1,6 @@
"""Session management for conversation history."""
import json
import shutil
from pathlib import Path
from dataclasses import dataclass, field
from datetime import datetime
@@ -16,20 +15,15 @@ from nanobot.utils.helpers import ensure_dir, safe_filename
class Session:
"""
A conversation session.
Stores messages in JSONL format for easy reading and persistence.
Important: Messages are append-only for LLM cache efficiency.
The consolidation process writes summaries to MEMORY.md/HISTORY.md
but does NOT modify the messages list or get_history() output.
"""
key: str # channel:chat_id
messages: list[dict[str, Any]] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
metadata: dict[str, Any] = field(default_factory=dict)
last_consolidated: int = 0 # Number of messages already consolidated to files
def add_message(self, role: str, content: str, **kwargs: Any) -> None:
"""Add a message to the session."""
@@ -41,56 +35,57 @@ class Session:
}
self.messages.append(msg)
self.updated_at = datetime.now()
def get_history(self, max_messages: int = 500) -> list[dict[str, Any]]:
"""Return unconsolidated messages for LLM input, aligned to a user turn."""
unconsolidated = self.messages[self.last_consolidated:]
sliced = unconsolidated[-max_messages:]
# Drop leading non-user messages to avoid orphaned tool_result blocks
for i, m in enumerate(sliced):
if m.get("role") == "user":
sliced = sliced[i:]
break
def add_raw_message(self, msg: dict[str, Any]) -> None:
"""Add a pre-formed message dict to the session, preserving all fields."""
stored = dict(msg)
if "timestamp" not in stored:
stored["timestamp"] = datetime.now().isoformat()
self.messages.append(stored)
self.updated_at = datetime.now()
out: list[dict[str, Any]] = []
for m in sliced:
entry: dict[str, Any] = {"role": m["role"], "content": m.get("content", "")}
for k in ("tool_calls", "tool_call_id", "name"):
if k in m:
entry[k] = m[k]
out.append(entry)
return out
# Fields that are valid in the Anthropic/OpenAI messages API.
# Everything else (timestamp, tools_used, etc.) is internal metadata.
_API_FIELDS = {"role", "content", "tool_calls", "tool_call_id", "name", "reasoning_content"}
def get_history(self, max_messages: int = 50) -> list[dict[str, Any]]:
"""
Get message history for LLM context.
Args:
max_messages: Maximum messages to return.
Returns:
List of messages in LLM format (API-relevant fields only).
"""
recent = self.messages[-max_messages:] if len(self.messages) > max_messages else self.messages
return [
{k: v for k, v in m.items() if k in self._API_FIELDS and v is not None}
for m in recent
]
def clear(self) -> None:
"""Clear all messages and reset session to initial state."""
"""Clear all messages in the session."""
self.messages = []
self.last_consolidated = 0
self.updated_at = datetime.now()
class SessionManager:
"""
Manages conversation sessions.
Sessions are stored as JSONL files in the sessions directory.
"""
def __init__(self, workspace: Path):
self.workspace = workspace
self.sessions_dir = ensure_dir(self.workspace / "sessions")
self.legacy_sessions_dir = Path.home() / ".nanobot" / "sessions"
self.sessions_dir = ensure_dir(Path.home() / ".nanobot" / "sessions")
self._cache: dict[str, Session] = {}
def _get_session_path(self, key: str) -> Path:
"""Get the file path for a session."""
safe_key = safe_filename(key.replace(":", "_"))
return self.sessions_dir / f"{safe_key}.jsonl"
def _get_legacy_session_path(self, key: str) -> Path:
"""Legacy global session path (~/.nanobot/sessions/)."""
safe_key = safe_filename(key.replace(":", "_"))
return self.legacy_sessions_dir / f"{safe_key}.jsonl"
def get_or_create(self, key: str) -> Session:
"""
@@ -102,9 +97,11 @@ class SessionManager:
Returns:
The session.
"""
# Check cache
if key in self._cache:
return self._cache[key]
# Try to load from disk
session = self._load(key)
if session is None:
session = Session(key=key)
@@ -115,72 +112,78 @@ class SessionManager:
def _load(self, key: str) -> Session | None:
"""Load a session from disk."""
path = self._get_session_path(key)
if not path.exists():
legacy_path = self._get_legacy_session_path(key)
if legacy_path.exists():
try:
shutil.move(str(legacy_path), str(path))
logger.info("Migrated session {} from legacy path", key)
except Exception:
logger.exception("Failed to migrate session {}", key)
if not path.exists():
return None
try:
messages = []
metadata = {}
created_at = None
last_consolidated = 0
with open(path, encoding="utf-8") as f:
with open(path) as f:
for line in f:
line = line.strip()
if not line:
continue
data = json.loads(line)
if data.get("_type") == "metadata":
metadata = data.get("metadata", {})
created_at = datetime.fromisoformat(data["created_at"]) if data.get("created_at") else None
last_consolidated = data.get("last_consolidated", 0)
else:
messages.append(data)
return Session(
key=key,
messages=messages,
created_at=created_at or datetime.now(),
metadata=metadata,
last_consolidated=last_consolidated
metadata=metadata
)
except Exception as e:
logger.warning("Failed to load session {}: {}", key, e)
logger.warning(f"Failed to load session {key}: {e}")
return None
def save(self, session: Session) -> None:
"""Save a session to disk."""
path = self._get_session_path(session.key)
with open(path, "w", encoding="utf-8") as f:
with open(path, "w") as f:
# Write metadata first
metadata_line = {
"_type": "metadata",
"key": session.key,
"created_at": session.created_at.isoformat(),
"updated_at": session.updated_at.isoformat(),
"metadata": session.metadata,
"last_consolidated": session.last_consolidated
"metadata": session.metadata
}
f.write(json.dumps(metadata_line, ensure_ascii=False) + "\n")
f.write(json.dumps(metadata_line) + "\n")
# Write messages
for msg in session.messages:
f.write(json.dumps(msg, ensure_ascii=False) + "\n")
f.write(json.dumps(msg) + "\n")
self._cache[session.key] = session
def invalidate(self, key: str) -> None:
"""Remove a session from the in-memory cache."""
def delete(self, key: str) -> bool:
"""
Delete a session.
Args:
key: Session key.
Returns:
True if deleted, False if not found.
"""
# Remove from cache
self._cache.pop(key, None)
# Remove file
path = self._get_session_path(key)
if path.exists():
path.unlink()
return True
return False
def list_sessions(self) -> list[dict[str, Any]]:
"""
@@ -194,14 +197,13 @@ class SessionManager:
for path in self.sessions_dir.glob("*.jsonl"):
try:
# Read just the metadata line
with open(path, encoding="utf-8") as f:
with open(path) as f:
first_line = f.readline().strip()
if first_line:
data = json.loads(first_line)
if data.get("_type") == "metadata":
key = data.get("key") or path.stem.replace("_", ":", 1)
sessions.append({
"key": key,
"key": path.stem.replace("_", ":"),
"created_at": data.get("created_at"),
"updated_at": data.get("updated_at"),
"path": str(path)