Staying ahead of competitors requires constant vigilance—tracking product launches, funding rounds, partnerships, and strategic moves across the web. The open-source Competitive Intelligence Monitor project demonstrates how to automate this process using CocoIndex, Tavily Search, and LLM extraction to continuously track and structure competitor news into a queryable PostgreSQL database.
<h2>What the Project Does</h2>
The system automates web monitoring by using Tavily’s AI-native search to pull full-text articles, then feeding them through a GPT-4o-mini–based extraction layer to detect structured “competitive events” such as:
<ul>
<li>Product launches and feature releases</li>
<li>Partnerships and collaborations</li>
<li>Funding rounds and financial news</li>
<li>Key executive hires/departures</li>
<li>Acquisitions and mergers</li>
</ul>
These events and their source articles are stored in PostgreSQL so teams can ask natural questions like “What has Anthropic been doing recently?” or “Which competitors are making the most news this week?”
<h2>Core Architecture</h2>
<pre><code>┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Tavily AI │────▶│ CocoIndex │────▶│ PostgreSQL │
│ Search │ │ Pipeline │ │ Database │
└──────────────┘ └──────────────┘ └──────────────┘
 │ │ │
 ▼ ▼ ▼
 Articles Extraction Intelligence
 (web data) (GPT-4o-mini) (structured)
</code></pre>
Data flows from Tavily search results into an LLM extraction step that produces <code>CompetitiveEvent</code> objects, then into dual indexes—one table for raw articles and another for normalized events.
<h2>Data Model: The CompetitiveEvent Class</h2>
At the heart of the extraction is the <code>CompetitiveEvent</code> dataclass that defines what the LLM should extract from each article:
<pre><code class="language-python">@dataclasses.dataclass
class CompetitiveEvent:
 &quot;&quot;&quot;A competitive intelligence event extracted from text.
 
 Examples:
 - Product Launch: &quot;OpenAI released GPT-5 with multimodal capabilities&quot;
 - Partnership: &quot;Anthropic partnered with Google Cloud for enterprise AI&quot;
 - Funding: &quot;Mistral AI raised $400M Series B led by Andreessen Horowitz&quot;
 - Key Hire: &quot;Former Meta AI director joined Cohere as Chief Scientist&quot;
 - Strategic Move: &quot;Microsoft acquired AI startup Inflection for $650M&quot;
 &quot;&quot;&quot;
 event_type: str # &quot;product_launch&quot;, &quot;partnership&quot;, &quot;funding&quot;, &quot;key_hire&quot;, &quot;acquisition&quot;, &quot;other&quot;
 competitor: str # Company name (e.g., &quot;OpenAI&quot;, &quot;Anthropic&quot;, &quot;Google AI&quot;)
 description: str # Brief description of the event
 significance: str # &quot;high&quot;, &quot;medium&quot;, &quot;low&quot; - based on market impact
 related_companies: list[str] # Other companies mentioned
</code></pre>
<h2>Custom Tavily Source Connector</h2>
The project implements a custom CocoIndex source connector that interfaces with Tavily’s AI-native search API:
<pre><code class="language-python">class TavilySearchSource(SourceSpec):
 &quot;&quot;&quot;Fetches competitive intelligence using Tavily AI Search API.&quot;&quot;&quot;
 competitor: str
 days_back: int = 7
 max_results: int = 10

@source_connector(
 spec_cls=TavilySearchSource,
 key_type=_ArticleKey,
 value_type=_Article,
)
class TavilySearchConnector:
 async def list(self) -&gt; AsyncIterator[PartialSourceRow[_ArticleKey, _Article]]:
 &quot;&quot;&quot;List articles from Tavily search.&quot;&quot;&quot;
 search_query = (
 f&quot;{self._spec.competitor} AND &quot;
 f&quot;(funding OR partnership OR product launch OR acquisition OR executive hire)&quot;
 )
 
 client = TavilyClient(api_key=self._api_key)
 response = client.search(
 query=search_query,
 search_depth=&quot;advanced&quot;,
 max_results=self._spec.max_results,
 include_raw_content=True,
 )
 
 for result in response.get(&quot;results&quot;, []):
 url = result[&quot;url&quot;]
 yield PartialSourceRow(
 key=_ArticleKey(url=url),
 data=PartialSourceRowData(ordinal=ordinal),
 )
</code></pre>
<h2>The CocoIndex Pipeline Definition</h2>
The main pipeline uses CocoIndex’s flow builder to orchestrate data collection and LLM extraction:
<pre><code class="language-python">@cocoindex.flow_def(name=&quot;CompetitiveIntelligence&quot;)
def competitive_intelligence_flow(
 flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope
) -&gt; None:
 &quot;&quot;&quot;Main pipeline for competitive intelligence monitoring.&quot;&quot;&quot;
 
 competitors = os.getenv(&quot;COMPETITORS&quot;, &quot;OpenAI,Anthropic&quot;).split(&quot;,&quot;)
 refresh_interval = int(os.getenv(&quot;REFRESH_INTERVAL_SECONDS&quot;, &quot;3600&quot;))
 
 # Add Tavily search source for each competitor
 for competitor in competitors:
 data_scope[f&quot;articles_{competitor.strip()}&quot;] = flow_builder.add_source(
 TavilySearchSource(
 competitor=competitor.strip(),
 days_back=search_days_back,
 max_results=10,
 ),
 refresh_interval=timedelta(seconds=refresh_interval),
 )
 
 articles_index = data_scope.add_collector()
 events_index = data_scope.add_collector()
 
 # Process each competitor&#x27;s articles
 for competitor in competitors:
 articles = data_scope[f&quot;articles_{competitor.strip()}&quot;]
 
 with articles.row() as article:
 # Extract competitive events using GPT-4o-mini via OpenRouter
 article[&quot;events&quot;] = article[&quot;content&quot;].transform(
 cocoindex.functions.ExtractByLlm(
 llm_spec=cocoindex.LlmSpec(
 api_type=cocoindex.LlmApiType.OPENAI,
 model=&quot;openai/gpt-4o-mini&quot;,
 address=&quot;https://openrouter.ai/api/v1&quot;,
 ),
 output_type=list[CompetitiveEvent],
 instruction=(
 &quot;Extract competitive intelligence events from this article. &quot;
 &quot;Focus on: product launches, partnerships, funding rounds, key hires, &quot;
 &quot;acquisitions, and other strategic moves.&quot;
 ),
 )
 )
