|
|
"""Logger utility for LocalMate - Structured logging for debugging. |
|
|
|
|
|
Provides colored console logging with structured output for: |
|
|
- API request/response |
|
|
- Tool execution |
|
|
- LLM calls |
|
|
- Workflow tracing |
|
|
""" |
|
|
|
|
|
import logging |
|
|
import json |
|
|
import sys |
|
|
from datetime import datetime |
|
|
from typing import Any |
|
|
from dataclasses import dataclass, field, asdict |
|
|
|
|
|
|
|
|
|
|
|
logging.basicConfig( |
|
|
level=logging.INFO, |
|
|
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", |
|
|
datefmt="%H:%M:%S", |
|
|
stream=sys.stdout, |
|
|
) |
|
|
|
|
|
|
|
|
COLORS = { |
|
|
"RESET": "\033[0m", |
|
|
"BOLD": "\033[1m", |
|
|
"CYAN": "\033[36m", |
|
|
"GREEN": "\033[32m", |
|
|
"YELLOW": "\033[33m", |
|
|
"MAGENTA": "\033[35m", |
|
|
"BLUE": "\033[34m", |
|
|
"RED": "\033[31m", |
|
|
} |
|
|
|
|
|
|
|
|
def colorize(text: str, color: str) -> str: |
|
|
"""Add color to text for terminal output.""" |
|
|
return f"{COLORS.get(color, '')}{text}{COLORS['RESET']}" |
|
|
|
|
|
|
|
|
class LocalMateLogger: |
|
|
"""Structured logger for LocalMate with colored output.""" |
|
|
|
|
|
def __init__(self, name: str): |
|
|
self.logger = logging.getLogger(name) |
|
|
self.name = name |
|
|
|
|
|
def _format_data(self, data: Any, max_len: int = 500) -> str: |
|
|
"""Format data for logging, truncating if needed.""" |
|
|
if data is None: |
|
|
return "None" |
|
|
|
|
|
if isinstance(data, (dict, list)): |
|
|
try: |
|
|
formatted = json.dumps(data, ensure_ascii=False, default=str) |
|
|
if len(formatted) > max_len: |
|
|
return formatted[:max_len] + "..." |
|
|
return formatted |
|
|
except: |
|
|
return str(data)[:max_len] |
|
|
|
|
|
text = str(data) |
|
|
return text[:max_len] + "..." if len(text) > max_len else text |
|
|
|
|
|
def api_request(self, endpoint: str, method: str, params: dict = None, body: Any = None): |
|
|
"""Log API request.""" |
|
|
msg = f"{colorize('→ REQUEST', 'CYAN')} {colorize(method, 'BOLD')} {endpoint}" |
|
|
if params: |
|
|
msg += f"\n Params: {self._format_data(params)}" |
|
|
if body: |
|
|
msg += f"\n Body: {self._format_data(body)}" |
|
|
self.logger.info(msg) |
|
|
|
|
|
def api_response(self, endpoint: str, status: int, data: Any = None, duration_ms: float = None): |
|
|
"""Log API response.""" |
|
|
status_color = "GREEN" if status < 400 else "RED" |
|
|
msg = f"{colorize('← RESPONSE', status_color)} {endpoint} [{status}]" |
|
|
if duration_ms: |
|
|
msg += f" ({duration_ms:.0f}ms)" |
|
|
if data: |
|
|
msg += f"\n Data: {self._format_data(data)}" |
|
|
self.logger.info(msg) |
|
|
|
|
|
def tool_call(self, tool_name: str, arguments: dict): |
|
|
"""Log tool call start.""" |
|
|
msg = f"{colorize('🔧 TOOL', 'MAGENTA')} {colorize(tool_name, 'BOLD')}" |
|
|
msg += f"\n Args: {self._format_data(arguments)}" |
|
|
self.logger.info(msg) |
|
|
|
|
|
def tool_result(self, tool_name: str, result_count: int, sample: Any = None): |
|
|
"""Log tool result.""" |
|
|
msg = f"{colorize('✓ RESULT', 'GREEN')} {tool_name} → {result_count} results" |
|
|
if sample: |
|
|
msg += f"\n Sample: {self._format_data(sample, max_len=200)}" |
|
|
self.logger.info(msg) |
|
|
|
|
|
def llm_call(self, provider: str, model: str, prompt_preview: str = None): |
|
|
"""Log LLM call.""" |
|
|
msg = f"{colorize('🤖 LLM', 'BLUE')} {provider}/{model}" |
|
|
if prompt_preview: |
|
|
preview = prompt_preview[:100] + "..." if len(prompt_preview) > 100 else prompt_preview |
|
|
msg += f"\n Prompt: {preview}" |
|
|
self.logger.info(msg) |
|
|
|
|
|
def llm_response(self, provider: str, response_preview: str = None, tokens: int = None): |
|
|
"""Log LLM response.""" |
|
|
msg = f"{colorize('💬 LLM RESPONSE', 'BLUE')} {provider}" |
|
|
if tokens: |
|
|
msg += f" ({tokens} tokens)" |
|
|
if response_preview: |
|
|
preview = response_preview[:150] + "..." if len(response_preview) > 150 else response_preview |
|
|
msg += f"\n Response: {preview}" |
|
|
self.logger.info(msg) |
|
|
|
|
|
def workflow_step(self, step: str, details: str = None): |
|
|
"""Log workflow step.""" |
|
|
msg = f"{colorize('▶', 'YELLOW')} {step}" |
|
|
if details: |
|
|
msg += f": {details}" |
|
|
self.logger.info(msg) |
|
|
|
|
|
def error(self, message: str, error: Exception = None): |
|
|
"""Log error.""" |
|
|
msg = f"{colorize('❌ ERROR', 'RED')} {message}" |
|
|
if error: |
|
|
msg += f"\n {type(error).__name__}: {str(error)}" |
|
|
self.logger.error(msg) |
|
|
|
|
|
def debug(self, message: str, data: Any = None): |
|
|
"""Log debug info.""" |
|
|
msg = f"{colorize('DEBUG', 'CYAN')} {message}" |
|
|
if data: |
|
|
msg += f": {self._format_data(data)}" |
|
|
self.logger.debug(msg) |
|
|
|
|
|
|
|
|
@dataclass |
|
|
class WorkflowStep: |
|
|
"""A step in the agent workflow.""" |
|
|
|
|
|
step_name: str |
|
|
tool_name: str | None = None |
|
|
purpose: str = "" |
|
|
input_summary: str = "" |
|
|
output_summary: str = "" |
|
|
result_count: int = 0 |
|
|
duration_ms: float = 0 |
|
|
|
|
|
|
|
|
@dataclass |
|
|
class AgentWorkflow: |
|
|
"""Complete workflow trace for a chat request.""" |
|
|
|
|
|
query: str |
|
|
intent_detected: str = "" |
|
|
steps: list[WorkflowStep] = field(default_factory=list) |
|
|
total_duration_ms: float = 0 |
|
|
tools_used: list[str] = field(default_factory=list) |
|
|
|
|
|
def add_step(self, step: WorkflowStep): |
|
|
"""Add a step to the workflow.""" |
|
|
self.steps.append(step) |
|
|
if step.tool_name and step.tool_name not in self.tools_used: |
|
|
self.tools_used.append(step.tool_name) |
|
|
|
|
|
def to_dict(self) -> dict: |
|
|
"""Convert to dictionary for JSON serialization.""" |
|
|
return { |
|
|
"query": self.query, |
|
|
"intent_detected": self.intent_detected, |
|
|
"tools_used": self.tools_used, |
|
|
"steps": [ |
|
|
{ |
|
|
"step": s.step_name, |
|
|
"tool": s.tool_name, |
|
|
"purpose": s.purpose, |
|
|
"results": s.result_count, |
|
|
} |
|
|
for s in self.steps |
|
|
], |
|
|
"total_duration_ms": round(self.total_duration_ms, 1), |
|
|
} |
|
|
|
|
|
def to_summary(self) -> str: |
|
|
"""Generate human-readable workflow summary.""" |
|
|
lines = [f"📊 **Workflow Summary**"] |
|
|
lines.append(f"- Query: \"{self.query[:50]}{'...' if len(self.query) > 50 else ''}\"") |
|
|
lines.append(f"- Intent: {self.intent_detected}") |
|
|
lines.append(f"- Tools: {', '.join(self.tools_used) or 'None'}") |
|
|
|
|
|
if self.steps: |
|
|
lines.append("\n**Steps:**") |
|
|
for i, step in enumerate(self.steps, 1): |
|
|
tool_info = f" ({step.tool_name})" if step.tool_name else "" |
|
|
results_info = f" → {step.result_count} results" if step.result_count else "" |
|
|
lines.append(f"{i}. {step.step_name}{tool_info}{results_info}") |
|
|
|
|
|
lines.append(f"\n⏱️ Total: {self.total_duration_ms:.0f}ms") |
|
|
return "\n".join(lines) |
|
|
|
|
|
|
|
|
|
|
|
agent_logger = LocalMateLogger("agent") |
|
|
api_logger = LocalMateLogger("api") |
|
|
tool_logger = LocalMateLogger("tools") |
|
|
|
