Spaces:
Paused
Paused
File size: 4,343 Bytes
403ebf9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | 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) |