</code></pre>
<h2>Query Handlers for Analysis</h2>
The project includes built-in query handlers that enable SQL-powered intelligence retrieval:
<pre><code class="language-python">@competitive_intelligence_flow.query_handler()
def search_by_competitor(
 competitor: str, 
 event_type: str | None = None, 
 limit: int = 20
) -&gt; cocoindex.QueryOutput:
 &quot;&quot;&quot;Find recent competitive intelligence about a specific competitor.&quot;&quot;&quot;
 
 with connection_pool().connection() as conn:
 with conn.cursor() as cur:
 sql = f&quot;&quot;&quot;
 SELECT e.competitor, e.event_type, e.description, e.significance,
 e.related_companies, a.title, a.url, a.source, a.published_at
 FROM {events_table} e
 JOIN {articles_table} a ON e.article_id = a.id
 WHERE LOWER(e.competitor) LIKE LOWER(%s)
 &quot;&quot;&quot;
 params = [f&quot;%{competitor}%&quot;]
 
 if event_type:
 sql += &quot; AND e.event_type = %s&quot;
 params.append(event_type)
 
 sql += &quot; ORDER BY a.published_at DESC LIMIT %s&quot;
 cur.execute(sql, params)
 
 return cocoindex.QueryOutput(results=[...])
</code></pre>
<h2>Getting Started</h2>
Configuration is controlled through environment variables:
<pre><code class="language-bash">DATABASE_URL=postgresql://user:password@localhost:5432/competitive_intel
COCOINDEX_DATABASE_URL=postgresql://user:password@localhost:5432/competitive_intel
OPENAI_API_KEY=sk-or-v1-...
TAVILY_API_KEY=tvly-...
COMPETITORS=OpenAI,Anthropic,Google AI,Meta AI,Mistral AI
REFRESH_INTERVAL_SECONDS=3600
SEARCH_DAYS_BACK=7
</code></pre>
Run the interactive CLI for first-time setup:
<pre><code class="language-bash">python3 run_interactive.py
</code></pre>
Or use CocoIndex directly for automated deployments:
<pre><code class="language-bash">cocoindex update main -f # Initial sync
cocoindex update -L main.py # Continuous monitoring
</code></pre>
<h2>Why This Approach Matters</h2>
By combining AI-native search with structured LLM extraction, the monitor:
<ul>
<li>Avoids brittle scraping - Tavily handles content extraction</li>
<li>De-duplicates work - CocoIndex tracks processed articles via incremental processing</li>
<li>Turns noise into signal - Structured events with significance scoring</li>
<li>Enables flexible analysis - Dual indexing (raw + extracted) for maximum flexibility</li>
</ul>
The project supports multiple query types:
<ul>
<li>Search by competitor name</li>
<li>Filter by event type (funding, partnerships, acquisitions, etc.)</li>
<li>Rank by significance (high=3, medium=2, low=1 weighted scoring)</li>
<li>Trend analysis across time periods</li>
</ul>
<h2>Get the Code</h2>
The project is MIT-licensed and available on GitHub:
<a href="https://github.com/Laksh-star/competitive-intelligence" target="_blank" rel="noopener nofollow">https://github.com/Laksh-star/competitive-intelligence</a>
<hr>
Built with:
<ul>
<li><a href="https://cocoindex.io/" target="_blank" rel="noopener nofollow">CocoIndex</a> - Modern data pipeline framework</li>
<li><a href="https://tavily.com/" target="_blank" rel="noopener nofollow">Tavily AI Search</a> - AI-native search engine</li>
<li><a href="https://openrouter.ai/" target="_blank" rel="noopener nofollow">OpenRouter</a> - Multi-model API gateway</li>
</ul>
Have questions or want to contribute? Drop a comment below or open an issue on GitHub!

An open-source competitive intelligence monitoring system automates tracking of competitor activities by combining Tavily AI Search for web data collection, GPT-4o-mini for structured event extraction, and CocoIndex for pipeline orchestration. The system continuously monitors product launches, funding rounds, partnerships, and executive moves, storing structured events in PostgreSQL for natural language querying. It uses a custom Tavily connector, defines competitive events as dataclasses, and provides query handlers for filtering by competitor, event type, and significance. The architecture avoids brittle web scraping, handles deduplication through incremental processing, and enables flexible analysis through dual indexing of raw articles and extracted events.

Build an AI-Powered Competitive Intelligence Monitor