admin-valentin/awesome-llm-apps
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

update : multi-agent with file ss upload

Browse source
This commit is contained in:
priyanshm07 2025-04-01 23:02:55 +05:30
parent 3735946ad0
commit a25e103189
6 changed files with 167 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

BIN
ai_agent_tutorials/ai_breakup_recovery_agent/__pycache__/agents.cpython-39.pyc Normal file
View file

Binary file not shown.

BIN
ai_agent_tutorials/ai_breakup_recovery_agent/__pycache__/image_input.cpython-39.pyc Normal file
View file

Binary file not shown.

32
ai_agent_tutorials/ai_breakup_recovery_agent/agents.py
View file

@ -1,9 +1,10 @@
from agno.agent import Agent
from agno.team.team import Team
from agno.models.openai import OpenAIChat
from agno.models.google import Gemini
from agno.tools.duckduckgo import DuckDuckGoTools
from PIL import Image
import easyocr
import os
# --- Therapist Agent ---
def create_therapist_agent():
@ -16,7 +17,10 @@ def create_therapist_agent():
"Offer coping strategies without being dismissive.",
"Validate their experiences and provide emotional support.",
],
model=OpenAIChat(id="gpt-4o-mini"),
model=Gemini(
id="gemini-2.0-flash",
api_key=os.environ.get("GEMINI_API_KEY")
),
tools=[DuckDuckGoTools()],
add_datetime_to_instructions=True, # Add timestamp context
show_tool_calls=True,
@ -31,10 +35,13 @@ def create_closure_agent():
instructions=[
"Create emotional messages that express raw, honest feelings.",
"The messages should NOT be sent — they are only for emotional release.",
"Format the output clearly with a header: **Message Drafts You Shouldn’t Send**",
"Format the output clearly with a header: **Message Drafts You Shouldn't Send**",
"Ensure the tone is heartfelt, authentic, and honest.",
],
model=OpenAIChat(id="gpt-4o-mini"),
model=Gemini(
id="gemini-2.0-flash",
api_key=os.environ.get("GEMINI_API_KEY")
),
tools=[DuckDuckGoTools()],
add_datetime_to_instructions=True, # Add timestamp context
show_tool_calls=True,
@ -55,7 +62,10 @@ def create_routine_agent():
"Include healthy distractions, such as hobbies, reading, or new experiences.",
"Format the output clearly with a header: **Daily Recovery Routine**",
],
model=OpenAIChat(id="gpt-4o-mini"),
model=Gemini(
id="gemini-2.0-flash",
api_key=os.environ.get("GEMINI_API_KEY")
),
tools=[DuckDuckGoTools()],
add_datetime_to_instructions=True,
show_tool_calls=True,
@ -75,7 +85,11 @@ def create_honesty_agent():
"Include reasons why the user should move on, based on the situation.",
"Format the output clearly with a header: **Brutal Honesty**",
],
model=OpenAIChat(id="gpt-4o-mini"),
# model=Gemini(id="gemini-2.0-flash"),
model=Gemini(
id="gemini-2.0-flash",
api_key=os.environ.get("GEMINI_API_KEY")
),
tools=[DuckDuckGoTools()],
add_datetime_to_instructions=True,
show_tool_calls=True,
@ -98,7 +112,11 @@ def create_breakup_team():
return Team(
name="Breakup Recovery Team",
mode="coordinate", # Team execution mode: coordinate or parallel
model=OpenAIChat(id="gpt-4o-mini"),
# model=Gemini(id="gemini-2.0-flash"),
model=Gemini(
id="gemini-2.0-flash",
api_key=os.environ.get("GEMINI_API_KEY")
),
members=[
create_therapist_agent(),
create_closure_agent(),

91
ai_agent_tutorials/ai_breakup_recovery_agent/app.py
View file

@ -1,5 +1,6 @@
import streamlit as st
from agents import create_breakup_team
from image_input import analyze_chat_screenshot
from PIL import Image
# --- Streamlit Layout ---
@ -7,16 +8,16 @@ st.title("💔 Breakup Recovery Agent Team")
st.write("Receive support and emotional outlet messages from specialized AI agents.")
# --- API Keys Input ---
openai_api_key = st.sidebar.text_input("OpenAI API Key", type="password")
gemini_api_key = st.sidebar.text_input("Gemini API Key", type="password")
submit_button = st.sidebar.button("Submit")
# --- Store API Keys ---
if submit_button:
if openai_api_key:
st.session_state["openai_api_key"] = openai_api_key
st.success("API keys saved successfully!")
if gemini_api_key:
st.session_state["gemini_api_key"] = gemini_api_key
st.success("API key saved successfully!")
else:
st.error("Please enter both API keys!")
st.error("Please enter the API key!")
# --- User Input ---
user_input = st.text_area("Describe how you're feeling...")
@ -24,47 +25,65 @@ user_input = st.text_area("Describe how you're feeling...")
# --- Image Upload ---
uploaded_image = st.file_uploader("Upload a chat screenshot (optional)", type=["png", "jpg", "jpeg"])
# --- Display Uploaded Image ---
extracted_text = ""
# --- Display Uploaded Image and Analysis ---
if uploaded_image:
image = Image.open(uploaded_image)
st.image(image, caption="Uploaded Screenshot", use_column_width=True)
if st.button("Analyze Chat"):
with st.spinner("Analyzing chat patterns..."):
if not st.session_state.get("gemini_api_key"):
st.error("Please enter your Gemini API key in the sidebar first.")
else:
try:
# Set environment variable for the analysis
import os
os.environ["GEMINI_API_KEY"] = st.session_state["gemini_api_key"]
# Analyze the chat screenshot
analysis = analyze_chat_screenshot(uploaded_image)
# Display the analysis
st.subheader("🔍 Chat Analysis")
st.markdown(analysis)
except Exception as e:
st.error(f"An error occurred during analysis: {str(e)}")
st.info("Please check your Gemini API key and try again.")
# # ✅ Extract text from the image
# extracted_text = extract_text_with_easyocr(image)
# st.text_area("Extracted Text from Image:", extracted_text)
# --- Execute Agent Team ---
# --- Execute Agent Team (only for text input) ---
if st.button("Get Recovery Support"):
with st.spinner("Agents are processing..."):
if not user_input and not uploaded_image:
if not user_input:
st.error("Please enter what you feel.")
elif not st.session_state.get("gemini_api_key"):
st.error("Please enter your Gemini API key in the sidebar first.")
else:
breakup_team = create_breakup_team()
try:
breakup_team = create_breakup_team()
# Set environment variables
import os
os.environ["OPENAI_API_KEY"] = st.session_state.get("openai_api_key", "")
# Set environment variables
import os
os.environ["GEMINI_API_KEY"] = st.session_state["gemini_api_key"]
# Prepare the input
input_data = user_input
if uploaded_image:
input_data += "\n\n[Attached Image Included]" # Add image indicator
# Execute the team with text input
response = breakup_team.run(user_input)
# Execute the team
response = breakup_team.run(user_input)
# Display responses
st.subheader("💡 Team's Responses")
# Display responses
st.subheader("💡 Team's Responses")
# Display individual agent responses
if response.member_responses:
for index, member_response in enumerate(response.member_responses, start=0):
st.markdown(f"### 🛡️ {response.tools[index]['tool_args']['agent_name']}")
st.markdown(member_response.content)
else:
st.warning("⚠️ No individual agent responses received. This might be due to an API error or configuration issue.")
st.info("Please check your Gemini API key and try again.")
# Display individual agent responses
if response.member_responses:
for member_response in response.member_responses:
st.markdown(f"### 🛡️ {member_response.agent_id}")
st.write(member_response.content)
else:
st.warning("⚠️ No individual agent responses received. Showing team leader's summary only.")
# Display team leader's final summary
st.subheader("📜 Team Leader's Summary")
st.write(response.content)
# Display team leader's final summary
if response.content:
st.subheader("📜 Team Leader's Summary")
st.markdown(response.content)
except Exception as e:
st.error(f"An error occurred: {str(e)}")
st.info("Please check your Gemini API key and try again.")

86
ai_agent_tutorials/ai_breakup_recovery_agent/image_input.py Normal file
View file

@ -0,0 +1,86 @@
from agno.agent import Agent
from agno.media import Image
from agno.models.google import Gemini
from agno.tools.duckduckgo import DuckDuckGoTools
from PIL import Image as PILImage
import base64
from io import BytesIO
import os
def load_image(image_data):
"""Load and prepare image using PIL"""
# Convert image data to PIL Image
img = PILImage.open(image_data).convert('RGB')
# Convert PIL image to bytes
buffered = BytesIO()
img.save(buffered, format="JPEG")
img_bytes = buffered.getvalue()
return img_bytes
# # Check for Google API key
# if "GOOGLE_API_KEY" not in os.environ:
# raise EnvironmentError(
# "Please set your GOOGLE_API_KEY environment variable. "
# "You can get one from https://makersuite.google.com/app/apikey"
# )
def create_chat_analysis_agent():
"""Creates a specialized agent for analyzing chat patterns"""
return Agent(
name="Chat Analysis Agent",
role="You are a specialized agent that analyzes chat screenshots for relationship patterns and communication issues.",
instructions=[
"Analyze the chat screenshot for the following patterns:",
"1. Communication Patterns: Identify recurring themes, response times, and conversation flow",
"2. Passive-Aggression: Look for subtle hostile or manipulative language",
"3. Manipulation: Identify gaslighting, guilt-tripping, or controlling behaviors",
"4. Mixed Signals: Detect inconsistent messages or unclear intentions",
"Provide a detailed analysis with specific examples from the chat",
"Format the output clearly with sections for each pattern type",
"Be objective and focus on observable patterns rather than making assumptions",
],
model=Gemini(
id="gemini-2.0-flash",
api_key=os.environ.get("GEMINI_API_KEY")
),
tools=[DuckDuckGoTools()],
markdown=True,
)
def analyze_chat_screenshot(image_data):
"""Analyze a chat screenshot using the specialized agent"""
try:
# Load and process the image
image_bytes = load_image(image_data)
# Create the chat analysis agent
agent = create_chat_analysis_agent()
# Analyze the chat screenshot
response = agent.run(
"Please analyze this chat screenshot for communication patterns, passive-aggression, manipulation, and mixed signals.",
images=[Image(content=image_bytes)]
)
return response.content
except Exception as e:
return f"Error analyzing chat screenshot: {str(e)}"
# Example local image path - replace with your image path
image_path = "breakup.jpg"
try:
# Load and process the image
image_data = load_image(image_path)
# Create the response using the agent
agent.print_response(
"Tell me about this image and give me the latest news about it.",
images=[Image(content=image_data)],
stream=True,
)
except Exception as e:
print(f"Error processing image: {e}")

2
ai_agent_tutorials/ai_breakup_recovery_agent/requirements.txt
View file

@ -1,3 +1,3 @@
agno==1.2.4
agno>=1.2.4
pillow==11.1.0
streamlit==1.44.0