from agents import Agent, Runner from composio_openai_agents import ComposioToolSet, App, Action import streamlit as st from dotenv import load_dotenv import asyncio from datetime import datetime, timedelta import pandas as pd import plotly.express as px from wordcloud import WordCloud import matplotlib.pyplot as plt from io import BytesIO import base64 from pydantic import BaseModel from typing import List, Optional import os load_dotenv() max_date = datetime.now().strftime("%Y-%m-%d") min_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d") class SentimentDatum(BaseModel): sentiment: str count: int class IncidentDatum(BaseModel): source: str incidents: int class CompetitiveDatum(BaseModel): company: str mentions: int class Citation(BaseModel): source: str text: str url: str sentiment: str class DashboardOutput(BaseModel): executive_summary: str sentiment_data: List[SentimentDatum] incident_data: List[IncidentDatum] competitive_data: List[CompetitiveDatum] wordcloud_text: str citations: List[Citation] openai_key = os.environ.get("OPENAI_API_KEY") composio_key = os.environ.get("COMPOSIO_API_KEY") async def run_analysis(company_name): toolset = ComposioToolSet(api_key=composio_key) tools = toolset.get_tools(actions=[Action.TWITTER_RECENT_SEARCH, Action.REDDIT_SEARCH_ACROSS_SUBREDDITS, Action.HACKERNEWS_SEARCH_POSTS, Action.HACKERNEWS_GET_ITEM_WITH_ID]) agent = Agent( name="Social Media Analyst", instructions=f""" Your ONLY job is to deliver a crisp, actionable, clustered executive summary of new, high-utility, real-time technical incidents, security chatter, actionable customer/developer feedback for the company, and competitive intelligence. - Do NOT cover general market or sector sentiment, financial/market positioning, or broad sentiment. - Do NOT explain what you are not doing, do NOT ask for clarification, do NOT disclaim, and do NOT ask for permission or confirmation. - Just deliver the report as specified, following the structure below. - You must use Reddit, Twitter, and Hacker News to get the data. For each source, find and cluster at least 3 relevant, recent signals (if available) and include them in the summary and citations. - The search query for twitter has to be something like this (Adjust the parameters according to the context): "{company_name} to:{max_date} from:{min_date}", don't pass any other fields. Just the query and max results. - For Twitter: If no technical incidents, bugs, outages, or actionable feedback are found, instead summarize the general sentiment about {company_name} on Twitter, with clickable links to the original tweets. - For hackernews, use the search query: "{company_name}", then use the get item with id tool to get the item. - Don't give updates that the company already knows about, like product launches, new features, etc. - For competitive intelligence, look for posts, tweets, or comments where users compare {company_name} to other competitors. Summarize the key points, advantages, disadvantages, and user sentiment in these comparisons. Include citations for each comparison. - Provide more research-detailed information in both the executive summary and competitive intelligence sections. Go deeper into technical, user, and product details, and cite specific sources and findings. Report structure: 1. # Executive Summary: A crisp, actionable, clustered summary of all new, high-utility signals. No fluff, only what matters. Include detailed research and findings. 2. # Competitive Intelligence: What users are saying when comparing {company_name} to competitors. Summarize key themes, pros/cons, and user sentiment. Include citations and research-level detail. 3. # Exact Citations: For each cluster or item, provide clickable links to the original sources (tweets, posts, etc.). Everything should be cited and referenced. Focus only on the tech side and actionable intelligence. You will be given the name of the company. If tool call fails, try again with different permutations of the query. """, tools=tools, model='gpt-4.1', ) res = await Runner.run(starting_agent=agent, input=f"What is the summary of {company_name}?") return res.final_output async def dashboard_result(result): agent = Agent( name="Dashboard Generator", instructions=f""" Your job is to generate a dashboard from the given result. Competitors include direct competitors only. """, output_type=DashboardOutput, ) res = await Runner.run(starting_agent=agent, input=result) return res.final_output name_from_url = st.query_params.get("name", "") company_from_url = st.query_params.get("company", "") if name_from_url: st.title(f"{name_from_url}'s Market Pulse Agent ") else: st.title("Market Pulse Agent") company_name = st.text_input("Company name:", value=company_from_url) if st.button("Analyze"): if company_name: with st.spinner("Analyzing social media sentiment..."): result = asyncio.run(run_analysis(company_name)) dashboard_result_response = asyncio.run(dashboard_result(str(result))) st.subheader("Summary and Sentiment Analysis") st.write(result) st.markdown("---") st.header("📊 Dashboard") # Metrics Row metric1, metric2, metric3 = st.columns(3) with metric1: st.metric("Total Mentions", sum([s.count for s in dashboard_result_response.sentiment_data])) with metric2: st.metric("Incidents", sum([i.incidents for i in dashboard_result_response.incident_data])) with metric3: st.metric("Competitors Compared", len(dashboard_result_response.competitive_data)-1) # Main Grid grid1, grid2 = st.columns([2, 1]) with grid1: st.subheader(":green[Sentiment Breakdown]") df_sentiment = pd.DataFrame([s.model_dump() for s in dashboard_result_response.sentiment_data]) fig = px.pie(df_sentiment, names='sentiment', values='count', color='sentiment', color_discrete_map={"Positive": "#22c55e", "Negative": "#ef4444", "Neutral": "#a3a3a3"}) fig.update_traces(textinfo='percent+label', pull=[0.05, 0.05, 0.05]) st.plotly_chart(fig, use_container_width=True) # Competitive Intelligence full width st.subheader(":blue[Competitive Intelligence]") df_comp = pd.DataFrame([c.model_dump() for c in dashboard_result_response.competitive_data]) fig3 = px.bar(df_comp, x='company', y='mentions', color='company', text='mentions', color_discrete_sequence=px.colors.qualitative.Pastel) fig3.update_layout(yaxis_title=None, xaxis_title=None, plot_bgcolor='#f8fafc') st.plotly_chart(fig3, use_container_width=True) with grid2: if sum([i.incidents for i in dashboard_result_response.incident_data]) > 0: st.subheader(":orange[Incident Frequency]") df_incident = pd.DataFrame([i.model_dump() for i in dashboard_result_response.incident_data]) fig2 = px.bar(df_incident, x='source', y='incidents', color='source', text='incidents', color_discrete_sequence=px.colors.qualitative.Set2) fig2.update_layout(yaxis_title=None, xaxis_title=None, plot_bgcolor='#f8fafc') st.plotly_chart(fig2, use_container_width=True) st.markdown("
", unsafe_allow_html=True) st.subheader(":gray[Top Citations]") df_cite = pd.DataFrame([c.model_dump() for c in dashboard_result_response.citations]) def make_clickable(url): return f'link' df_cite['url'] = df_cite['url'].apply(make_clickable) st.write(df_cite.to_html(escape=False, index=False, classes='tw-table tw-table-auto tw-bg-white tw-shadow-md tw-rounded-lg'), unsafe_allow_html=True) else: st.warning("Please enter a company name.")