Spaces:
Sleeping
Sleeping
File size: 6,862 Bytes
65ade69 | 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 | import os
import json
from urllib.parse import urlparse
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from src.state import AgentState
from src.tools import web_search, hn_search, github_search
from langgraph.prebuilt import create_react_agent
# Initialize the OpenAI model we will use for our agents.
def get_llm():
return ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
# ──────────────────────────────────────────────
# Source Extraction Utility
# ──────────────────────────────────────────────
def _extract_sources(messages):
"""
Parse the ReAct agent's tool-response messages and pull out
every (title, url) pair that was returned by our tools.
"""
sources = []
seen_urls = set()
for msg in messages:
# Only look at ToolMessage objects
if not (hasattr(msg, "type") and msg.type == "tool"):
continue
content = getattr(msg, "content", "")
if not content:
continue
lines = content.split("\n")
current_title = None
for line in lines:
stripped = line.strip()
# Capture title / repo name from the line before the URL
if stripped.startswith("Title:"):
current_title = stripped[len("Title:"):].strip()
elif stripped.startswith("Repo:"):
current_title = stripped[len("Repo:"):].strip()
# Capture URL
elif stripped.startswith("URL:"):
url = stripped[len("URL:"):].strip()
if url and url.startswith("http") and url not in seen_urls:
seen_urls.add(url)
domain = urlparse(url).netloc.replace("www.", "")
sources.append({
"title": current_title or domain,
"url": url,
"domain": domain,
})
current_title = None
return sources
# ──────────────────────────────────────────────
# Agent 1: Competitor Analysis
# ──────────────────────────────────────────────
def competitor_agent(state: AgentState) -> dict:
idea = state["idea"]
print(f"[Agent: Competitor] Starting research on idea: {idea[:30]}...")
llm = get_llm()
agent = create_react_agent(llm, tools=[web_search])
prompt = f"""You are a top-tier startup analyst. Your job is to find direct and indirect competitors for this startup idea: '{idea}'
Use the web_search tool to find existing companies doing this.
Provide a concise summary (max 3-4 sentences) of who the biggest players are and how crowded the space is."""
response = agent.invoke({"messages": [HumanMessage(content=prompt)]})
final_message = response["messages"][-1].content
sources = _extract_sources(response["messages"])
return {"results": [{"agent": "competitor_analysis", "findings": final_message, "sources": sources}]}
# ──────────────────────────────────────────────
# Agent 2: Market Sizing
# ──────────────────────────────────────────────
def market_sizing_agent(state: AgentState) -> dict:
idea = state["idea"]
print(f"[Agent: Market Sizing] Starting research on idea: {idea[:30]}...")
llm = get_llm()
agent = create_react_agent(llm, tools=[web_search])
prompt = f"""You are a market researcher. Estimate the Total Addressable Market (TAM) for this startup idea: '{idea}'
Use the web_search tool to find market reports, industry growth rates (CAGR), and overall market value.
Provide a concise summary (max 3-4 sentences) with rough dollar estimates if possible."""
response = agent.invoke({"messages": [HumanMessage(content=prompt)]})
final_message = response["messages"][-1].content
sources = _extract_sources(response["messages"])
return {"results": [{"agent": "market_sizing", "findings": final_message, "sources": sources}]}
# ──────────────────────────────────────────────
# Agent 3: Community Sentiment
# ──────────────────────────────────────────────
def sentiment_agent(state: AgentState) -> dict:
idea = state["idea"]
print(f"[Agent: Sentiment] Starting research on idea: {idea[:30]}...")
llm = get_llm()
agent = create_react_agent(llm, tools=[hn_search])
prompt = f"""You are a community manager and developer advocate. Analyze the sentiment around this startup idea: '{idea}'
Use the hn_search tool to find past discussions, Show HN posts, or similar concepts on HackerNews.
Provide a concise summary (max 3-4 sentences) of whether the tech community is generally positive, skeptical, or ignores this problem."""
response = agent.invoke({"messages": [HumanMessage(content=prompt)]})
final_message = response["messages"][-1].content
sources = _extract_sources(response["messages"])
return {"results": [{"agent": "community_sentiment", "findings": final_message, "sources": sources}]}
# ──────────────────────────────────────────────
# Agent 4: Open-Source Risk Detection
# ──────────────────────────────────────────────
def os_risk_agent(state: AgentState) -> dict:
idea = state["idea"]
print(f"[Agent: OS Risk] Starting research on idea: {idea[:30]}...")
llm = get_llm()
agent = create_react_agent(llm, tools=[github_search])
prompt = f"""You are a technical due diligence expert. Find open-source alternatives to this startup idea: '{idea}'
Use the github_search tool to find repositories that solve the same problem.
Provide a concise summary (max 3-4 sentences) naming the top repos (if any) and whether they are a major threat to a paid SaaS product."""
response = agent.invoke({"messages": [HumanMessage(content=prompt)]})
final_message = response["messages"][-1].content
sources = _extract_sources(response["messages"])
return {"results": [{"agent": "os_risk", "findings": final_message, "sources": sources}]}
|