Spaces:
Sleeping
Sleeping
File size: 6,033 Bytes
a532053 13904c3 a532053 13904c3 a532053 |
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 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 |
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task, before_kickoff
from crewai.llm import LLM
from crewai.knowledge.source.json_knowledge_source import JSONKnowledgeSource
from dotenv import load_dotenv
import os
from typing import Dict, Any, Optional
import json
from expressly_server.utils.utils import load_json_data, sanitize_input
load_dotenv()
MODEL_API_KEY = os.getenv("MODEL_API_KEY")
MODEL = os.getenv("MODEL")
FORMAT_JSON_FILE = "format.json"
TONE_JSON_FILE = "tone.json"
TARGET_AUDIENCE_JSON_FILE = "target_audience.json"
CONTENT_STYLE_MAPPING_JSON_FILE = "content_style_mapping.json"
KNOWLEDGE_SOURCE_PATH = "knowledge"
@CrewBase
class ExpresslyServer:
"""ExpresslyServer crew"""
# Create a knowledge source
json_knowledge_source = JSONKnowledgeSource(
file_paths=["format.json", "tone.json", "target_audience.json"],
)
agents_config = "config/agents.yaml"
tasks_config = "config/tasks.yaml"
llm = LLM(model=MODEL, api_key=MODEL_API_KEY, temperature=0.7)
@before_kickoff
def validate_inputs(
self, inputs: Optional[Dict[str, Any]]
) -> Optional[Dict[str, Any]]:
"""
Validate and process user inputs based on the active tab selection.
This method checks the integrity and presence of required inputs, loads
necessary JSON data, and validates the active_tab value to ensure the
appropriate fields are populated. It formats the inputs for further processing.
Parameters:
inputs (Optional[Dict[str, Any]]): The dictionary containing user inputs,
including 'target_audience', 'format', 'tone', 'active_tab', and 'prompt'.
Returns:
Optional[Dict[str, Any]]: A dictionary formatted with context, format, tone,
and target_audience details based on the inputs provided.
Raises:
ValueError: If inputs are missing, not a dictionary, or required fields
('active_tab', 'prompt', 'format', 'tone', 'target_audience') are not provided
or invalid.
"""
if inputs is None or len(inputs) == 0 or not isinstance(inputs, dict):
raise ValueError("Inputs is required and must be a dictionary")
## Get the first element from the list of inputs and get the value of target, format, active_tab and prompt
query = inputs
target_audience: str = sanitize_input(query.get("target_audience"))
content_format: str = sanitize_input(query.get("format"))
content_tone: str = sanitize_input(query.get("tone"))
prompt: str = query.get("prompt")
# Check if prompt are not None
if prompt is None:
raise ValueError("Prompt is required")
# Load JSON data from content_style_mapping.json
content_style_mapping_json = load_json_data(
CONTENT_STYLE_MAPPING_JSON_FILE, KNOWLEDGE_SOURCE_PATH
)
tone_json = load_json_data(TONE_JSON_FILE, KNOWLEDGE_SOURCE_PATH)
format_json = load_json_data(FORMAT_JSON_FILE, KNOWLEDGE_SOURCE_PATH)
target_audience_json = load_json_data(
TARGET_AUDIENCE_JSON_FILE, KNOWLEDGE_SOURCE_PATH
)
if target_audience != "":
## Resetting the format and tone as per the target audience
mappings: dict = content_style_mapping_json.get("target_audience").get(
target_audience
)
format_dict = format_json.get("format").get(mappings.get("format"))
tone_dict = tone_json.get("tone").get(mappings.get("tone"))
target_audience_dict = target_audience_json.get("target_audience").get(
target_audience
)
elif content_format != "" and content_tone != "":
## Constructing a target_audience_dict with empty values and populating the format and tone as per the input
target_audience_dict = {
"name": "",
"description": "",
"ideal_audience": "",
}
format_dict = format_json.get("format").get(content_format)
tone_dict = tone_json.get("tone").get(content_tone)
else:
raise ValueError("Provide either target audience or format and tone")
## Format the inputs
inputs = {
"context": prompt,
"format": format_dict,
"tone": tone_dict,
"target_audience": target_audience_dict,
}
return inputs
@agent
def content_creator(self) -> Agent:
"""
Initializes and returns an Agent for content creation.
This agent is configured using predefined settings for the content creator
and utilizes a language model (LLM) for generating content. The agent
accesses a JSON knowledge source to enhance its capabilities and operates
in verbose mode for detailed output logging.
Returns:
Agent: An initialized agent configured for content creation.
"""
return Agent(
config=self.agents_config["content_creator"],
llm=self.llm,
knowledge_source=[self.json_knowledge_source],
verbose=True,
)
@task
def content_creator_task(self) -> Task:
"""
Initializes and returns a Task for content creation.
This task is configured using predefined settings for content creation
and is used by the content creator agent to generate content.
Returns:
Task: An initialized task configured for content creation.
"""
return Task(
config=self.tasks_config["content_creator_task"],
)
@crew
def crew(self) -> Crew:
"""Creates the ExpresslyServer crew"""
return Crew(
agents=self.agents,
tasks=self.tasks,
process=Process.sequential,
knowledge_source=[self.json_knowledge_source],
verbose=True,
)
|