agents.py•8.34 kB
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
from config import OPENAI_API_KEY
from typing import Sequence
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.messages import AgentEvent, ChatMessage
from autogen_agentchat.teams import RoundRobinGroupChat
from src.helper.tav import search_tool
from .prompts import (
TITLE_AGENT_PROMPT,
TONE_AGENT_PROMPT,
TONE_AGENT_REEL_PROMPT,
CONTENT_AGENT_PROMPT,
CONTENT_AGENT_REEL_PROMPT,
INTRO_HOOK_AGENT_PROMPT,
INTRO_HOOK_REEL_AGENT_PROMPT,
FORMATTER_AGENT_PROMPT,
SPOKEN_ENGLISH_AGENT_PROMPT,
OUTRO_AGENT_PROMPT,
)
model_client = OpenAIChatCompletionClient(
model="gpt-4o",
api_key=OPENAI_API_KEY,
)
########################################################################################################################################
Master_Agent = AssistantAgent(
name="Master_Agent",
system_message="""
You are a helpful AI assistant that determines
from the previous chat whether the user is asking for a reel or a video.
If the user doesn't mention explicitly consider it a video .
If the user say that the video is 60 sec or less than that consider it is a reel.
If it is a reel say "It is a reel." else say "It is a video.
""",
model_client=model_client,
)
user_proxy = UserProxyAgent(
"user_proxy",
description=" A human admin. Give the task, and send instructions to writer to refine the content generated by the Content Generator for a video",
)
Research_Agent = AssistantAgent(
name="Research_Agent",
system_message="""
Your input is the output of the user_proxy agent.Only use the tool you have been provided with.
""",
tools=[search_tool],
model_client=model_client,
)
Tone_Agent = AssistantAgent(
name="Tone_Agent",
system_message=TONE_AGENT_PROMPT,
model_client=model_client,
)
Title_agent = AssistantAgent(
name="Title_Agent",
system_message=TITLE_AGENT_PROMPT,
model_client=model_client,
)
Intro_Hook_Agent = AssistantAgent(
name="Intro_Hook_Agent",
system_message=INTRO_HOOK_AGENT_PROMPT,
model_client=model_client,
)
Content_Agent = AssistantAgent(
name="Content_Agent",
system_message=CONTENT_AGENT_PROMPT,
model_client=model_client,
)
Formatter_Agent = AssistantAgent(
name="Formatter_Agent",
system_message=FORMATTER_AGENT_PROMPT,
model_client=model_client,
)
Outro_Agent = AssistantAgent(
name="Outro_Agent",
system_message=OUTRO_AGENT_PROMPT,
model_client=model_client,
)
Spoken_English_Agent = AssistantAgent(
name="Spoken_English_Agent",
system_message=SPOKEN_ENGLISH_AGENT_PROMPT,
model_client=model_client,
)
Intro_Hook_Agent_Reel = AssistantAgent(
name="Intro_Hook_Agent_Reel",
system_message=INTRO_HOOK_REEL_AGENT_PROMPT,
model_client=model_client,
)
Content_Agent_Reel = AssistantAgent(
name="Content_Agent_Reel",
system_message=CONTENT_AGENT_REEL_PROMPT,
model_client=model_client,
)
Tone_Agent_Reel = AssistantAgent(
name="Tone_Agent_Reel",
system_message=TONE_AGENT_REEL_PROMPT,
model_client=model_client,
)
Reviewer_Agent = AssistantAgent(
name="Reviewer",
system_message="""
You are Reviewer. Your task is to check and validate the facts in the content by Tone_Agent_Reel.
If you find the facts are not correct, either correct them or remove them. Also adjust sentent structure if you remove them. Don't do any other changes.
only return the script in text.
""",
model_client=model_client,
)
# max_messages_termination = MaxMessageTermination(max_messages=10)
# #termination = max_messages_termination
text_mention_termination = TextMentionTermination("TERMINATE")
max_messages_termination = MaxMessageTermination(max_messages=10)
# termination = text_mention_termination | max_messages_termination
selector_prompt = """Select an agent to perform task.
{roles}
Current conversation context:
{history}
Read the above conversation, then select an agent from {participants} to perform the next task.
Make sure the planner agent has assigned tasks before other agents start working.
Only select one agent.
"""
# def custom_selector_func_reel(history):
# """
# Custom selector function to determine the next speaker based on conversation history.
# """
# if not history:
# return Research_Agent # Start with the user if no history exists
# last_message = history[-1] # Get the last message
# if isinstance(last_message, ChatMessage):
# last_speaker = last_message.sender
# else:
# return None # If unable to determine, let the model decide
# # Define the speaker transition order
# speaker_order = {
# Research_Agent: Intro_Hook_Agent_Reel,
# Intro_Hook_Agent_Reel: Content_Agent_Reel,
# Content_Agent_Reel: Tone_Agent_Reel,
# Tone_Agent_Reel: Reviewer_Agent,
# Reviewer_Agent: None, # End of cycle
# }
# return speaker_order.get(last_speaker, "auto")
def selector_func(messages: Sequence[AgentEvent | ChatMessage]) -> str | None:
if messages[-1].source != Master_Agent.name:
return Master_Agent.name
return None
# team = SelectorGroupChat(
# [Master_Agent,user_proxy,Research_Agent,Intro_Hook_Agent_Reel,Content_Agent_Reel,Tone_Agent_Reel,Reviewer_Agent],
# model_client=model_client,
# termination_condition=termination,
# selector_prompt=selector_prompt,
# allow_repeated_speaker=False, # Allow an agent to speak multiple turns in a row.
# selector_func = selector_func
# )
team = RoundRobinGroupChat(
[
Research_Agent,
Intro_Hook_Agent_Reel,
Content_Agent_Reel,
Tone_Agent_Reel,
Reviewer_Agent,
],
termination_condition=max_messages_termination,
)
system_massage = """You are a helpful AI assistant to generate script. You are professional script writer for short vides. User provide list to you.
In this List there have total 10 items. All that 10 items realted to script liess topic,keywords, time, short scrip but not proffesuonl. Your work is analysis hole list and create Best script for video.
And last main thing make sure script must be More professional, crisp and easy to understand, make sure not to miss any important highlight from user input. Please give me output as script.
Script doses not contain folling things.
1. title 2. time. 3. objective. 4. Host 5. Opening Scene – Engaging Hook , likes etc stuffIn,**HOST [V.O.]**, .
to show script, More professional script contain words near about 200 words"""
Reviewer_Agent = AssistantAgent(
name="script_writer",
system_message=system_massage,
model_client=model_client,
)
st = ""
async def assistant_run(content_list):
# l1 = str(l)
response = await Reviewer_Agent.on_messages(
[TextMessage(content=content_list, source="user")],
cancellation_token=CancellationToken(),
)
return response
async def script_using_topic(task):
# global st
a = await team.run(task=task)
# a = await Console(team.run_stream(task=task))
# return(a.messages[0].content)
# result = a.decode("utf-8", errors="replace") #str(a.messages[11].content)
# for i in range(10):
# l.append(a.messages[i].content)
content_list = [message.content for message in a.messages]
# content_list = str(content_list)
string_list = [(item) for item in content_list if not isinstance(item, list)]
new_string = " ".join(string_list)
s = await assistant_run(new_string)
aa = s.chat_message.content
# #aaa = aa.encode("utf-8")
return aa.encode("latin-1", errors="ignore")
# asyncio.run(script_using_topic(
# topic="Impact of AI in the financial sector",
# video_type="SHORT",
# keywords="bank"
# ))