Spaces:
Paused
Paused
| import asyncio | |
| from typing import Type, Dict, Any, ClassVar | |
| from pydantic import BaseModel, Field | |
| from langchain_core.tools import BaseTool | |
| import logging | |
| from src.agents.content_enhancer import ContentEnhancer, EnhancedContent | |
| logger = logging.getLogger(__name__) | |
| # Define descriptions separately | |
| original_content_desc = "The original text content that needs enhancement." | |
| content_type_desc = "The type of content (LinkedIn profile, resume, cover letter, etc.)." | |
| enhancement_goals_desc = "Specific enhancement goals (e.g., 'more impactful', 'highlight achievements', 'SEO optimization')." | |
| # Input schema for the tool | |
| class ContentEnhancerInput(BaseModel): | |
| original_content: str = Field(description=original_content_desc) | |
| content_type: str = Field(description=content_type_desc) | |
| enhancement_goals: str = Field(description=enhancement_goals_desc) | |
| class ContentEnhancerTool(BaseTool): | |
| """Tool that enhances professional content to make it more effective and impactful.""" | |
| # Use ClassVar for class attributes that aren't meant to be Pydantic fields | |
| tool_description: ClassVar[str] = ( | |
| "Enhances professional content like LinkedIn profiles, resumes, or cover letters to make them " | |
| "more impactful, engaging, and optimized for both search algorithms and human readers. " | |
| "Use this when the user wants to improve the quality and effectiveness of their professional " | |
| "content or marketing materials." | |
| ) | |
| name: str = "content_enhancer" | |
| description: str = tool_description | |
| args_schema: Type[BaseModel] = ContentEnhancerInput | |
| def _run(self, original_content: str, content_type: str, enhancement_goals: str) -> str: | |
| """Use the tool synchronously by running the async method in a new event loop.""" | |
| try: | |
| # Create a new event loop | |
| loop = asyncio.new_event_loop() | |
| asyncio.set_event_loop(loop) | |
| # Run the async method in the loop | |
| result = loop.run_until_complete( | |
| self._arun( | |
| original_content=original_content, | |
| content_type=content_type, | |
| enhancement_goals=enhancement_goals | |
| ) | |
| ) | |
| # Close the loop | |
| loop.close() | |
| return result | |
| except Exception as e: | |
| logger.error(f"Error in ContentEnhancerTool synchronous execution: {e}", exc_info=True) | |
| return "Error: Could not enhance the content. Please try again with different content or goals." | |
| async def _arun(self, original_content: str, content_type: str, enhancement_goals: str) -> str: | |
| """Use the tool asynchronously.""" | |
| logger.info(f"ContentEnhancerTool invoked for {content_type}") | |
| try: | |
| enhancer = ContentEnhancer() # Instantiate the enhancer | |
| enhancement_result: EnhancedContent = await enhancer.enhance_content( | |
| original_content=original_content, | |
| content_type=content_type, | |
| enhancement_goals=enhancement_goals | |
| ) | |
| # Format the structured output into a readable string | |
| formatted_output = self._format_enhancement_output(enhancement_result) | |
| logger.info("ContentEnhancerTool successfully formatted enhancement.") | |
| return formatted_output | |
| except Exception as e: | |
| logger.error(f"Error in ContentEnhancerTool: {e}", exc_info=True) | |
| return "Error: Could not enhance the content." | |
| def _format_enhancement_output(self, enhancement: EnhancedContent) -> str: | |
| """Formats the structured EnhancedContent object into a readable string.""" | |
| sections = [ | |
| "# Enhanced Content", | |
| enhancement.enhanced_text, | |
| "", | |
| "## Key Improvements", | |
| "\n".join([f"- {improvement}" for improvement in enhancement.key_improvements]), | |
| "", | |
| "## SEO Keywords", | |
| ", ".join(enhancement.seo_keywords), | |
| "", | |
| f"**Tone:** {enhancement.tone_description}", | |
| "", | |
| "## Additional Suggestions", | |
| "\n".join([f"- {suggestion}" for suggestion in enhancement.additional_suggestions]) | |
| ] | |
| return "\n".join(sections) |