| 1 | """Debug hook that prints each iteration's LLM output and context stats.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | from typing import Any |
| 7 | |
| 8 | from nanobot.agent.hook import AgentHook, AgentHookContext |
| 9 | from nanobot.utils.helpers import estimate_message_tokens |
| 10 | |
| 11 | |
| 12 | class DebugHook(AgentHook): |
| 13 | """Prints detailed per-iteration info for debugging context management. |
| 14 | |
| 15 | Enable by passing this hook to AgentLoop or using --debug flag. |
| 16 | """ |
| 17 | |
| 18 | def __init__(self, *, show_full_content: bool = True, show_messages: bool = False) -> None: |
| 19 | super().__init__() |
| 20 | self._show_full_content = show_full_content |
| 21 | self._show_messages = show_messages |
| 22 | |
| 23 | async def after_iteration(self, context: AgentHookContext) -> None: |
| 24 | sep = "=" * 80 |
| 25 | print(f"\n{sep}") |
| 26 | print(f"[DEBUG] Iteration {context.iteration}") |
| 27 | print(f"{sep}") |
| 28 | |
| 29 | # Token usage |
| 30 | if context.usage: |
| 31 | print(f" Tokens — prompt: {context.usage.get('prompt_tokens', '?')}, " |
| 32 | f"completion: {context.usage.get('completion_tokens', '?')}") |
| 33 | |
| 34 | # Response content |
| 35 | if context.response: |
| 36 | content = context.response.content or "" |
| 37 | print(f" Response length: {len(content)} chars") |
| 38 | if self._show_full_content and content: |
| 39 | print(f" --- Response content ---") |
| 40 | print(f" {content[:2000]}") |
| 41 | if len(content) > 2000: |
| 42 | print(f" ... [truncated, total {len(content)} chars]") |
| 43 | print(f" --- End response ---") |
| 44 | |
| 45 | # Reasoning/thinking |
| 46 | if context.response.reasoning_content: |
| 47 | rc = context.response.reasoning_content |
| 48 | print(f" Reasoning: {len(rc)} chars") |
| 49 | |
| 50 | # Tool calls |
| 51 | if context.tool_calls: |
| 52 | print(f" Tool calls ({len(context.tool_calls)}):") |
| 53 | for tc in context.tool_calls: |
| 54 | args_preview = json.dumps(tc.arguments, ensure_ascii=False) |
| 55 | if len(args_preview) > 200: |
| 56 | args_preview = args_preview[:200] + "..." |
| 57 | print(f" - {tc.name}({args_preview})") |
| 58 | |
| 59 | # Tool results |
| 60 | if context.tool_results: |
| 61 | print(f" Tool results ({len(context.tool_results)}):") |
| 62 | for i, result in enumerate(context.tool_results): |
| 63 | result_str = str(result) |
| 64 | print(f" [{i}] {result_str[:300]}") |
| 65 | if len(result_str) > 300: |
| 66 | print(f" ... [truncated, total {len(result_str)} chars]") |
| 67 | |
| 68 | # Messages context size |
| 69 | if context.messages: |
| 70 | total_msgs = len(context.messages) |
| 71 | estimated_tokens = estimate_prompt_tokens_simple(context.messages) |
| 72 | print(f" Messages in context: {total_msgs}, ~{estimated_tokens} tokens (estimate)") |
| 73 | |
| 74 | # Show full messages if requested |
| 75 | if self._show_messages and context.messages: |
| 76 | print(f" --- Full messages ---") |
| 77 | for i, msg in enumerate(context.messages[-5:]): |
| 78 | role = msg.get("role", "?") |
| 79 | content = msg.get("content", "") |
| 80 | if isinstance(content, str): |
| 81 | preview = content[:200] |
| 82 | else: |
| 83 | preview = str(content)[:200] |
| 84 | print(f" [{total_msgs - 5 + i}] {role}: {preview}") |
| 85 | print(f" --- End messages (showing last 5 of {total_msgs}) ---") |
| 86 | |
| 87 | # Stop reason / error |
| 88 | if context.stop_reason: |
| 89 | print(f" Stop reason: {context.stop_reason}") |
| 90 | if context.error: |
| 91 | print(f" ERROR: {context.error}") |
| 92 | |
| 93 | print(f"{sep}\n") |
| 94 | |
| 95 | |
| 96 | def estimate_prompt_tokens_simple(messages: list[dict[str, Any]]) -> int: |
| 97 | """Quick estimate of total tokens in message list.""" |
| 98 | total = 0 |
| 99 | for msg in messages: |
| 100 | content = msg.get("content", "") |
| 101 | if isinstance(content, str): |
| 102 | total += len(content) // 4 |
| 103 | elif isinstance(content, list): |
| 104 | for block in content: |
| 105 | if isinstance(block, dict): |
| 106 | text = block.get("text", "") or block.get("content", "") |
| 107 | total += len(str(text)) // 4 |
| 108 | # tool calls |
| 109 | tool_calls = msg.get("tool_calls", []) |
| 110 | for tc in tool_calls: |
| 111 | if isinstance(tc, dict): |
| 112 | args = tc.get("function", {}).get("arguments", "") |
| 113 | total += len(str(args)) // 4 |
| 114 | return total |
| 115 |