diff --git a/.env.example b/.env.example index 778272bf3..cc49fccd5 100644 --- a/.env.example +++ b/.env.example @@ -2,6 +2,7 @@ OPENAI_API_KEY= ANTHROPIC_API_KEY= GOOGLE_API_KEY= TAVILY_API_KEY= +EXA_API_KEY= LANGSMITH_API_KEY= LANGSMITH_PROJECT= LANGSMITH_TRACING= diff --git a/pyproject.toml b/pyproject.toml index 54b13248d..01208ed24 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,7 +24,7 @@ dependencies = [ "xmltodict>=0.14.2", "linkup-sdk>=0.2.3", "duckduckgo-search>=3.0.0", - "exa-py>=1.8.8", + "exa-py>=2.0.0", "requests>=2.32.3", "beautifulsoup4==4.14.3", "python-dotenv>=1.0.1", diff --git a/src/open_deep_research/configuration.py b/src/open_deep_research/configuration.py index 1c5bac9e9..a7e0672aa 100644 --- a/src/open_deep_research/configuration.py +++ b/src/open_deep_research/configuration.py @@ -10,10 +10,11 @@ class SearchAPI(Enum): """Enumeration of available search API providers.""" - + ANTHROPIC = "anthropic" OPENAI = "openai" TAVILY = "tavily" + EXA = "exa" NONE = "none" class MCPConfig(BaseModel): @@ -84,6 +85,7 @@ class Configuration(BaseModel): "description": "Search API to use for research. NOTE: Make sure your Researcher Model supports the selected search API.", "options": [ {"label": "Tavily", "value": SearchAPI.TAVILY.value}, + {"label": "Exa", "value": SearchAPI.EXA.value}, {"label": "OpenAI Native Web Search", "value": SearchAPI.OPENAI.value}, {"label": "Anthropic Native Web Search", "value": SearchAPI.ANTHROPIC.value}, {"label": "None", "value": SearchAPI.NONE.value} diff --git a/src/open_deep_research/utils.py b/src/open_deep_research/utils.py index 82ce304e2..a58ce4e32 100644 --- a/src/open_deep_research/utils.py +++ b/src/open_deep_research/utils.py @@ -29,6 +29,11 @@ from mcp import McpError from tavily import AsyncTavilyClient +try: + from exa_py import Exa +except ImportError: + Exa = None + from open_deep_research.configuration import Configuration, SearchAPI from open_deep_research.prompts import summarize_webpage_prompt from open_deep_research.state import ResearchComplete, Summary @@ -212,6 +217,286 @@ async def summarize_webpage(model: BaseChatModel, webpage_content: str) -> str: logging.warning(f"Summarization failed with error: {str(e)}, returning original content") return webpage_content +########################## +# Exa Search Tool Utils +########################## +EXA_SEARCH_DESCRIPTION = ( + "An AI-powered search engine that returns high-quality, semantically relevant " + "web results with rich content extraction (full text, AI summaries, and " + "highlights). Useful for research-oriented queries that benefit from " + "neural search and on-demand content retrieval." +) + + +@tool(description=EXA_SEARCH_DESCRIPTION) +async def exa_search( + queries: List[str], + max_results: Annotated[int, InjectedToolArg] = 5, + search_type: Annotated[Literal["auto", "neural", "fast"], InjectedToolArg] = "auto", + category: Annotated[ + Literal[ + "company", + "research paper", + "news", + "personal site", + "financial report", + "people", + ] | None, + InjectedToolArg, + ] = None, + include_domains: Annotated[List[str] | None, InjectedToolArg] = None, + exclude_domains: Annotated[List[str] | None, InjectedToolArg] = None, + start_published_date: Annotated[str | None, InjectedToolArg] = None, + end_published_date: Annotated[str | None, InjectedToolArg] = None, + config: RunnableConfig = None, +) -> str: + """Fetch and summarize search results from the Exa search API. + + Args: + queries: List of search queries to execute + max_results: Maximum number of results to return per query + search_type: Exa search type ("auto", "neural", or "fast") + category: Optional category filter (e.g. "research paper", "news") + include_domains: Optional list of domains to restrict results to + exclude_domains: Optional list of domains to exclude from results + start_published_date: Optional ISO 8601 lower bound for publish date + end_published_date: Optional ISO 8601 upper bound for publish date + config: Runtime configuration for API keys and model settings + + Returns: + Formatted string containing summarized search results + """ + # Step 1: Execute search queries asynchronously + search_results = await exa_search_async( + queries, + max_results=max_results, + search_type=search_type, + category=category, + include_domains=include_domains, + exclude_domains=exclude_domains, + start_published_date=start_published_date, + end_published_date=end_published_date, + config=config, + ) + + # Step 2: Deduplicate results by URL to avoid processing the same content multiple times + unique_results: Dict[str, Dict[str, Any]] = {} + for response in search_results: + for result in response["results"]: + url = result["url"] + if url not in unique_results: + unique_results[url] = {**result, "query": response["query"]} + + # Step 3: Set up the summarization model with configuration + configurable = Configuration.from_runnable_config(config) + + # Character limit to stay within model token limits (configurable) + max_char_to_include = configurable.max_content_length + + # Initialize summarization model with retry logic + model_api_key = get_api_key_for_model(configurable.summarization_model, config) + summarization_model = init_chat_model( + model=configurable.summarization_model, + max_tokens=configurable.summarization_model_max_tokens, + api_key=model_api_key, + tags=["langsmith:nostream"], + ).with_structured_output(Summary).with_retry( + stop_after_attempt=configurable.max_structured_output_retries + ) + + # Step 4: Create summarization tasks (skip empty content) + async def noop(): + """No-op function for results without raw content.""" + return None + + summarization_tasks = [ + noop() + if not result.get("raw_content") + else summarize_webpage( + summarization_model, + result["raw_content"][:max_char_to_include], + ) + for result in unique_results.values() + ] + + # Step 5: Execute all summarization tasks in parallel + summaries = await asyncio.gather(*summarization_tasks) + + # Step 6: Combine results with their summaries + summarized_results = { + url: { + "title": result["title"], + "content": result["content"] if summary is None else summary, + } + for url, result, summary in zip( + unique_results.keys(), + unique_results.values(), + summaries, + ) + } + + # Step 7: Format the final output + if not summarized_results: + return "No valid search results found. Please try different search queries or use a different search API." + + formatted_output = "Search results: \n\n" + for i, (url, result) in enumerate(summarized_results.items()): + formatted_output += f"\n\n--- SOURCE {i + 1}: {result['title']} ---\n" + formatted_output += f"URL: {url}\n\n" + formatted_output += f"SUMMARY:\n{result['content']}\n\n" + formatted_output += "\n\n" + "-" * 80 + "\n" + + return formatted_output + + +async def exa_search_async( + search_queries, + max_results: int = 5, + search_type: Literal["auto", "neural", "fast"] = "auto", + category: str | None = None, + include_domains: List[str] | None = None, + exclude_domains: List[str] | None = None, + start_published_date: str | None = None, + end_published_date: str | None = None, + config: RunnableConfig = None, +): + """Execute multiple Exa search queries asynchronously. + + Args: + search_queries: List of search query strings to execute + max_results: Maximum number of results per query + search_type: Exa search type ("auto", "neural", or "fast") + category: Optional category filter + include_domains: Optional list of domains to restrict results to + exclude_domains: Optional list of domains to exclude + start_published_date: Optional ISO 8601 lower bound for publish date + end_published_date: Optional ISO 8601 upper bound for publish date + config: Runtime configuration for API key access + + Returns: + List of search response dictionaries with the same shape used by the + Tavily search path, so downstream summarization is identical. + """ + if Exa is None: + raise ImportError( + "The exa-py package is required for Exa search. " + "Install it with `pip install exa-py`." + ) + + api_key = get_exa_api_key(config) + if not api_key: + raise ValueError( + "EXA_API_KEY is not set. Get a key at https://dashboard.exa.ai " + "and configure it in your environment." + ) + + if include_domains and exclude_domains: + raise ValueError( + "Cannot specify both include_domains and exclude_domains for Exa search." + ) + + exa_client = Exa(api_key=api_key) + exa_client.headers["x-exa-integration"] = "open-deep-research" + + def run_query(query: str) -> Dict[str, Any]: + kwargs: Dict[str, Any] = { + "num_results": max_results, + "type": search_type, + "text": True, + "highlights": True, + "summary": True, + } + if category is not None: + kwargs["category"] = category + if include_domains: + kwargs["include_domains"] = include_domains + if exclude_domains: + kwargs["exclude_domains"] = exclude_domains + if start_published_date: + kwargs["start_published_date"] = start_published_date + if end_published_date: + kwargs["end_published_date"] = end_published_date + + response = exa_client.search_and_contents(query, **kwargs) + return _format_exa_response(query, response) + + # Run the synchronous SDK calls off the event loop, in parallel per query. + return await asyncio.gather( + *[asyncio.to_thread(run_query, q) for q in search_queries] + ) + + +def _format_exa_response(query: str, response: Any) -> Dict[str, Any]: + """Convert an Exa SearchResponse into the Tavily-shaped result dict. + + Exa returns either dicts or typed result objects depending on the SDK + version, so attribute access is normalized via ``_get_attr``. Content for + each result cascades through text -> summary -> highlights so the + downstream summarizer always has something to work with. + """ + results_list = _get_attr(response, "results", []) or [] + formatted_results: List[Dict[str, Any]] = [] + seen_urls: set = set() + + for result in results_list: + url = _get_attr(result, "url") + if not url or url in seen_urls: + continue + seen_urls.add(url) + + text = _get_attr(result, "text") or "" + summary = _get_attr(result, "summary") or "" + highlights = _get_attr(result, "highlights") or [] + if isinstance(highlights, list): + highlights_str = "\n".join(str(h) for h in highlights if h) + else: + highlights_str = str(highlights) + + # Snippet shown in the deduped "content" field. Prefer a concise + # source: summary first, then highlights, then a text excerpt. + if summary: + snippet = summary + elif highlights_str: + snippet = highlights_str + elif text: + snippet = text[:1000] + else: + snippet = "" + + # Raw content fed into the summarization step. Prefer full text; + # fall back to summary/highlights so empty-text pages still + # contribute something. + raw_content_parts = [p for p in (text, summary, highlights_str) if p] + raw_content = "\n\n".join(raw_content_parts) if raw_content_parts else None + + formatted_results.append( + { + "title": _get_attr(result, "title") or "", + "url": url, + "content": snippet, + "score": _get_attr(result, "score"), + "raw_content": raw_content, + } + ) + + return { + "query": query, + "follow_up_questions": None, + "answer": None, + "images": [], + "results": formatted_results, + } + + +def _get_attr(obj: Any, key: str, default: Any = None) -> Any: + """Read ``key`` from a dict or attribute-access object.""" + if obj is None: + return default + if isinstance(obj, dict): + return obj.get(key, default) + return getattr(obj, key, default) + + ########################## # Reflection Tool Utils ########################## @@ -553,12 +838,22 @@ async def get_search_tool(search_api: SearchAPI): # Configure Tavily search tool with metadata search_tool = tavily_search search_tool.metadata = { - **(search_tool.metadata or {}), - "type": "search", + **(search_tool.metadata or {}), + "type": "search", "name": "web_search" } return [search_tool] - + + elif search_api == SearchAPI.EXA: + # Configure Exa search tool with metadata + search_tool = exa_search + search_tool.metadata = { + **(search_tool.metadata or {}), + "type": "search", + "name": "web_search", + } + return [search_tool] + elif search_api == SearchAPI.NONE: # No search functionality configured return [] @@ -923,3 +1218,15 @@ def get_tavily_api_key(config: RunnableConfig): return api_keys.get("TAVILY_API_KEY") else: return os.getenv("TAVILY_API_KEY") + + +def get_exa_api_key(config: RunnableConfig): + """Get Exa API key from environment or config.""" + should_get_from_config = os.getenv("GET_API_KEYS_FROM_CONFIG", "false") + if should_get_from_config.lower() == "true": + api_keys = (config or {}).get("configurable", {}).get("apiKeys", {}) + if not api_keys: + return None + return api_keys.get("EXA_API_KEY") + else: + return os.getenv("EXA_API_KEY") diff --git a/tests/test_exa_search.py b/tests/test_exa_search.py new file mode 100644 index 000000000..9559c0be4 --- /dev/null +++ b/tests/test_exa_search.py @@ -0,0 +1,260 @@ +"""Unit tests for the Exa search provider.""" + +from unittest.mock import MagicMock, patch + +import pytest + +from open_deep_research.configuration import SearchAPI +from open_deep_research.utils import ( + _format_exa_response, + exa_search, + exa_search_async, + get_exa_api_key, + get_search_tool, +) + + +def _make_result(**kwargs): + """Build an attribute-style result object similar to the Exa SDK's typed model.""" + return MagicMock(spec_set=list(kwargs.keys()), **kwargs) + + +class TestFormatExaResponse: + """Tests for _format_exa_response normalization and content fallback.""" + + def test_full_content_uses_summary_as_snippet(self): + response = MagicMock() + response.results = [ + _make_result( + title="Hello", + url="https://example.com/a", + text="Long body text", + summary="One-line summary", + highlights=["highlight one", "highlight two"], + score=0.9, + ) + ] + + out = _format_exa_response("hello", response) + + assert out["query"] == "hello" + assert len(out["results"]) == 1 + result = out["results"][0] + assert result["title"] == "Hello" + assert result["url"] == "https://example.com/a" + assert result["content"] == "One-line summary" + assert "Long body text" in result["raw_content"] + assert "highlight one" in result["raw_content"] + assert result["score"] == 0.9 + + def test_only_text_falls_back_to_text_excerpt(self): + response = MagicMock() + response.results = [ + _make_result( + title="Title", + url="https://example.com/b", + text="x" * 2000, + summary=None, + highlights=None, + score=None, + ) + ] + + out = _format_exa_response("q", response) + result = out["results"][0] + + # Snippet uses first 1000 chars of text when summary/highlights are absent + assert result["content"] == "x" * 1000 + assert result["raw_content"] == "x" * 2000 + + def test_only_highlights_used_as_snippet(self): + response = MagicMock() + response.results = [ + _make_result( + title="t", + url="https://example.com/c", + text=None, + summary=None, + highlights=["only highlight"], + score=None, + ) + ] + + out = _format_exa_response("q", response) + assert out["results"][0]["content"] == "only highlight" + assert out["results"][0]["raw_content"] == "only highlight" + + def test_no_content_yields_empty_snippet_and_none_raw(self): + response = MagicMock() + response.results = [ + _make_result( + title="t", + url="https://example.com/d", + text=None, + summary=None, + highlights=None, + score=None, + ) + ] + + out = _format_exa_response("q", response) + assert out["results"][0]["content"] == "" + assert out["results"][0]["raw_content"] is None + + def test_dict_results_are_supported(self): + response = { + "results": [ + { + "title": "Dict Title", + "url": "https://example.com/e", + "text": "body", + "summary": "sum", + "highlights": ["h"], + "score": 0.42, + } + ] + } + + out = _format_exa_response("q", response) + result = out["results"][0] + assert result["title"] == "Dict Title" + assert result["content"] == "sum" + + def test_duplicate_urls_are_skipped(self): + response = MagicMock() + response.results = [ + _make_result( + title="first", + url="https://example.com/dup", + text="t1", + summary="s1", + highlights=[], + score=None, + ), + _make_result( + title="second", + url="https://example.com/dup", + text="t2", + summary="s2", + highlights=[], + score=None, + ), + ] + + out = _format_exa_response("q", response) + assert len(out["results"]) == 1 + assert out["results"][0]["title"] == "first" + + def test_results_without_url_are_skipped(self): + response = MagicMock() + response.results = [ + _make_result( + title="no url", + url=None, + text="t", + summary="s", + highlights=[], + score=None, + ) + ] + + out = _format_exa_response("q", response) + assert out["results"] == [] + + +class TestGetExaApiKey: + """Tests for get_exa_api_key env vs config-driven lookup.""" + + def test_returns_value_from_env(self, monkeypatch): + monkeypatch.setenv("EXA_API_KEY", "env-key") + monkeypatch.delenv("GET_API_KEYS_FROM_CONFIG", raising=False) + assert get_exa_api_key(config=None) == "env-key" + + def test_returns_none_when_unset(self, monkeypatch): + monkeypatch.delenv("EXA_API_KEY", raising=False) + monkeypatch.delenv("GET_API_KEYS_FROM_CONFIG", raising=False) + assert get_exa_api_key(config=None) is None + + def test_returns_value_from_config_when_flag_set(self, monkeypatch): + monkeypatch.setenv("GET_API_KEYS_FROM_CONFIG", "true") + monkeypatch.delenv("EXA_API_KEY", raising=False) + cfg = {"configurable": {"apiKeys": {"EXA_API_KEY": "cfg-key"}}} + assert get_exa_api_key(cfg) == "cfg-key" + + +class TestExaSearchAsync: + """Tests for the async wrapper around the Exa SDK.""" + + @pytest.mark.asyncio + async def test_raises_when_api_key_missing(self, monkeypatch): + monkeypatch.delenv("EXA_API_KEY", raising=False) + monkeypatch.delenv("GET_API_KEYS_FROM_CONFIG", raising=False) + with pytest.raises(ValueError, match="EXA_API_KEY"): + await exa_search_async(["q"], config=None) + + @pytest.mark.asyncio + async def test_rejects_conflicting_domain_filters(self, monkeypatch): + monkeypatch.setenv("EXA_API_KEY", "k") + with pytest.raises(ValueError, match="include_domains and exclude_domains"): + await exa_search_async( + ["q"], + include_domains=["a.com"], + exclude_domains=["b.com"], + config=None, + ) + + @pytest.mark.asyncio + async def test_calls_sdk_and_sets_tracking_header(self, monkeypatch): + monkeypatch.setenv("EXA_API_KEY", "k") + fake_response = MagicMock() + fake_response.results = [] + + fake_client = MagicMock() + fake_client.headers = {} + fake_client.search_and_contents.return_value = fake_response + + with patch("open_deep_research.utils.Exa", return_value=fake_client) as ctor: + results = await exa_search_async( + ["alpha", "beta"], + max_results=3, + search_type="neural", + category="research paper", + include_domains=["arxiv.org"], + start_published_date="2025-01-01", + config=None, + ) + + ctor.assert_called_once_with(api_key="k") + assert fake_client.headers["x-exa-integration"] == "open-deep-research" + assert fake_client.search_and_contents.call_count == 2 + + _, kwargs = fake_client.search_and_contents.call_args_list[0] + assert kwargs["num_results"] == 3 + assert kwargs["type"] == "neural" + assert kwargs["text"] is True + assert kwargs["highlights"] is True + assert kwargs["summary"] is True + assert kwargs["category"] == "research paper" + assert kwargs["include_domains"] == ["arxiv.org"] + assert kwargs["start_published_date"] == "2025-01-01" + assert "exclude_domains" not in kwargs + assert "end_published_date" not in kwargs + + assert len(results) == 2 + assert results[0]["query"] == "alpha" + assert results[1]["query"] == "beta" + + +class TestGetSearchTool: + """Tests for the search tool dispatcher.""" + + @pytest.mark.asyncio + async def test_exa_returns_exa_tool(self): + tools = await get_search_tool(SearchAPI.EXA) + assert len(tools) == 1 + assert tools[0] is exa_search + assert tools[0].metadata.get("name") == "web_search" + + @pytest.mark.asyncio + async def test_none_returns_empty(self): + assert await get_search_tool(SearchAPI.NONE) == []