""" Dokumentations-Agent für die Erstellung von Dokumentation, Berichten und strukturierten Inhalten. Verwendet einen adaptiven Prozess zur Erstellung hochwertiger Dokumentation basierend auf der Komplexität des Auftrags. Angepasst für das refaktorisierte Core-Modul und AgentCommunicationProtocol. """ import logging import json import re import traceback from typing import List, Dict, Any, Optional, Tuple, Union from datetime import datetime import uuid from modules.agentservice_base import BaseAgent from modules.agentservice_utils import WorkflowUtils, MessageUtils, LoggingUtils from modules.agentservice_protocol import AgentMessage, AgentCommunicationProtocol from modules.agentservice_filemanager import FileManager # Import the file manager logger = logging.getLogger(__name__) class DocumentationAgent(BaseAgent): """Agent for creating documentation and structured content""" def __init__(self): """Initialize the documentation agent""" super().__init__() self.id = "documentation_agent" self.name = "Documentation Specialist" self.type = "documentation" self.description = "Creates documentation and structured content" self.capabilities = "report_generation,documentation,content_structuring,technical_writing,knowledge_organization" self.result_format = "FormattedDocument" # Initialize AI service self.ai_service = None # Initialize document handler self.document_handler = None # Document capabilities self.supports_documents = True self.document_capabilities = ["read", "reference", "create"] self.required_context = ["document_purpose", "target_audience"] # Initialize protocol self.protocol = AgentCommunicationProtocol() # Initialize utilities self.message_utils = MessageUtils() # Track the latest generated document self.last_document = {} def get_agent_info(self) -> Dict[str, Any]: """Get agent information for agent registry""" info = super().get_agent_info() info.update({ "metadata": { "document_types": ["manual", "report", "process", "presentation", "document"], "formats": ["markdown", "text"] } }) return info def set_document_handler(self, document_handler): """Set the document handler for file operations""" self.document_handler = document_handler async def process_message(self, message: Dict[str, Any], context: Dict[str, Any] = None) -> Dict[str, Any]: """ Process a message and create documentation. Args: message: Input message context: Optional context Returns: Response with documentation """ # Extract workflow_id from context or message workflow_id = context.get("workflow_id") if context else message.get("workflow_id", "unknown") # Get or create logging_utils log_func = context.get("log_func") if context else None logging_utils = LoggingUtils(workflow_id, log_func) # Create response structure response = { "role": "assistant", "content": "", "agent_id": self.id, "agent_type": self.type, "agent_name": self.name, "result_format": self.result_format, "workflow_id": workflow_id, "documents": [] } try: # Initial status update if log_func: status_message = self.protocol.create_status_update_message( status_description="Starting document creation", sender_id=self.id, status="in_progress", progress=0.0, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) # Extract task from message task = message.get("content", "") # Detect document type - 10% progress document_type = self._detect_document_type(task) logging_utils.info(f"Creating {document_type} documentation", "execution") if log_func: status_message = self.protocol.create_status_update_message( status_description=f"Identified document type: {document_type}", sender_id=self.id, status="in_progress", progress=0.1, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) # Process any attached documents - 30% progress document_context = "" if message.get("documents"): logging_utils.info("Processing reference documents", "execution") if log_func: status_message = self.protocol.create_status_update_message( status_description="Processing reference documents", sender_id=self.id, status="in_progress", progress=0.2, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) document_context = await self._process_documents(message) # Update progress if log_func: status_message = self.protocol.create_status_update_message( status_description="Reference documents processed", sender_id=self.id, status="in_progress", progress=0.3, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) # Enhanced prompt with document context enhanced_prompt = f"{task}\n\n{document_context}" # Assess complexity - 40% progress if log_func: status_message = self.protocol.create_status_update_message( status_description="Assessing document complexity", sender_id=self.id, status="in_progress", progress=0.4, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) is_complex = await self._assess_complexity(enhanced_prompt) complexity_type = "complex" if is_complex else "simple" logging_utils.info(f"Document complexity assessment: {complexity_type}", "execution") # Generate title - 50% progress if log_func: status_message = self.protocol.create_status_update_message( status_description="Generating document title", sender_id=self.id, status="in_progress", progress=0.5, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) title = await self._generate_title(enhanced_prompt, document_type) logging_utils.info(f"Document title: {title}", "execution") # Update progress if log_func: status_message = self.protocol.create_status_update_message( status_description=f"Generating {document_type}: {title}", sender_id=self.id, status="in_progress", progress=0.6, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) # Generate content based on complexity - 70% progress if is_complex: # For complex documents, use the AI service with enhanced prompt if log_func: status_message = self.protocol.create_status_update_message( status_description=f"Creating complex {document_type} document: {title}", sender_id=self.id, status="in_progress", progress=0.7, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) content = await self._generate_complex_document(enhanced_prompt, document_type, title) logging_utils.info("Complex document generated", "execution") else: # For simple documents, use direct generation if log_func: status_message = self.protocol.create_status_update_message( status_description=f"Creating simple {document_type} document: {title}", sender_id=self.id, status="in_progress", progress=0.7, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) content = await self._generate_simple_document(enhanced_prompt, document_type, title) logging_utils.info("Simple document generated", "execution") # Finalize document - 90% progress if log_func: status_message = self.protocol.create_status_update_message( status_description="Finalizing document", sender_id=self.id, status="in_progress", progress=0.9, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) # Create a document artifact if document handler is available if self.document_handler: doc_id = f"doc_{uuid.uuid4()}" document = { "id": doc_id, "source": { "type": "generated", "id": doc_id, "name": title, "content_type": "text/markdown", "size": len(content) }, "contents": [ { "type": "text", "text": content, "is_extracted": True } ] } # Add document to response response["documents"].append(document) # Store the latest document self.last_document = document # Update response content to reference the document response["content"] = f"I've created a document titled '{title}' that contains the requested information. The document is attached to this message." # If protocol message is required, send it if context and context.get("require_protocol_message"): result_message = self.send_document_result( document_title=title, document_content=content, sender_id=self.id, receiver_id=context.get("receiver_id", "workflow"), context_id=workflow_id ) # Just log the message creation logging_utils.info(f"Created protocol result message: {result_message.id}", "execution") else: # If no document handler, just put content in response response["content"] = content # Final progress update if log_func: status_message = self.protocol.create_status_update_message( status_description="Document creation completed", sender_id=self.id, status="completed", progress=1.0, context_id=workflow_id ) log_func(workflow_id, status_message.content, "info", self.id, self.name) return response except Exception as e: error_msg = f"Error in documentation agent: {str(e)}" logging_utils.error(error_msg, "error") # Create error response using protocol error_message = self.protocol.create_error_message( error_description=error_msg, sender_id=self.id, error_type="documentation", error_details={"traceback": traceback.format_exc()}, context_id=workflow_id ) # Log error status if log_func: status_message = self.protocol.create_status_update_message( status_description=f"Error creating documentation: {str(e)}", sender_id=self.id, status="error", progress=1.0, context_id=workflow_id ) log_func(workflow_id, status_message.content, "error", self.id, self.name) # Set error in response response["content"] = f"## Error creating documentation\n\n{error_msg}\n\n```\n{traceback.format_exc()}\n```" response["status"] = "error" return response # Helper method to process document content with enhanced logging async def _process_documents(self, message: Dict[str, Any]) -> str: """Process documents in the message with detailed logging""" if not message.get("documents"): return "" document_context = "" if self.document_handler: # Use document handler to merge contents document_context = self.document_handler.merge_document_contents(message) else: # Manual processing for document in message.get("documents", []): source = document.get("source", {}) doc_name = source.get("name", "unnamed") document_context += f"\n\n--- {doc_name} ---\n" for content in document.get("contents", []): if content.get("type") == "text": document_context += content.get("text", "") # Log summary of processed documents doc_count = len(message.get("documents", [])) context_size = len(document_context) logger.info(f"Processed {doc_count} documents, extracted {context_size} characters of context") return document_context async def _assess_complexity(self, task: str) -> bool: """ Assess task complexity to determine document structure. Args: task: The task description Returns: True if complex document needed, False otherwise """ if not self.ai_service: # Default to complex if no AI service return True prompt = f""" Analyze this task and determine if it requires a complex or simple document structure: {task} Respond with only "COMPLEX" or "SIMPLE". """ try: response = await self.ai_service.call_api([ {"role": "system", "content": "You determine document complexity requirements."}, {"role": "user", "content": prompt} ]) return "COMPLEX" in response.upper() except Exception: # Default to complex on error return True async def _generate_title(self, task: str, document_type: str) -> str: """ Generate a title for the document. Args: task: The task description document_type: Type of document Returns: Generated title """ if not self.ai_service: # Default title if no AI service return f"{document_type.capitalize()} Document" prompt = f""" Create a concise, professional title for this {document_type}: {task} Respond with ONLY the title, nothing else. """ try: title = await self.ai_service.call_api([ {"role": "system", "content": "You create document titles."}, {"role": "user", "content": prompt} ]) # Clean up the title return title.strip('"\'#*- \n\t') except Exception: # Default title on error return f"{document_type.capitalize()} Document" async def _generate_complex_document(self, task: str, document_type: str, title: str) -> str: """ Generate a complex document with structure. Args: task: The task description document_type: Type of document title: Document title Returns: Generated document content """ if not self.ai_service: return f"# {title}\n\nUnable to generate complex document: AI service not available." prompt = f""" Create a comprehensive, well-structured {document_type} titled "{title}" based on: {task} The document should include: 1. A clear introduction with purpose and scope 2. Logically organized sections with headings 3. Detailed content with examples and evidence 4. A conclusion with key takeaways 5. Appropriate formatting using Markdown Format the document in Markdown with proper headings, lists, and emphasis. """ try: content = await self.ai_service.call_api([ {"role": "system", "content": "You create comprehensive, well-structured documentation."}, {"role": "user", "content": prompt} ]) # Ensure title is at the top if not content.strip().startswith("# "): content = f"# {title}\n\n{content}" return content except Exception as e: return f"# {title}\n\nError generating document: {str(e)}" async def _generate_simple_document(self, task: str, document_type: str, title: str) -> str: """ Generate a simple document without complex structure. Args: task: The task description document_type: Type of document title: Document title Returns: Generated document content """ if not self.ai_service: return f"# {title}\n\nUnable to generate document: AI service not available." prompt = f""" Create a concise, focused {document_type} titled "{title}" based on: {task} The document should be clear, precise, and to the point without complex chapter structure. Format using Markdown with appropriate headings and formatting. """ try: content = await self.ai_service.call_api([ {"role": "system", "content": "You create concise, focused documentation."}, {"role": "user", "content": prompt} ]) # Ensure title is at the top if not content.strip().startswith("# "): content = f"# {title}\n\n{content}" return content except Exception as e: return f"# {title}\n\nError generating document: {str(e)}" def _detect_document_type(self, message: str) -> str: """ Detect document type from the message. Args: message: User message Returns: Detected document type """ message = message.lower() if any(term in message for term in ["manual", "guide", "instruction", "tutorial"]): return "manual" elif any(term in message for term in ["report", "analysis", "assessment", "review"]): return "report" elif any(term in message for term in ["process", "workflow", "procedure", "steps"]): return "process" elif any(term in message for term in ["presentation", "slides", "deck"]): return "presentation" else: return "document" def send_document_result(self, document_title: str, document_content: str, sender_id: str, receiver_id: str, context_id: str = None) -> AgentMessage: """Send a document result using the protocol""" metadata = { "document_type": self._detect_document_type(document_content), "title": document_title, "created_at": datetime.now().isoformat() } return self.protocol.create_result_message( result_content=document_content, sender_id=sender_id, receiver_id=receiver_id, task_id=f"doc_{uuid.uuid4()}", output_data=metadata, result_format=self.result_format, context_id=context_id ) def send_error_message(self, error_description: str, sender_id: str, receiver_id: str = None, context_id: str = None) -> AgentMessage: """Send an error message using the protocol""" return self.protocol.create_error_message( error_description=error_description, sender_id=sender_id, receiver_id=receiver_id, error_type="documentation_error", error_details={"timestamp": datetime.now().isoformat()}, context_id=context_id ) # Singleton instance _documentation_agent = None def get_documentation_agent(): """Returns a singleton instance of the documentation agent""" global _documentation_agent if _documentation_agent is None: _documentation_agent = DocumentationAgent() return _documentation_agent