Spaces:
Sleeping
Sleeping
| """ | |
| GenAI CEO Crew v5 - LLM-Driven Reasoning Architecture | |
| ====================================================== | |
| CORE PRINCIPLE: No hardcoding. Every decision is made by LLM reasoning. | |
| FLOW: | |
| 1. CEO Agent analyzes question → reasons about which experts are needed | |
| 2. Each selected Agent reasons about what tools (if any) they need | |
| 3. Agents use tools or answer from knowledge | |
| 4. Debate and synthesis | |
| NO KEYWORDS, NO MAPPINGS - Pure reasoning. | |
| """ | |
| import gradio as gr | |
| import os | |
| import json | |
| import re | |
| from typing import List, Dict, Any, Optional, Callable | |
| from dataclasses import dataclass, field | |
| from datetime import datetime | |
| import time | |
| import requests | |
| from concurrent.futures import ThreadPoolExecutor, as_completed | |
| import plotly.express as px | |
| import plotly.graph_objects as go | |
| import pandas as pd | |
| # ============================================================================ | |
| # PROGRESS STREAM | |
| # ============================================================================ | |
| class ProgressStream: | |
| def __init__(self): | |
| self.messages = [] | |
| def log(self, message: str, indent: int = 0): | |
| timestamp = datetime.now().strftime('%H:%M:%S') | |
| prefix = " " * indent | |
| formatted = f"[{timestamp}] {prefix}{message}" | |
| self.messages.append(formatted) | |
| print(formatted) | |
| def phase(self, name: str): | |
| self.messages.append(f"\n{'='*50}") | |
| self.messages.append(f"📍 {name}") | |
| self.messages.append(f"{'='*50}") | |
| print(f"\n📍 {name}") | |
| def get_log(self) -> str: | |
| return "\n".join(self.messages) | |
| def clear(self): | |
| self.messages = [] | |
| progress = ProgressStream() | |
| # ============================================================================ | |
| # LLM PROVIDER | |
| # ============================================================================ | |
| class LLM: | |
| def __init__(self): | |
| self.provider = os.getenv("LLM_PROVIDER", "huggingface").lower() | |
| self.model = "Qwen/Qwen2.5-72B-Instruct" if self.provider == "huggingface" else "llama-3.3-70b-versatile" | |
| def generate(self, messages: list, temperature: float = 0.7) -> str: | |
| if self.provider == "groq": | |
| return self._groq(messages, temperature) | |
| return self._huggingface(messages, temperature) | |
| def _huggingface(self, messages: list, temp: float) -> str: | |
| from huggingface_hub import InferenceClient | |
| token = os.getenv("HF_TOKEN", "").strip() or None | |
| client = InferenceClient(model=self.model, token=token) | |
| resp = client.chat_completion(messages=messages, max_tokens=4096, temperature=temp) | |
| return resp.choices[0].message.content | |
| def _groq(self, messages: list, temp: float) -> str: | |
| key = os.getenv("GROQ_API_KEY", "").strip() | |
| resp = requests.post( | |
| "https://api.groq.com/openai/v1/chat/completions", | |
| headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, | |
| json={"model": self.model, "messages": messages, "max_tokens": 4096, "temperature": temp}, | |
| timeout=120 | |
| ) | |
| resp.raise_for_status() | |
| return resp.json()["choices"][0]["message"]["content"] | |
| llm = LLM() | |
| # ============================================================================ | |
| # TOOLS - Available tools that agents can choose to use | |
| # ============================================================================ | |
| class Tools: | |
| """Collection of tools available to agents""" | |
| def get_available_tools() -> str: | |
| """Return description of available tools for agent reasoning""" | |
| return """ | |
| AVAILABLE TOOLS: | |
| 1. YAHOO_FINANCE | |
| - Purpose: Get real-time stock data, financial metrics, company fundamentals | |
| - Use when: You need current stock price, market cap, P/E ratio, revenue, profit margins, etc. | |
| - Requires: Company name or stock ticker symbol | |
| - Returns: Stock price, market cap, P/E, revenue, margins, cash flow, debt, employee count, etc. | |
| 2. WEB_SEARCH | |
| - Purpose: Search the internet for current information | |
| - Use when: You need recent news, market trends, competitor info, industry analysis | |
| - Requires: Search query | |
| - Returns: List of relevant web pages with titles and snippets | |
| 3. NEWS_SEARCH | |
| - Purpose: Search for recent news articles | |
| - Use when: You need latest news about a company, industry, or topic | |
| - Requires: Search query | |
| - Returns: Recent news articles with headlines, sources, and dates | |
| NOTE: You don't HAVE to use tools. If the question can be answered from your expertise and general knowledge, you can skip tools entirely. | |
| """ | |
| def yahoo_finance_search_ticker(input_text: str) -> Optional[str]: | |
| """Extract company name from input, then search for its ticker""" | |
| progress.log(f"🔍 Processing input: {input_text}", indent=2) | |
| # Step 1: Extract the actual company name from input | |
| company_name = Tools.extract_company_name(input_text) | |
| if not company_name: | |
| progress.log(f"❌ No valid company name found in: {input_text}", indent=2) | |
| return None | |
| progress.log(f"📌 Extracted company: {company_name}", indent=2) | |
| # Step 2: Search for ticker | |
| try: | |
| from duckduckgo_search import DDGS | |
| search_query = f"{company_name} stock ticker symbol" | |
| progress.log(f"🔍 Searching: {search_query}", indent=2) | |
| with DDGS() as ddgs: | |
| results = list(ddgs.text(search_query, max_results=5)) | |
| if not results: | |
| progress.log(f"❌ No search results for {company_name}", indent=2) | |
| return None | |
| # Compile results | |
| search_text = "\n".join([ | |
| f"- {r.get('title', '')}: {r.get('body', '')}" | |
| for r in results | |
| ])[:2000] | |
| # Step 3: Ask LLM to extract ticker from search results | |
| prompt = f"""From these search results, extract the stock ticker symbol for "{company_name}". | |
| Search Results: | |
| {search_text} | |
| Rules: | |
| - Return ONLY the ticker symbol (e.g., AAPL, MSFT, ZM, SUBEX.NS) | |
| - For Indian stocks on NSE, add .NS suffix (e.g., SUBEX.NS, INFY.NS) | |
| - For Indian stocks on BSE, add .BO suffix | |
| - If cannot find, return NOT_FOUND | |
| Ticker:""" | |
| response = llm.generate([{"role": "user", "content": prompt}], temperature=0.1) | |
| ticker = response.strip().split()[0].upper().replace('"', '').replace("'", "") if response else "" | |
| if ticker and ticker != "NOT_FOUND" and len(ticker) <= 15: | |
| # Verify ticker | |
| if Tools._verify_ticker(ticker): | |
| progress.log(f"✅ Found ticker: {ticker}", indent=2) | |
| return ticker | |
| # Try with .NS | |
| if Tools._verify_ticker(f"{ticker}.NS"): | |
| progress.log(f"✅ Found ticker: {ticker}.NS", indent=2) | |
| return f"{ticker}.NS" | |
| # Try with .BO | |
| if Tools._verify_ticker(f"{ticker}.BO"): | |
| progress.log(f"✅ Found ticker: {ticker}.BO", indent=2) | |
| return f"{ticker}.BO" | |
| progress.log(f"❌ Could not find valid ticker", indent=2) | |
| return None | |
| except Exception as e: | |
| progress.log(f"❌ Ticker search error: {e}", indent=2) | |
| return None | |
| def extract_company_name(input_text: str) -> Optional[str]: | |
| """Use LLM to extract actual company name from input text""" | |
| extract_prompt = f"""Extract the company name from this text that we need to look up stock/financial data for. | |
| Text: "{input_text}" | |
| Rules: | |
| - Return ONLY the company name, nothing else | |
| - Ignore geographic locations (India, USA, Europe, etc.) - those are NOT companies | |
| - Ignore generic words like "acquire", "buy", "target", "merger", etc. | |
| - Ignore descriptions like "telecom company", "tech firm", etc. | |
| - If multiple companies mentioned, return the PRIMARY one being researched | |
| - If no valid company name found, return "NONE" | |
| Examples: | |
| - "Subex in India" → Subex | |
| - "acquire Microsoft" → Microsoft | |
| - "the Indian telecom company" → NONE | |
| - "Zoom video communications" → Zoom | |
| Company name:""" | |
| try: | |
| company_name = llm.generate([{"role": "user", "content": extract_prompt}], temperature=0.1) | |
| company_name = company_name.strip().strip('"').strip("'").strip() | |
| if company_name and company_name.upper() != "NONE" and len(company_name) >= 2: | |
| return company_name | |
| return None | |
| except: | |
| return None | |
| def _verify_ticker(ticker: str) -> bool: | |
| """Verify ticker with yfinance""" | |
| try: | |
| import yfinance as yf | |
| info = yf.Ticker(ticker).info | |
| return bool(info.get('regularMarketPrice') or info.get('currentPrice') or info.get('previousClose')) | |
| except: | |
| return False | |
| def yahoo_finance_get_data(ticker: str) -> Dict[str, Any]: | |
| """Get financial data for a ticker""" | |
| progress.log(f"📊 Fetching data for {ticker}", indent=2) | |
| try: | |
| import yfinance as yf | |
| t = yf.Ticker(ticker) | |
| info = t.info | |
| if not info.get('regularMarketPrice') and not info.get('currentPrice'): | |
| progress.log(f"❌ No data for {ticker}", indent=2) | |
| return {"success": False, "error": "No data found"} | |
| def fmt_num(n): | |
| if n is None: return "N/A" | |
| if abs(n) >= 1e12: return f"${n/1e12:.2f}T" | |
| if abs(n) >= 1e9: return f"${n/1e9:.2f}B" | |
| if abs(n) >= 1e6: return f"${n/1e6:.1f}M" | |
| return f"${n:,.0f}" | |
| def fmt_pct(n): | |
| return f"{n*100:.1f}%" if n else "N/A" | |
| data = { | |
| "success": True, | |
| "ticker": ticker, | |
| "company": info.get("longName") or info.get("shortName") or ticker, | |
| "price": info.get("currentPrice") or info.get("regularMarketPrice"), | |
| "market_cap": fmt_num(info.get("marketCap")), | |
| "enterprise_value": fmt_num(info.get("enterpriseValue")), | |
| "pe_ratio": info.get("trailingPE"), | |
| "forward_pe": info.get("forwardPE"), | |
| "peg_ratio": info.get("pegRatio"), | |
| "price_to_book": info.get("priceToBook"), | |
| "ev_to_revenue": info.get("enterpriseToRevenue"), | |
| "ev_to_ebitda": info.get("enterpriseToEbitda"), | |
| "revenue": fmt_num(info.get("totalRevenue")), | |
| "gross_margin": fmt_pct(info.get("grossMargins")), | |
| "operating_margin": fmt_pct(info.get("operatingMargins")), | |
| "profit_margin": fmt_pct(info.get("profitMargins")), | |
| "revenue_growth": fmt_pct(info.get("revenueGrowth")), | |
| "total_cash": fmt_num(info.get("totalCash")), | |
| "total_debt": fmt_num(info.get("totalDebt")), | |
| "debt_to_equity": info.get("debtToEquity"), | |
| "free_cash_flow": fmt_num(info.get("freeCashflow")), | |
| "employees": info.get("fullTimeEmployees"), | |
| "sector": info.get("sector"), | |
| "industry": info.get("industry"), | |
| "52_week_high": info.get("fiftyTwoWeekHigh"), | |
| "52_week_low": info.get("fiftyTwoWeekLow"), | |
| "description": (info.get("longBusinessSummary") or "")[:500], | |
| "url": f"https://finance.yahoo.com/quote/{ticker}" | |
| } | |
| progress.log(f"✅ {data['company']}: ${data['price']} | MCap: {data['market_cap']}", indent=2) | |
| return data | |
| except Exception as e: | |
| progress.log(f"❌ Error: {e}", indent=2) | |
| return {"success": False, "error": str(e)} | |
| def web_search(query: str) -> Dict[str, Any]: | |
| """Search the web""" | |
| progress.log(f"🔍 Web search: {query}", indent=2) | |
| try: | |
| from duckduckgo_search import DDGS | |
| with DDGS() as ddgs: | |
| results = list(ddgs.text(query, max_results=5)) | |
| formatted = [{"title": r.get("title", ""), "snippet": r.get("body", "")[:200], "url": r.get("href", "")} for r in results] | |
| progress.log(f"✅ Found {len(formatted)} results", indent=2) | |
| return {"success": True, "results": formatted} | |
| except Exception as e: | |
| progress.log(f"❌ Error: {e}", indent=2) | |
| return {"success": False, "error": str(e), "results": []} | |
| def news_search(query: str) -> Dict[str, Any]: | |
| """Search for news""" | |
| progress.log(f"📰 News search: {query}", indent=2) | |
| try: | |
| from duckduckgo_search import DDGS | |
| with DDGS() as ddgs: | |
| results = list(ddgs.news(query, max_results=5)) | |
| formatted = [{"title": r.get("title", ""), "source": r.get("source", ""), "date": r.get("date", ""), "url": r.get("url", "")} for r in results] | |
| progress.log(f"✅ Found {len(formatted)} articles", indent=2) | |
| return {"success": True, "articles": formatted} | |
| except Exception as e: | |
| progress.log(f"❌ Error: {e}", indent=2) | |
| return {"success": False, "error": str(e), "articles": []} | |
| # ============================================================================ | |
| # AGENT PROFILES - Descriptions for LLM to understand each role | |
| # ============================================================================ | |
| AGENT_PROFILES = { | |
| "cfo": { | |
| "id": "cfo", | |
| "title": "Chief Financial Officer (CFO)", | |
| "icon": "💰", | |
| "description": """The CFO is the senior executive responsible for managing the financial health of the organization. | |
| EXPERTISE: | |
| - Corporate finance and capital structure | |
| - Financial planning, analysis, and forecasting | |
| - Valuation and deal structuring (M&A) | |
| - Accounting, reporting, and compliance | |
| - Treasury, cash management, and investor relations | |
| - Cost management and profitability analysis | |
| WHEN TO INVOLVE: | |
| - Questions involving company valuation, stock price, financial metrics | |
| - M&A transactions, investments, or divestitures | |
| - Financial health assessment (revenue, margins, cash flow, debt) | |
| - Capital allocation decisions | |
| - Cost-benefit analysis of strategic initiatives | |
| - Financial due diligence""" | |
| }, | |
| "cso": { | |
| "id": "cso", | |
| "title": "Chief Strategy Officer (CSO)", | |
| "icon": "♟️", | |
| "description": """The CSO leads the development and execution of corporate strategy. | |
| EXPERTISE: | |
| - Corporate strategy and long-term planning | |
| - Market analysis and competitive positioning | |
| - Strategic partnerships and alliances | |
| - Business model innovation | |
| - Portfolio strategy and resource allocation | |
| - Strategic transformation initiatives | |
| WHEN TO INVOLVE: | |
| - Strategic direction and vision questions | |
| - Market entry or exit decisions | |
| - Competitive response strategies | |
| - Business model changes | |
| - Strategic fit assessment for M&A | |
| - Long-term growth planning""" | |
| }, | |
| "cto": { | |
| "id": "cto", | |
| "title": "Chief Technology Officer (CTO)", | |
| "icon": "⚙️", | |
| "description": """The CTO oversees all technology strategy, architecture, and innovation. | |
| EXPERTISE: | |
| - Technology strategy and roadmap | |
| - Software architecture and platforms | |
| - Engineering team leadership | |
| - Technical due diligence | |
| - Innovation and R&D | |
| - Technology partnerships and vendor management | |
| - Cybersecurity oversight | |
| WHEN TO INVOLVE: | |
| - Technology stack decisions | |
| - Technical feasibility assessments | |
| - Technology integration (M&A, partnerships) | |
| - Build vs buy decisions | |
| - AI/ML strategy | |
| - Platform scalability questions | |
| - Technical talent assessment""" | |
| }, | |
| "clo": { | |
| "id": "clo", | |
| "title": "Chief Legal Officer (CLO)", | |
| "icon": "⚖️", | |
| "description": """The CLO is the top legal executive responsible for all legal matters. | |
| EXPERTISE: | |
| - Corporate law and governance | |
| - Regulatory compliance | |
| - Antitrust and competition law | |
| - Intellectual property | |
| - Contracts and negotiations | |
| - Litigation management | |
| - Risk management (legal) | |
| WHEN TO INVOLVE: | |
| - Regulatory approval requirements | |
| - Antitrust concerns in M&A | |
| - Legal risks and liabilities | |
| - Contract negotiations | |
| - Intellectual property issues | |
| - Compliance requirements | |
| - Litigation exposure""" | |
| }, | |
| "cmo": { | |
| "id": "cmo", | |
| "title": "Chief Marketing Officer (CMO)", | |
| "icon": "📣", | |
| "description": """The CMO leads all marketing, brand, and customer strategy. | |
| EXPERTISE: | |
| - Brand strategy and management | |
| - Customer acquisition and retention | |
| - Market research and insights | |
| - Product marketing | |
| - Digital marketing and analytics | |
| - Communications and PR | |
| WHEN TO INVOLVE: | |
| - Brand value and perception questions | |
| - Customer impact of strategic decisions | |
| - Market positioning changes | |
| - Go-to-market strategy | |
| - Customer synergies in M&A | |
| - Competitive messaging""" | |
| }, | |
| "chro": { | |
| "id": "chro", | |
| "title": "Chief Human Resources Officer (CHRO)", | |
| "icon": "👥", | |
| "description": """The CHRO leads all people and organizational strategy. | |
| EXPERTISE: | |
| - Talent acquisition and retention | |
| - Organizational design | |
| - Culture and change management | |
| - Compensation and benefits | |
| - Leadership development | |
| - HR operations and compliance | |
| WHEN TO INVOLVE: | |
| - Talent and team questions | |
| - Organizational restructuring | |
| - Culture integration in M&A | |
| - Key personnel retention | |
| - Workforce planning | |
| - Change management""" | |
| }, | |
| "cro_risk": { | |
| "id": "cro_risk", | |
| "title": "Chief Risk Officer (CRO)", | |
| "icon": "🛡️", | |
| "description": """The CRO identifies, assesses, and mitigates enterprise risks. | |
| EXPERTISE: | |
| - Enterprise risk management | |
| - Operational risk | |
| - Financial risk | |
| - Strategic risk | |
| - Compliance risk | |
| - Crisis management | |
| WHEN TO INVOLVE: | |
| - Risk identification and assessment | |
| - Downside scenario analysis | |
| - Risk mitigation strategies | |
| - Crisis response planning | |
| - Due diligence (risk perspective) | |
| - Regulatory risk""" | |
| }, | |
| "coo": { | |
| "id": "coo", | |
| "title": "Chief Operating Officer (COO)", | |
| "icon": "🔧", | |
| "description": """The COO oversees day-to-day operations and execution. | |
| EXPERTISE: | |
| - Operations management | |
| - Process optimization | |
| - Supply chain management | |
| - Quality and efficiency | |
| - Operational integration (M&A) | |
| - Scaling operations | |
| WHEN TO INVOLVE: | |
| - Operational feasibility | |
| - Integration planning | |
| - Process synergies | |
| - Operational due diligence | |
| - Execution planning | |
| - Efficiency improvements""" | |
| }, | |
| } | |
| def get_all_profiles_description() -> str: | |
| """Get formatted description of all agents for team selection""" | |
| lines = ["AVAILABLE EXPERTS:\n"] | |
| for pid, profile in AGENT_PROFILES.items(): | |
| lines.append(f"{profile['icon']} {profile['title']}") | |
| lines.append(f" ID: {profile['id']}") | |
| lines.append(f" {profile['description'][:300]}...") | |
| lines.append("") | |
| return "\n".join(lines) | |
| # ============================================================================ | |
| # AGENT EXECUTION | |
| # ============================================================================ | |
| class ToolCall: | |
| tool: str | |
| input: str | |
| output: str | |
| success: bool | |
| class AgentOutput: | |
| agent_id: str | |
| title: str | |
| icon: str | |
| reasoning: str # Agent's reasoning about tools needed | |
| tool_calls: List[ToolCall] = field(default_factory=list) | |
| analysis: str = "" | |
| citations: List[Dict] = field(default_factory=list) | |
| confidence: float = 0.0 | |
| execution_time: float = 0.0 | |
| def run_agent(agent_id: str, question: str, task: str) -> AgentOutput: | |
| """Run a single agent with reasoning about tools""" | |
| profile = AGENT_PROFILES[agent_id] | |
| start_time = time.time() | |
| progress.log(f"🚀 {profile['icon']} {profile['title']} starting...") | |
| # Step 1: Agent reasons about what tools (if any) they need | |
| progress.log(f"🤔 Reasoning about required tools...", indent=1) | |
| tool_reasoning_prompt = f"""You are the {profile['title']}. | |
| {profile['description']} | |
| CEO's QUESTION: {question} | |
| YOUR SPECIFIC TASK: {task} | |
| {Tools.get_available_tools()} | |
| THINK THROUGH: | |
| 1. What information do I need to provide a high-quality answer? | |
| 2. Do I need real-time data (stock prices, financials) or can I answer from expertise? | |
| 3. Do I need recent news or market information? | |
| Based on your reasoning, decide which tools (if any) you need. | |
| Respond in this JSON format: | |
| {{ | |
| "reasoning": "Your reasoning about what information you need and why", | |
| "tools_needed": [ | |
| {{"tool": "TOOL_NAME", "input": "specific input for the tool", "why": "reason for using this tool"}} | |
| ] | |
| }} | |
| If no tools are needed, return empty tools_needed array. | |
| Only request tools that will genuinely help answer THIS specific question. | |
| JSON:""" | |
| reasoning_response = llm.generate([{"role": "user", "content": tool_reasoning_prompt}], temperature=0.3) | |
| # Parse reasoning | |
| tool_calls = [] | |
| citations = [] | |
| tool_data_context = "" | |
| reasoning = "" | |
| try: | |
| # Extract JSON from response | |
| json_match = re.search(r'\{[\s\S]*\}', reasoning_response) | |
| if json_match: | |
| reasoning_json = json.loads(json_match.group()) | |
| reasoning = reasoning_json.get("reasoning", "") | |
| tools_needed = reasoning_json.get("tools_needed", []) | |
| progress.log(f"💭 {reasoning[:100]}...", indent=1) | |
| # Step 2: Execute requested tools | |
| for tool_req in tools_needed: | |
| tool_name = tool_req.get("tool", "").upper() | |
| tool_input = tool_req.get("input", "") | |
| progress.log(f"🔧 Using {tool_name}: {tool_input}", indent=1) | |
| if tool_name == "YAHOO_FINANCE": | |
| # First find ticker, then get data | |
| ticker = Tools.yahoo_finance_search_ticker(tool_input) | |
| if ticker: | |
| data = Tools.yahoo_finance_get_data(ticker) | |
| if data.get("success"): | |
| tool_calls.append(ToolCall( | |
| tool="Yahoo Finance", | |
| input=f"{tool_input} → {ticker}", | |
| output=f"{data['company']}: ${data['price']} | MCap: {data['market_cap']}", | |
| success=True | |
| )) | |
| citations.append({ | |
| "title": f"{data['company']} ({ticker}) - Yahoo Finance", | |
| "url": data['url'], | |
| "type": "Financial Data", | |
| "source": "Yahoo Finance" | |
| }) | |
| tool_data_context += f"\n\n### YAHOO FINANCE: {data['company']} ({ticker})\n" | |
| tool_data_context += f"Price: ${data['price']} | Market Cap: {data['market_cap']} | EV: {data['enterprise_value']}\n" | |
| tool_data_context += f"P/E: {data['pe_ratio']} | Forward P/E: {data['forward_pe']} | PEG: {data['peg_ratio']}\n" | |
| tool_data_context += f"EV/Revenue: {data['ev_to_revenue']} | EV/EBITDA: {data['ev_to_ebitda']}\n" | |
| tool_data_context += f"Revenue: {data['revenue']} | Growth: {data['revenue_growth']}\n" | |
| tool_data_context += f"Gross Margin: {data['gross_margin']} | Op Margin: {data['operating_margin']} | Profit Margin: {data['profit_margin']}\n" | |
| tool_data_context += f"Cash: {data['total_cash']} | Debt: {data['total_debt']} | D/E: {data['debt_to_equity']}\n" | |
| tool_data_context += f"FCF: {data['free_cash_flow']} | Employees: {data['employees']}\n" | |
| tool_data_context += f"52W: ${data['52_week_low']} - ${data['52_week_high']}\n" | |
| tool_data_context += f"Industry: {data['industry']} | Sector: {data['sector']}\n" | |
| else: | |
| tool_calls.append(ToolCall(tool="Yahoo Finance", input=tool_input, output="Failed to get data", success=False)) | |
| else: | |
| tool_calls.append(ToolCall(tool="Yahoo Finance", input=tool_input, output="Could not find ticker", success=False)) | |
| elif tool_name == "WEB_SEARCH": | |
| data = Tools.web_search(tool_input) | |
| if data.get("success") and data.get("results"): | |
| tool_calls.append(ToolCall( | |
| tool="Web Search", | |
| input=tool_input, | |
| output=f"{len(data['results'])} results", | |
| success=True | |
| )) | |
| tool_data_context += f"\n\n### WEB SEARCH: {tool_input}\n" | |
| for i, r in enumerate(data["results"][:4], 1): | |
| tool_data_context += f"{i}. {r['title']}\n {r['snippet']}\n" | |
| if r.get("url"): | |
| citations.append({"title": r["title"][:60], "url": r["url"], "type": "Web", "source": "Web"}) | |
| else: | |
| tool_calls.append(ToolCall(tool="Web Search", input=tool_input, output="No results", success=False)) | |
| elif tool_name == "NEWS_SEARCH": | |
| data = Tools.news_search(tool_input) | |
| if data.get("success") and data.get("articles"): | |
| tool_calls.append(ToolCall( | |
| tool="News Search", | |
| input=tool_input, | |
| output=f"{len(data['articles'])} articles", | |
| success=True | |
| )) | |
| tool_data_context += f"\n\n### NEWS: {tool_input}\n" | |
| for i, a in enumerate(data["articles"][:4], 1): | |
| tool_data_context += f"{i}. [{a['date']}] {a['title']} - {a['source']}\n" | |
| if a.get("url"): | |
| citations.append({"title": a["title"][:60], "url": a["url"], "type": "News", "source": a["source"]}) | |
| else: | |
| tool_calls.append(ToolCall(tool="News Search", input=tool_input, output="No results", success=False)) | |
| except json.JSONDecodeError: | |
| reasoning = reasoning_response[:200] | |
| progress.log(f"⚠️ Could not parse tool reasoning, proceeding without tools", indent=1) | |
| # Step 3: Generate analysis | |
| progress.log(f"📝 Generating analysis...", indent=1) | |
| data_section = f"\n\n=== REAL-TIME DATA GATHERED ===\n{tool_data_context}\n=== END DATA ===" if tool_data_context else "\n\n(No external data gathered - answering from expertise)" | |
| analysis_prompt = f"""You are the {profile['title']}. | |
| {profile['description']} | |
| CEO's QUESTION: {question} | |
| YOUR TASK: {task} | |
| {data_section} | |
| Provide your expert analysis. If you have real-time data above, reference specific numbers. | |
| If no data was gathered, provide your best analysis from your expertise. | |
| Structure: | |
| ## Key Finding | |
| (Most important insight) | |
| ## Analysis | |
| (Your expert assessment) | |
| ## Recommendations | |
| (3-4 specific actions) | |
| ## Risks & Concerns | |
| (What could go wrong) | |
| ## Metrics to Track | |
| (How to measure success)""" | |
| analysis = llm.generate([{"role": "user", "content": analysis_prompt}]) | |
| # Calculate confidence | |
| successful_tools = sum(1 for t in tool_calls if t.success) | |
| total_tools = len(tool_calls) | |
| if total_tools > 0: | |
| confidence = 0.5 + (successful_tools / total_tools) * 0.4 | |
| else: | |
| confidence = 0.7 # No tools needed, answering from expertise | |
| execution_time = time.time() - start_time | |
| progress.log(f"✅ Complete ({execution_time:.1f}s, {confidence:.0%} confidence)") | |
| return AgentOutput( | |
| agent_id=agent_id, | |
| title=profile['title'], | |
| icon=profile['icon'], | |
| reasoning=reasoning, | |
| tool_calls=tool_calls, | |
| analysis=analysis, | |
| citations=citations, | |
| confidence=confidence, | |
| execution_time=execution_time | |
| ) | |
| # ============================================================================ | |
| # ORCHESTRATOR - CEO reasons about team selection | |
| # ============================================================================ | |
| class Orchestrator: | |
| def select_team(self, question: str) -> List[Dict]: | |
| """CEO agent reasons about which experts are needed""" | |
| progress.log("🤔 CEO analyzing question and selecting team...") | |
| prompt = f"""You are the CEO. You need to assemble the right team to advise on this strategic question. | |
| QUESTION: {question} | |
| {get_all_profiles_description()} | |
| THINK THROUGH: | |
| 1. What is this question really asking? | |
| 2. What types of expertise are ESSENTIAL to answer it well? | |
| 3. Who would provide unique, non-overlapping perspectives? | |
| IMPORTANT: | |
| - Select 3-5 experts maximum (quality over quantity) | |
| - Only select experts whose expertise is DIRECTLY relevant | |
| - Don't select everyone - be selective | |
| - Each expert should bring a DIFFERENT perspective | |
| Respond in JSON format: | |
| {{ | |
| "question_analysis": "Your analysis of what the question is really asking", | |
| "team": [ | |
| {{"id": "agent_id", "task": "Specific task for this expert", "why": "Why this expert is essential"}} | |
| ] | |
| }} | |
| JSON:""" | |
| response = llm.generate([{"role": "user", "content": prompt}], temperature=0.3) | |
| try: | |
| json_match = re.search(r'\{[\s\S]*\}', response) | |
| if json_match: | |
| data = json.loads(json_match.group()) | |
| question_analysis = data.get("question_analysis", "") | |
| progress.log(f"📋 Analysis: {question_analysis[:150]}...") | |
| team = [] | |
| for member in data.get("team", []): | |
| if member.get("id") in AGENT_PROFILES: | |
| team.append(member) | |
| profile = AGENT_PROFILES[member["id"]] | |
| progress.log(f" ✓ {profile['icon']} {profile['title']}: {member.get('why', '')[:50]}...") | |
| if len(team) >= 2: | |
| return team | |
| except json.JSONDecodeError: | |
| pass | |
| # Fallback | |
| progress.log("⚠️ Using default team") | |
| return [ | |
| {"id": "cfo", "task": "Provide financial analysis"}, | |
| {"id": "cso", "task": "Provide strategic analysis"}, | |
| {"id": "cro_risk", "task": "Identify key risks"} | |
| ] | |
| def run(self, question: str, progress_callback: Callable = None) -> tuple: | |
| """Run full analysis""" | |
| progress.clear() | |
| # Phase 1: Team Selection | |
| progress.phase("PHASE 1: CEO ANALYZING & SELECTING TEAM") | |
| if progress_callback: | |
| progress_callback(progress.get_log(), 0.05, "CEO analyzing question...") | |
| team = self.select_team(question) | |
| if progress_callback: | |
| progress_callback(progress.get_log(), 0.10, "Team selected") | |
| # Phase 2: Agents Execute | |
| progress.phase("PHASE 2: EXPERTS GATHERING INTELLIGENCE") | |
| outputs = [] | |
| total = len(team) | |
| for i, member in enumerate(team): | |
| if progress_callback: | |
| profile = AGENT_PROFILES[member["id"]] | |
| progress_callback(progress.get_log(), 0.10 + (0.55 * i / total), f"{profile['icon']} {profile['title']} analyzing...") | |
| output = run_agent(member["id"], question, member.get("task", "")) | |
| outputs.append(output) | |
| if progress_callback: | |
| progress_callback(progress.get_log(), 0.10 + (0.55 * (i + 1) / total), f"{output.icon} {output.title} complete") | |
| # Phase 3: Debate | |
| progress.phase("PHASE 3: EXPERT DEBATE") | |
| if progress_callback: | |
| progress_callback(progress.get_log(), 0.70, "Running expert debate...") | |
| debate = self._run_debate(question, outputs) | |
| # Phase 4: CEO Brief | |
| progress.phase("PHASE 4: CEO DECISION BRIEF") | |
| if progress_callback: | |
| progress_callback(progress.get_log(), 0.85, "Generating decision brief...") | |
| brief = self._generate_brief(question, outputs, debate) | |
| progress.phase("✅ ANALYSIS COMPLETE") | |
| if progress_callback: | |
| progress_callback(progress.get_log(), 1.0, "Complete!") | |
| return outputs, debate, brief, progress.get_log() | |
| def _run_debate(self, question: str, outputs: List[AgentOutput]) -> str: | |
| progress.log("⚖️ Synthesizing expert debate...") | |
| positions = "\n\n".join([ | |
| f"**{o.icon} {o.title}** (Confidence: {o.confidence:.0%}):\n{o.analysis[:700]}..." | |
| for o in outputs | |
| ]) | |
| prompt = f"""You are moderating a strategic debate among executives. | |
| QUESTION: {question} | |
| EXPERT POSITIONS: | |
| {positions} | |
| Synthesize the debate: | |
| ## Areas of Agreement | |
| (Where do experts align?) | |
| ## Key Tensions | |
| (Where do they disagree and why?) | |
| ## Open Questions | |
| (What needs more research?) | |
| ## Synthesis | |
| (How to reconcile different views into a coherent recommendation?)""" | |
| return llm.generate([{"role": "user", "content": prompt}]) | |
| def _generate_brief(self, question: str, outputs: List[AgentOutput], debate: str) -> str: | |
| progress.log("📋 Generating CEO decision brief...") | |
| # Collect citations | |
| all_citations = [] | |
| for o in outputs: | |
| all_citations.extend(o.citations) | |
| citations_text = "\n".join([f"- [{c['title']}]({c['url']}) ({c['source']})" for c in all_citations[:12]]) | |
| analysis = "\n\n".join([f"### {o.icon} {o.title}\n{o.analysis}" for o in outputs]) | |
| prompt = f"""Create a board-ready EXECUTIVE DECISION BRIEF. | |
| QUESTION: {question} | |
| EXPERT ANALYSIS: | |
| {analysis} | |
| DEBATE: | |
| {debate} | |
| SOURCES: | |
| {citations_text} | |
| --- | |
| # EXECUTIVE DECISION BRIEF | |
| ## Executive Summary | |
| (4-5 sentences: situation, key findings, recommendation) | |
| ## DECISION: [GO / NO-GO / CONDITIONAL] | |
| (Clear recommendation with rationale) | |
| ## Key Data Points | |
| (8-10 bullets with SPECIFIC numbers if available) | |
| ## 90-Day Action Plan | |
| | Week | Action | Owner | Success Metric | | |
| |------|--------|-------|----------------| | |
| (8+ rows) | |
| ## Risk Matrix | |
| | Risk | Likelihood | Impact | Mitigation | | |
| |------|------------|--------|------------| | |
| (Top 5 risks) | |
| ## Investment Required | |
| - Budget: $X | |
| - Team: X people | |
| - Timeline: X months | |
| ## Board Talking Points | |
| (5 key messages)""" | |
| return llm.generate([{"role": "user", "content": prompt}]) | |
| # ============================================================================ | |
| # VISUALIZATIONS | |
| # ============================================================================ | |
| def build_tool_chart(outputs: List[AgentOutput]) -> go.Figure: | |
| data = [] | |
| for o in outputs: | |
| for tc in o.tool_calls: | |
| data.append({ | |
| "Agent": f"{o.icon} {o.title.split('(')[0].strip()}", | |
| "Tool": tc.tool, | |
| "Status": "✅" if tc.success else "❌" | |
| }) | |
| if not data: | |
| fig = go.Figure() | |
| fig.add_annotation(text="No tools used (answered from expertise)", x=0.5, y=0.5, showarrow=False, font=dict(size=14)) | |
| fig.update_layout(height=300) | |
| return fig | |
| df = pd.DataFrame(data) | |
| fig = px.scatter(df, x="Tool", y="Agent", color="Status", title="🔧 Tool Usage", | |
| color_discrete_map={"✅": "#22c55e", "❌": "#ef4444"}) | |
| fig.update_traces(marker=dict(size=20)) | |
| fig.update_layout(height=350) | |
| return fig | |
| def build_confidence_chart(outputs: List[AgentOutput]) -> go.Figure: | |
| df = pd.DataFrame([{ | |
| "Agent": f"{o.icon} {o.title.split('(')[0].strip()[:15]}", | |
| "Confidence": o.confidence * 100 | |
| } for o in outputs]) | |
| fig = px.bar(df, x="Agent", y="Confidence", color="Confidence", | |
| color_continuous_scale="RdYlGn", title="📊 Confidence", | |
| text=df["Confidence"].apply(lambda x: f"{x:.0f}%")) | |
| fig.update_traces(textposition="outside") | |
| fig.update_layout(yaxis_range=[0, 110], coloraxis_showscale=False, height=350) | |
| return fig | |
| def format_citations(outputs: List[AgentOutput]) -> str: | |
| by_type = {} | |
| seen = set() | |
| for o in outputs: | |
| for c in o.citations: | |
| url = c.get("url", "") | |
| if url and url not in seen: | |
| seen.add(url) | |
| t = c.get("type", "Other") | |
| if t not in by_type: | |
| by_type[t] = [] | |
| by_type[t].append(c) | |
| if not by_type: | |
| return "No external sources consulted (answered from expertise)." | |
| lines = ["# 📚 Sources\n"] | |
| for source_type, citations in by_type.items(): | |
| lines.append(f"\n## {source_type}") | |
| for i, c in enumerate(citations[:8], 1): | |
| lines.append(f"{i}. [{c['title']}]({c['url']}) - {c.get('source', '')}") | |
| return "\n".join(lines) | |
| # ============================================================================ | |
| # GRADIO APP | |
| # ============================================================================ | |
| def create_app(): | |
| orchestrator = Orchestrator() | |
| def run_analysis(query: str, prog=gr.Progress()): | |
| if not query.strip(): | |
| return "", None, None, "", "", "", "" | |
| def update(log, pct, desc=""): | |
| prog(pct, desc=desc) | |
| prog(0.02, desc="Starting...") | |
| outputs, debate, brief, final_log = orchestrator.run(query, update) | |
| prog(0.92, desc="Building visualizations...") | |
| tool_chart = build_tool_chart(outputs) | |
| conf_chart = build_confidence_chart(outputs) | |
| insights = "\n\n---\n\n".join([ | |
| f"# {o.icon} {o.title}\n\n" | |
| f"**Reasoning:** {o.reasoning[:200]}...\n\n" | |
| f"**Confidence:** {o.confidence:.0%} | **Time:** {o.execution_time:.1f}s\n\n" | |
| f"**Tools Used:** {', '.join([f'{t.tool} ({t.input})' for t in o.tool_calls]) or 'None (answered from expertise)'}\n\n" | |
| f"{o.analysis}" | |
| for o in outputs | |
| ]) | |
| citations = format_citations(outputs) | |
| prog(1.0, desc="Complete!") | |
| return final_log, tool_chart, conf_chart, insights, debate, brief, citations | |
| with gr.Blocks(title="CEO Crew v5") as app: | |
| gr.Markdown(""" | |
| # 🏢 CEO Crew v5 - LLM-Driven Reasoning | |
| **No hardcoding.** CEO reasons about team selection. Each agent reasons about tools needed. | |
| """) | |
| with gr.Row(): | |
| query = gr.Textbox(label="Strategic Question", lines=3, scale=4, | |
| placeholder="e.g., Should we acquire Subex for our telecom fraud detection capabilities?") | |
| run_btn = gr.Button("🚀 Analyze", variant="primary", scale=1) | |
| log_output = gr.Textbox(label="📋 Live Progress", lines=18, interactive=False) | |
| with gr.Tabs(): | |
| with gr.TabItem("🔧 Tools & Confidence"): | |
| with gr.Row(): | |
| tool_chart = gr.Plot() | |
| conf_chart = gr.Plot() | |
| with gr.TabItem("💡 Expert Analysis"): | |
| insights_out = gr.Markdown() | |
| with gr.TabItem("⚖️ Debate"): | |
| debate_out = gr.Markdown() | |
| with gr.TabItem("🎯 Decision Brief"): | |
| brief_out = gr.Markdown() | |
| with gr.TabItem("📚 Sources"): | |
| citations_out = gr.Markdown() | |
| gr.Examples([ | |
| ["Should we acquire Zoom for our enterprise communications strategy? They're trading around $70/share."], | |
| ["A major competitor was just acquired by Microsoft. How should we respond strategically?"], | |
| ["Should we expand our fraud detection business into the healthcare revenue cycle management market?"], | |
| ["Our top 3 engineers were poached by a competitor. What should we do?"], | |
| ], inputs=query) | |
| run_btn.click(run_analysis, [query], | |
| [log_output, tool_chart, conf_chart, insights_out, debate_out, brief_out, citations_out]) | |
| return app | |
| if __name__ == "__main__": | |
| app = create_app() | |
| app.launch() |