""" Formatting utilities for iterative reasoning. This module handles formatting data for Oracle context and Matrix display. """ from typing import Dict, Any from core.logger import setup_logger from .models import IterativeContext, StepResult, StepAction logger = setup_logger('formatters', service_name='think_service') class KnowledgeFormatter: """Formats knowledge and results for Oracle and Matrix display""" def format_for_oracle(self, context: IterativeContext) -> str: """Format accumulated knowledge as natural language for Oracle context""" if not context.accumulated_knowledge and not context.completed_steps: return "No information gathered yet." parts = [] # Describe what actions were taken if context.completed_steps: actions_taken = [] for step in context.completed_steps: if step.step.action == StepAction.CALL_SERVICE.value and step.step.target: actions_taken.append(f"consulted {step.step.target}") if actions_taken: parts.append(f"You have {', '.join(actions_taken)}.") # Present the knowledge gathered in prose if context.accumulated_knowledge: parts.append("\nInformation gathered:") for key, data in context.accumulated_knowledge.items(): formatted = self._format_data_item(data) if formatted: parts.append(formatted) return "\n".join(parts) if parts else "Some information was gathered." def _format_data_item(self, data: Any) -> str: """Format a single data item""" if isinstance(data, dict): return self._format_dict_data(data) elif isinstance(data, str): return f" • {data}" else: return f" • {str(data)}" def _format_dict_data(self, data: Dict[str, Any]) -> str: """Format dictionary data based on its type""" # Short-term memory response if 'memories' in data and data.get('type') == 'short_term': return self._format_short_memory(data) # Long-term memory response elif 'memories' in data and data.get('type') == 'long_term': return self._format_long_memory(data) # Facts response elif 'facts' in data: return self._format_facts(data) # save_fact response elif 'status' in data and 'fact_id' in data: return self._format_save_fact(data) # update_fact response elif 'status' in data and 'message' in data and 'fact_id' not in data: return self._format_update_fact(data) # Legacy memory format elif 'memories' in data and 'type' not in data: return self._format_legacy_memory(data) # DuckDuckGo search response elif 'query' in data and ('answer' in data or 'results' in data or 'related_topics' in data): return self._format_duckduckgo(data) # Todo list response elif 'todos' in data and 'summary' in data: return self._format_todos(data) # Todo create/update/complete response elif 'todo_id' in data and 'message' in data: return self._format_todo_action(data) # Generic dict data else: return self._format_generic_dict(data) def _format_short_memory(self, data: Dict[str, Any]) -> str: """Format short-term memory response""" success = data.get('success', True) if not success: error = data.get('error', 'Unknown error') return f" ⚠️ Failed to retrieve recent memories: {error}" memories = data['memories'] count = data.get('count', len(memories)) offset = data.get('offset', 0) if count > 0: label = f"Recent literal memories ({count} messages" if offset > 0: label += f", starting from {offset} back" label += "):" parts = [f"\n {label}"] for i, msg in enumerate(memories, 1): if isinstance(msg, dict): content = msg.get('content', '') timestamp = msg.get('timestamp', '') parts.append(f" {i}. [{timestamp}] {content}") else: parts.append(f" {i}. {msg}") return "\n".join(parts) else: return f" • No recent memories found (successfully queried, but empty)" def _format_long_memory(self, data: Dict[str, Any]) -> str: """Format long-term memory response""" success = data.get('success', True) if not success: error = data.get('error', 'Unknown error') return f" ⚠️ Failed to retrieve long-term memories: {error}" memories = data['memories'] count = data.get('count', len(memories)) query = data.get('query') if count > 0: label = f"Historical context ({count} summaries" if query: label += f" about '{query}'" label += "):" parts = [f"\n {label}"] for i, mem in enumerate(memories, 1): if isinstance(mem, dict): content = mem.get('content', '') parts.append(f" {i}. {content}") else: parts.append(f" {i}. {mem}") return "\n".join(parts) else: query_text = f" about '{query}'" if query else "" return f" • No historical context found{query_text} (successfully queried, but empty)" def _format_facts(self, data: Dict[str, Any]) -> str: """Format facts response""" success = data.get('success', True) if not success: error = data.get('error', 'Unknown error') return f" ⚠️ Failed to retrieve facts: {error}" facts = data['facts'] count = data.get('count', len(facts)) query = data.get('query', '') if count > 0: parts = [f"\n Known facts ({count}):"] for i, fact in enumerate(facts, 1): if isinstance(fact, dict): content = fact.get('content', '') category = fact.get('category', 'general') mutable = fact.get('mutable', True) fact_id = fact.get('id', '') mut_marker = " [mutable]" if mutable else " [immutable]" parts.append(f" {i}. [{category}]{mut_marker} {content}") if fact_id: parts.append(f" (id: {fact_id})") else: parts.append(f" {i}. {fact}") return "\n".join(parts) else: query_text = f" about '{query}'" if query else "" return f" • No facts found{query_text} (successfully queried, but empty)" def _format_save_fact(self, data: Dict[str, Any]) -> str: """Format save_fact response""" success = data.get('success', data.get('status') == 'success') if success: fact_id = data.get('fact_id', '') message = data.get('message', 'Fact saved successfully') parts = [f" ✓ {message}"] if fact_id: parts.append(f" (Fact ID: {fact_id})") return "\n".join(parts) else: error = data.get('error', 'Failed to save fact') return f" ✗ {error}" def _format_update_fact(self, data: Dict[str, Any]) -> str: """Format update_fact response""" success = data.get('success', data.get('status') == 'success') if success: message = data.get('message', 'Fact updated successfully') return f" ✓ {message}" else: error = data.get('error', 'Failed to update fact') return f" ✗ {error}" def _format_legacy_memory(self, data: Dict[str, Any]) -> str: """Format legacy memory format""" memories = data['memories'] count = data.get('count', len(memories)) if count > 0: parts = [f"\n Conversation history ({count} messages):"] for i, msg in enumerate(memories, 1): if isinstance(msg, dict): parts.append(f" Message {i}:") for field, value in msg.items(): value_str = str(value) parts.append(f" • {field}: {value_str}") else: parts.append(f" {i}. {msg}") return "\n".join(parts) else: return f" • No conversation history" def _format_duckduckgo(self, data: Dict[str, Any]) -> str: """Format DuckDuckGo search response""" success = data.get('success', True) if not success: error = data.get('error', 'Unknown error') return f" ⚠️ DuckDuckGo search failed: {error}" query = data.get('query', '') answer = data.get('answer') abstract = data.get('abstract') definition = data.get('definition') results = data.get('results', []) topics = data.get('related_topics', []) parts = [f"\n DuckDuckGo search for '{query}':"] if answer: parts.append(f" Instant answer: {answer}") if abstract: parts.append(f" Abstract: {abstract}") if data.get('abstract_source'): parts.append(f" Source: {data.get('abstract_source')}") if definition: parts.append(f" Definition: {definition}") if data.get('definition_source'): parts.append(f" Source: {data.get('definition_source')}") if topics: parts.append(f" Related topics ({len(topics)}):") for i, topic in enumerate(topics[:5], 1): parts.append(f" {i}. {topic.get('text', 'N/A')}") if results: parts.append(f" Results ({len(results)}):") for i, result in enumerate(results[:3], 1): parts.append(f" {i}. {result.get('text', 'N/A')}") return "\n".join(parts) def _format_generic_dict(self, data: Dict[str, Any]) -> str: """Format generic dictionary data""" parts = [] for field, value in data.items(): value_str = str(value) parts.append(f" • {field}: {value_str}") return "\n " + "\n".join(parts) if parts else "" def format_step_result_for_matrix(self, step_result: StepResult) -> str: """Format step execution result for Matrix display""" try: if not step_result.result_data: return "No data returned" result_data = step_result.result_data # Format based on step type if step_result.step.action == StepAction.CALL_SERVICE.value: service = step_result.step.target return self._format_service_result(service, result_data) elif step_result.step.action == StepAction.CHECK_GOAL_SATISFACTION.value: can_answer = result_data return f"Goal check: {'Can answer' if can_answer else 'Need more info'}" else: return self._format_generic_result(result_data) except Exception as e: logger.error(f"[📊] Error formatting step result: {e}") return "Result formatting error" def _format_service_result(self, service: str, result_data: Dict[str, Any]) -> str: """Format service call result""" if service == "short_memory": count = result_data.get("count", 0) offset = result_data.get("offset", 0) if offset > 0: return f"Retrieved {count} recent memories (starting from {offset} back)" return f"Retrieved {count} recent memories" elif service == "long_memory": count = result_data.get("count", 0) query = result_data.get("query") if query: return f"Retrieved {count} historical summaries for '{query}'" return f"Retrieved {count} historical summaries" elif service == "facts": count = result_data.get("count", 0) query = result_data.get("query") if query: return f"Found {count} facts about '{query}'" return f"Found {count} facts" elif service == "save_fact": if result_data.get("status") == "success": return f"✓ Fact saved" return f"✗ Failed to save fact" elif service == "update_fact": if result_data.get("status") == "success": return f"✓ Fact updated" return f"✗ Failed to update fact" elif service == "memory": count = result_data.get("count", 0) return f"Retrieved {count} memories [legacy]" elif service == "identity": identity = result_data.get("identity", {}) name = identity.get("name", "unknown") return f"Identity resolved: {name}" elif service == "health": return self._format_health_result(result_data) elif service == "duckduckgo": return self._format_duckduckgo_result(result_data) elif service == "plugin": checked = result_data.get("plugin_actions_checked", False) return f"Plugin actions {'checked' if checked else 'failed'}" else: return f"Service call to {service} completed" def _format_health_result(self, result_data: Dict[str, Any]) -> str: """Format health check result""" status = result_data.get("status", "unknown") cluster_health = result_data.get("cluster_health", {}) healthy_nodes = cluster_health.get("healthy_nodes", 0) total_nodes = cluster_health.get("total_nodes", 0) issues_count = cluster_health.get("issues_count", 0) if status == "error": return f"Health check failed: {result_data.get('error', 'Unknown error')}" elif issues_count > 0: return f"System status: {status} ({healthy_nodes}/{total_nodes} nodes healthy, {issues_count} issues)" else: return f"System status: {status} ({healthy_nodes}/{total_nodes} nodes healthy)" def _format_duckduckgo_result(self, result_data: Dict[str, Any]) -> str: """Format DuckDuckGo search result""" success = result_data.get("success", False) if success: query = result_data.get("query", "") answer = result_data.get("answer") results_count = len(result_data.get("results", [])) topics_count = len(result_data.get("related_topics", [])) if answer: return f"🦆 Found instant answer for '{query}'" elif results_count > 0 or topics_count > 0: return f"🦆 Found {results_count} results, {topics_count} topics for '{query}'" else: return f"🦆 No results for '{query}'" else: return f"🦆 Search failed: {result_data.get('error', 'unknown error')}" def _format_generic_result(self, result_data: Any) -> str: """Format generic result data""" if isinstance(result_data, dict): key_count = len(result_data) return f"Returned {key_count} data fields" elif isinstance(result_data, list): item_count = len(result_data) return f"Returned {item_count} items" else: result_str = str(result_data) return result_str[:100] + "..." if len(result_str) > 100 else result_str def _format_todos(self, data: Dict[str, Any]) -> str: """Format todo list response""" success = data.get('success', True) if not success: error = data.get('error', 'Unknown error') return f" ⚠️ Failed to retrieve todos: {error}" todos = data.get('todos', []) summary = data.get('summary', {}) count = data.get('count', len(todos)) if count == 0: return " 📋 No todos yet" # Build status summary total = summary.get('total', 0) pending = summary.get('pending', 0) in_progress = summary.get('in_progress', 0) completed = summary.get('completed', 0) status_line = f" 📋 Current Tasks ({total} total: {pending} pending, {in_progress} in progress, {completed} completed):" parts = [status_line] # Group todos by status status_icons = { 'pending': '⏳', 'in_progress': '🔄', 'completed': '✅' } for status in ['in_progress', 'pending', 'completed']: status_todos = [t for t in todos if t['status'] == status] if status_todos: parts.append(f"\n {status.replace('_', ' ').title()}:") for todo in status_todos: icon = status_icons.get(status, '•') todo_id = todo.get('todo_id', '')[:8] content = todo.get('content', '') active_form = todo.get('active_form', content) # Show active form for in_progress, content for others display_text = active_form if status == 'in_progress' else content parts.append(f" {icon} [{todo_id}] {display_text}") return "\n".join(parts) def _format_todo_action(self, data: Dict[str, Any]) -> str: """Format todo create/update/complete response""" success = data.get('success', True) message = data.get('message', '') todo_id = data.get('todo_id', '')[:8] if success: return f" ✓ {message}" else: error = data.get('error', 'Unknown error') return f" ✗ Todo action failed: {error}"