| | """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") |
| |
|
