Hua/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style(loop): remove formatting-only changes from upstream PR 881

Browse Source
This commit is contained in:
Alexander Minges
2026-02-20 12:46:11 +01:00
parent 755e424127
commit 5f9eca4664
1 changed files with 85 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
nanobot/agent/loop.py
View File

@@ -56,7 +56,6 @@ class AgentLoop:
): ):
from nanobot.config.schema import ExecToolConfig from nanobot.config.schema import ExecToolConfig
from nanobot.cron.service import CronService from nanobot.cron.service import CronService
self.bus = bus self.bus = bus
self.provider = provider self.provider = provider
self.workspace = workspace self.workspace = workspace
@@ -90,7 +89,7 @@ class AgentLoop:
self._mcp_stack: AsyncExitStack | None = None self._mcp_stack: AsyncExitStack | None = None
self._mcp_connected = False self._mcp_connected = False
self._consolidating: set[str] = set() # Session keys with consolidation in progress self._consolidating: set[str] = set() # Session keys with consolidation in progress
self._consolidation_tasks: set[asyncio.Task] = set() # Keep strong refs for in-flight tasks self._consolidation_tasks: set[asyncio.Task] = set() # Strong refs to in-flight tasks
self._consolidation_locks: dict[str, asyncio.Lock] = {} self._consolidation_locks: dict[str, asyncio.Lock] = {}
self._register_default_tools() self._register_default_tools()
@@ -104,13 +103,11 @@ class AgentLoop:
self.tools.register(ListDirTool(workspace=self.workspace, allowed_dir=allowed_dir)) self.tools.register(ListDirTool(workspace=self.workspace, allowed_dir=allowed_dir))
# Shell tool # Shell tool
self.tools.register( self.tools.register(ExecTool(
ExecTool( working_dir=str(self.workspace),
working_dir=str(self.workspace), timeout=self.exec_config.timeout,
timeout=self.exec_config.timeout, restrict_to_workspace=self.restrict_to_workspace,
restrict_to_workspace=self.restrict_to_workspace, ))
)
)
# Web tools # Web tools
self.tools.register(WebSearchTool(api_key=self.brave_api_key)) self.tools.register(WebSearchTool(api_key=self.brave_api_key))
@@ -134,7 +131,6 @@ class AgentLoop:
return return
self._mcp_connected = True self._mcp_connected = True
from nanobot.agent.tools.mcp import connect_mcp_servers from nanobot.agent.tools.mcp import connect_mcp_servers
self._mcp_stack = AsyncExitStack() self._mcp_stack = AsyncExitStack()
await self._mcp_stack.__aenter__() await self._mcp_stack.__aenter__()
await connect_mcp_servers(self._mcp_servers, self.tools, self._mcp_stack) await connect_mcp_servers(self._mcp_servers, self.tools, self._mcp_stack)
@@ -163,13 +159,11 @@ class AgentLoop:
@staticmethod @staticmethod
def _tool_hint(tool_calls: list) -> str: def _tool_hint(tool_calls: list) -> str:
"""Format tool calls as concise hint, e.g. 'web_search("query")'.""" """Format tool calls as concise hint, e.g. 'web_search("query")'."""
def _fmt(tc): def _fmt(tc):
val = next(iter(tc.arguments.values()), None) if tc.arguments else None val = next(iter(tc.arguments.values()), None) if tc.arguments else None
if not isinstance(val, str): if not isinstance(val, str):
return tc.name return tc.name
return f'{tc.name}("{val[:40]}…")' if len(val) > 40 else f'{tc.name}("{val}")' return f'{tc.name}("{val[:40]}…")' if len(val) > 40 else f'{tc.name}("{val}")'
return ", ".join(_fmt(tc) for tc in tool_calls) return ", ".join(_fmt(tc) for tc in tool_calls)
async def _run_agent_loop( async def _run_agent_loop(
@@ -217,15 +211,13 @@ class AgentLoop:
"type": "function", "type": "function",
"function": { "function": {
"name": tc.name, "name": tc.name,
"arguments": json.dumps(tc.arguments, ensure_ascii=False), "arguments": json.dumps(tc.arguments, ensure_ascii=False)
}, }
} }
for tc in response.tool_calls for tc in response.tool_calls
] ]
messages = self.context.add_assistant_message( messages = self.context.add_assistant_message(
messages, messages, response.content, tool_call_dicts,
response.content,
tool_call_dicts,
reasoning_content=response.reasoning_content, reasoning_content=response.reasoning_content,
) )
@@ -243,13 +235,9 @@ class AgentLoop:
# Give them one retry; don't forward the text to avoid duplicates. # Give them one retry; don't forward the text to avoid duplicates.
if not tools_used and not text_only_retried and final_content: if not tools_used and not text_only_retried and final_content:
text_only_retried = True text_only_retried = True
logger.debug( logger.debug("Interim text response (no tools used yet), retrying: {}", final_content[:80])
"Interim text response (no tools used yet), retrying: {}",
final_content[:80],
)
messages = self.context.add_assistant_message( messages = self.context.add_assistant_message(
messages, messages, response.content,
response.content,
reasoning_content=response.reasoning_content, reasoning_content=response.reasoning_content,
) )
final_content = None final_content = None
@@ -266,20 +254,21 @@ class AgentLoop:
while self._running: while self._running:
try: try:
msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0) msg = await asyncio.wait_for(
self.bus.consume_inbound(),
timeout=1.0
)
try: try:
response = await self._process_message(msg) response = await self._process_message(msg)
if response: if response:
await self.bus.publish_outbound(response) await self.bus.publish_outbound(response)
except Exception as e: except Exception as e:
logger.error("Error processing message: {}", e) logger.error("Error processing message: {}", e)
await self.bus.publish_outbound( await self.bus.publish_outbound(OutboundMessage(
OutboundMessage( channel=msg.channel,
channel=msg.channel, chat_id=msg.chat_id,
chat_id=msg.chat_id, content=f"Sorry, I encountered an error: {str(e)}"
content=f"Sorry, I encountered an error: {str(e)}", ))
)
)
except asyncio.TimeoutError: except asyncio.TimeoutError:
continue continue
@@ -298,7 +287,6 @@ class AgentLoop:
logger.info("Agent loop stopping") logger.info("Agent loop stopping")
def _get_consolidation_lock(self, session_key: str) -> asyncio.Lock: def _get_consolidation_lock(self, session_key: str) -> asyncio.Lock:
"""Return a per-session lock for memory consolidation writers."""
lock = self._consolidation_locks.get(session_key) lock = self._consolidation_locks.get(session_key)
if lock is None: if lock is None:
lock = asyncio.Lock() lock = asyncio.Lock()
@@ -348,23 +336,17 @@ class AgentLoop:
return OutboundMessage( return OutboundMessage(
channel=msg.channel, channel=msg.channel,
chat_id=msg.chat_id, chat_id=msg.chat_id,
content="Could not start a new session because memory archival failed. Please try again.", content="Could not start a new session because memory archival failed. Please try again."
) )
session.clear() session.clear()
self.sessions.save(session) self.sessions.save(session)
self.sessions.invalidate(session.key) self.sessions.invalidate(session.key)
return OutboundMessage( return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
channel=msg.channel, content="New session started. Memory consolidation in progress.")
chat_id=msg.chat_id,
content="New session started. Memory consolidation in progress.",
)
if cmd == "/help": if cmd == "/help":
return OutboundMessage( return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
channel=msg.channel, content="🐈 nanobot commands:\n/new — Start a new conversation\n/help — Show available commands")
chat_id=msg.chat_id,
content="🐈 nanobot commands:\n/new — Start a new conversation\n/help — Show available commands",
)
if len(session.messages) > self.memory_window and session.key not in self._consolidating: if len(session.messages) > self.memory_window and session.key not in self._consolidating:
self._consolidating.add(session.key) self._consolidating.add(session.key)
@@ -376,12 +358,12 @@ class AgentLoop:
await self._consolidate_memory(session) await self._consolidate_memory(session)
finally: finally:
self._consolidating.discard(session.key) self._consolidating.discard(session.key)
task = asyncio.current_task() _task = asyncio.current_task()
if task is not None: if _task is not None:
self._consolidation_tasks.discard(task) self._consolidation_tasks.discard(_task)
task = asyncio.create_task(_consolidate_and_unlock()) _task = asyncio.create_task(_consolidate_and_unlock())
self._consolidation_tasks.add(task) self._consolidation_tasks.add(_task)
self._set_tool_context(msg.channel, msg.chat_id, msg.metadata.get("message_id")) self._set_tool_context(msg.channel, msg.chat_id, msg.metadata.get("message_id"))
initial_messages = self.context.build_messages( initial_messages = self.context.build_messages(
@@ -393,18 +375,13 @@ class AgentLoop:
) )
async def _bus_progress(content: str) -> None: async def _bus_progress(content: str) -> None:
await self.bus.publish_outbound( await self.bus.publish_outbound(OutboundMessage(
OutboundMessage( channel=msg.channel, chat_id=msg.chat_id, content=content,
channel=msg.channel, metadata=msg.metadata or {},
chat_id=msg.chat_id, ))
content=content,
metadata=msg.metadata or {},
)
)
final_content, tools_used = await self._run_agent_loop( final_content, tools_used = await self._run_agent_loop(
initial_messages, initial_messages, on_progress=on_progress or _bus_progress,
on_progress=on_progress or _bus_progress,
) )
if final_content is None: if final_content is None:
@@ -414,17 +391,15 @@ class AgentLoop:
logger.info("Response to {}:{}: {}", msg.channel, msg.sender_id, preview) logger.info("Response to {}:{}: {}", msg.channel, msg.sender_id, preview)
session.add_message("user", msg.content) session.add_message("user", msg.content)
session.add_message( session.add_message("assistant", final_content,
"assistant", final_content, tools_used=tools_used if tools_used else None tools_used=tools_used if tools_used else None)
)
self.sessions.save(session) self.sessions.save(session)
return OutboundMessage( return OutboundMessage(
channel=msg.channel, channel=msg.channel,
chat_id=msg.chat_id, chat_id=msg.chat_id,
content=final_content, content=final_content,
metadata=msg.metadata metadata=msg.metadata or {}, # Pass through for channel-specific needs (e.g. Slack thread_ts)
or {}, # Pass through for channel-specific needs (e.g. Slack thread_ts)
) )
async def _process_system_message(self, msg: InboundMessage) -> OutboundMessage | None: async def _process_system_message(self, msg: InboundMessage) -> OutboundMessage | None:
@@ -465,7 +440,9 @@ class AgentLoop:
self.sessions.save(session) self.sessions.save(session)
return OutboundMessage( return OutboundMessage(
channel=origin_channel, chat_id=origin_chat_id, content=final_content channel=origin_channel,
chat_id=origin_chat_id,
content=final_content
) )
async def _consolidate_memory(self, session, archive_all: bool = False) -> None: async def _consolidate_memory(self, session, archive_all: bool = False) -> None:
@@ -480,49 +457,29 @@ class AgentLoop:
if archive_all: if archive_all:
old_messages = session.messages old_messages = session.messages
keep_count = 0 keep_count = 0
logger.info( logger.info("Memory consolidation (archive_all): {} total messages archived", len(session.messages))
"Memory consolidation (archive_all): {} total messages archived",
len(session.messages),
)
else: else:
keep_count = self.memory_window // 2 keep_count = self.memory_window // 2
if len(session.messages) <= keep_count: if len(session.messages) <= keep_count:
logger.debug( logger.debug("Session {}: No consolidation needed (messages={}, keep={})", session.key, len(session.messages), keep_count)
"Session {}: No consolidation needed (messages={}, keep={})",
session.key,
len(session.messages),
keep_count,
)
return return
messages_to_process = len(session.messages) - session.last_consolidated messages_to_process = len(session.messages) - session.last_consolidated
if messages_to_process <= 0: if messages_to_process <= 0:
logger.debug( logger.debug("Session {}: No new messages to consolidate (last_consolidated={}, total={})", session.key, session.last_consolidated, len(session.messages))
"Session {}: No new messages to consolidate (last_consolidated={}, total={})",
session.key,
session.last_consolidated,
len(session.messages),
)
return return
old_messages = session.messages[session.last_consolidated : -keep_count] old_messages = session.messages[session.last_consolidated:-keep_count]
if not old_messages: if not old_messages:
return return
logger.info( logger.info("Memory consolidation started: {} total, {} new to consolidate, {} keep", len(session.messages), len(old_messages), keep_count)
"Memory consolidation started: {} total, {} new to consolidate, {} keep",
len(session.messages),
len(old_messages),
keep_count,
)
lines = [] lines = []
for m in old_messages: for m in old_messages:
if not m.get("content"): if not m.get("content"):
continue continue
tools = f" [tools: {', '.join(m['tools_used'])}]" if m.get("tools_used") else "" tools = f" [tools: {', '.join(m['tools_used'])}]" if m.get("tools_used") else ""
lines.append( lines.append(f"[{m.get('timestamp', '?')[:16]}] {m['role'].upper()}{tools}: {m['content']}")
f"[{m.get('timestamp', '?')[:16]}] {m['role'].upper()}{tools}: {m['content']}"
)
conversation = "\n".join(lines) conversation = "\n".join(lines)
current_memory = memory.read_long_term() current_memory = memory.read_long_term()
@@ -551,10 +508,7 @@ Respond with ONLY valid JSON, no markdown fences."""
try: try:
response = await self.provider.chat( response = await self.provider.chat(
messages=[ messages=[
{ {"role": "system", "content": "You are a memory consolidation agent. Respond only with valid JSON."},
"role": "system",
"content": "You are a memory consolidation agent. Respond only with valid JSON.",
},
{"role": "user", "content": prompt}, {"role": "user", "content": prompt},
], ],
model=self.model, model=self.model,
@@ -567,10 +521,7 @@ Respond with ONLY valid JSON, no markdown fences."""
text = text.split("\n", 1)[-1].rsplit("```", 1)[0].strip() text = text.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
result = json_repair.loads(text) result = json_repair.loads(text)
if not isinstance(result, dict): if not isinstance(result, dict):
logger.warning( logger.warning("Memory consolidation: unexpected response type, skipping. Response: {}", text[:200])
"Memory consolidation: unexpected response type, skipping. Response: {}",
text[:200],
)
return return
if entry := result.get("history_entry"): if entry := result.get("history_entry"):
@@ -589,11 +540,7 @@ Respond with ONLY valid JSON, no markdown fences."""
session.last_consolidated = 0 session.last_consolidated = 0
else: else:
session.last_consolidated = len(session.messages) - keep_count session.last_consolidated = len(session.messages) - keep_count
logger.info( logger.info("Memory consolidation done: {} messages, last_consolidated={}", len(session.messages), session.last_consolidated)
"Memory consolidation done: {} messages, last_consolidated={}",
len(session.messages),
session.last_consolidated,
)
except Exception as e: except Exception as e:
logger.error("Memory consolidation failed: {}", e) logger.error("Memory consolidation failed: {}", e)
@@ -619,9 +566,12 @@ Respond with ONLY valid JSON, no markdown fences."""
The agent's response. The agent's response.
""" """
await self._connect_mcp() await self._connect_mcp()
msg = InboundMessage(channel=channel, sender_id="user", chat_id=chat_id, content=content) msg = InboundMessage(
channel=channel,
response = await self._process_message( sender_id="user",
msg, session_key=session_key, on_progress=on_progress chat_id=chat_id,
content=content
) )
response = await self._process_message(msg, session_key=session_key, on_progress=on_progress)
return response.content if response else "" return response.content if response else ""