diff --git a/python/.env.example b/python/.env.example index c3820e5a72ff..7d3fbb345fb8 100644 --- a/python/.env.example +++ b/python/.env.example @@ -21,6 +21,7 @@ WEAVIATE_URL="" WEAVIATE_API_KEY="" GOOGLE_SEARCH_ENGINE_ID="" BRAVE_API_KEY="" +EXA_API_KEY="" REDIS_CONNECTION_STRING="" AZCOSMOS_API="" AZCOSMOS_CONNSTR="" diff --git a/python/samples/concepts/README.md b/python/samples/concepts/README.md index bfde792aed39..8463907edcbb 100644 --- a/python/samples/concepts/README.md +++ b/python/samples/concepts/README.md @@ -229,6 +229,7 @@ - [Bing Text Search as Plugin](./search/bing_text_search_as_plugin.py) - [Brave Text Search as Plugin](./search/brave_text_search_as_plugin.py) +- [Exa Text Search as Plugin](./search/exa_text_search_as_plugin.py) - [Google Text Search as Plugin](./search/google_text_search_as_plugin.py) ### Service Selector - Shows how to create and use a custom service selector class diff --git a/python/samples/concepts/search/exa_text_search_as_plugin.py b/python/samples/concepts/search/exa_text_search_as_plugin.py new file mode 100644 index 000000000000..d4f104c54524 --- /dev/null +++ b/python/samples/concepts/search/exa_text_search_as_plugin.py @@ -0,0 +1,125 @@ +# Copyright (c) Microsoft. All rights reserved. + +from collections.abc import Awaitable, Callable + +from semantic_kernel import Kernel +from semantic_kernel.connectors.ai import FunctionChoiceBehavior +from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings +from semantic_kernel.connectors.exa import ExaSearch +from semantic_kernel.contents import ChatHistory +from semantic_kernel.filters import FilterTypes, FunctionInvocationContext +from semantic_kernel.functions import KernelArguments, KernelParameterMetadata + +""" +This project demonstrates how to integrate the Exa Search API as a plugin into the Semantic Kernel +framework to enable conversational AI capabilities with real-time web information. + +To use Exa Search, you need an API key, which can be obtained by login to +https://dashboard.exa.ai/api-keys and creating a key. +After that store it under the name `EXA_API_KEY` in a .env file or your environment variables. +""" + +kernel = Kernel() +kernel.add_service(OpenAIChatCompletion(service_id="chat")) +kernel.add_function( + plugin_name="exa", + function=ExaSearch().create_search_function( + function_name="exa_search", + description="Get details about Semantic Kernel concepts.", + parameters=[ + KernelParameterMetadata( + name="query", + description="The search query.", + type="str", + is_required=True, + type_object=str, + ), + KernelParameterMetadata( + name="top", + description="The number of results to return.", + type="int", + is_required=False, + default_value=2, + type_object=int, + ), + ], + ), +) +chat_function = kernel.add_function( + prompt="{{$chat_history}}{{$user_input}}", + plugin_name="ChatBot", + function_name="Chat", +) +execution_settings = OpenAIChatPromptExecutionSettings( + service_id="chat", + max_tokens=2000, + temperature=0.7, + top_p=0.8, + function_choice_behavior=FunctionChoiceBehavior.Auto(auto_invoke=True), +) + +history = ChatHistory() +system_message = """ +You are a chat bot, specialized in Semantic Kernel, Microsoft LLM orchestration SDK. +Assume questions are related to that, and use the Exa search plugin to find answers. +""" +history.add_system_message(system_message) +history.add_user_message("Hi there, who are you?") +history.add_assistant_message("I am Mosscap, a chat bot. I'm trying to figure out what people need.") + +arguments = KernelArguments(settings=execution_settings) + + +@kernel.filter(filter_type=FilterTypes.FUNCTION_INVOCATION) +async def log_exa_filter( + context: FunctionInvocationContext, next: Callable[[FunctionInvocationContext], Awaitable[None]] +): + if context.function.plugin_name == "exa": + print("Calling Exa search with arguments:") + if "query" in context.arguments: + print(f' Query: "{context.arguments["query"]}"') + if "top" in context.arguments: + print(f' Top: "{context.arguments["top"]}"') + await next(context) + print("Exa search completed.") + else: + await next(context) + + +async def chat() -> bool: + try: + user_input = input("User:> ") + except KeyboardInterrupt: + print("\n\nExiting chat...") + return False + except EOFError: + print("\n\nExiting chat...") + return False + + if user_input == "exit": + print("\n\nExiting chat...") + return False + arguments["user_input"] = user_input + arguments["chat_history"] = history + result = await kernel.invoke(chat_function, arguments=arguments) + print(f"Mosscap:> {result}") + history.add_user_message(user_input) + history.add_assistant_message(str(result)) + return True + + +async def main(): + chatting = True + print( + "Welcome to the chat bot!\ + \n Type 'exit' to exit.\ + \n Try to find out more about the inner workings of Semantic Kernel." + ) + while chatting: + chatting = await chat() + + +if __name__ == "__main__": + import asyncio + + asyncio.run(main()) diff --git a/python/semantic_kernel/connectors/exa.py b/python/semantic_kernel/connectors/exa.py new file mode 100644 index 000000000000..f8c56566f8b7 --- /dev/null +++ b/python/semantic_kernel/connectors/exa.py @@ -0,0 +1,281 @@ +# Copyright (c) Microsoft. All rights reserved. + +import ast +import logging +import sys +from collections.abc import AsyncIterable, Callable +from inspect import getsource +from typing import Any, ClassVar, Final, Literal +from urllib.parse import unquote_plus + +from httpx import AsyncClient, HTTPStatusError, RequestError +from pydantic import Field, SecretStr, ValidationError + +from semantic_kernel.connectors._search_shared import SearchLambdaVisitor +from semantic_kernel.data.text_search import ( + KernelSearchResults, + SearchOptions, + TextSearch, + TextSearchResult, + TSearchResult, +) +from semantic_kernel.exceptions import ServiceInitializationError, ServiceInvalidRequestError +from semantic_kernel.kernel_pydantic import KernelBaseModel, KernelBaseSettings +from semantic_kernel.kernel_types import OptionalOneOrList +from semantic_kernel.utils.feature_stage_decorator import experimental +from semantic_kernel.utils.telemetry.user_agent import SEMANTIC_KERNEL_USER_AGENT + +if sys.version_info >= (3, 12): + from typing import override +else: + from typing_extensions import override + +logger: logging.Logger = logging.getLogger(__name__) + +# region Constants +DEFAULT_URL: Final[str] = "https://api.exa.ai/search" +QUERY_PARAMETERS: Final[list[str]] = [ + "type", + "category", + "userLocation", + "includeDomains", + "excludeDomains", + "includeText", + "excludeText", +] +LIST_PARAMETERS: Final[list[str]] = [ + "includeDomains", + "excludeDomains", + "includeText", + "excludeText", +] +MAX_TOP: Final[int] = 100 + + +# endregion Constants + + +# region ExaSettings +class ExaSettings(KernelBaseSettings): + """Exa Connector settings. + + The settings are first loaded from environment variables with the prefix 'EXA_'. If the + environment variables are not found, the settings can be loaded from a .env file with the + encoding 'utf-8'. If the settings are not found in the .env file, the settings are ignored; + however, validation will fail alerting that the settings are missing. + + Required settings for prefix 'EXA_' are: + - api_key: SecretStr - The Exa API key (Env var EXA_API_KEY) + + """ + + env_prefix: ClassVar[str] = "EXA_" + + api_key: SecretStr + + +# endregion ExaSettings + + +# region ExaResults +@experimental +class ExaSearchResult(KernelBaseModel): + """A single result from an Exa search.""" + + id: str | None = None + title: str | None = None + url: str | None = None + text: str | None = None + highlights: list[str] | None = None + summary: str | None = None + author: str | None = None + published_date: str | None = Field(default=None, validation_alias="publishedDate") + score: float | None = None + image: str | None = None + favicon: str | None = None + + +@experimental +class ExaSearchResponse(KernelBaseModel): + """The response from an Exa search.""" + + request_id: str | None = Field(default=None, validation_alias="requestId") + results: list[ExaSearchResult] = Field(default_factory=list) + search_time: float | None = Field(default=None, validation_alias="searchTime") + cost_dollars: dict[str, Any] | None = Field(default=None, validation_alias="costDollars") + + +# endregion ExaResults + + +@experimental +class ExaSearch(KernelBaseModel, TextSearch): + """A search engine connector that uses the Exa Search API to perform a web search.""" + + settings: ExaSettings + + def __init__( + self, + api_key: str | None = None, + env_file_path: str | None = None, + env_file_encoding: str | None = None, + ) -> None: + """Initializes a new instance of the Exa Search class. + + Args: + api_key: The Exa Search API key. If provided, will override + the value in the env vars or .env file. + env_file_path: The optional path to the .env file. If provided, + the settings are read from this file path location. + env_file_encoding: The optional encoding of the .env file. If provided, + the settings are read from this file path location. + """ + try: + settings = ExaSettings( + api_key=api_key, + env_file_path=env_file_path, + env_file_encoding=env_file_encoding, + ) + except ValidationError as ex: + raise ServiceInitializationError("Failed to create Exa settings.") from ex + + super().__init__(settings=settings) # type: ignore[call-arg] + + @override + async def search( + self, + query: str, + output_type: type[str] | type[TSearchResult] | Literal["Any"] = str, + *, + filter: OptionalOneOrList[Callable | str] = None, + skip: int = 0, + top: int = 5, + include_total_count: bool = False, + **kwargs: Any, + ) -> "KernelSearchResults[TSearchResult]": + options = SearchOptions(filter=filter, skip=skip, top=top, include_total_count=include_total_count, **kwargs) + results = await self._inner_search(query=query, options=options) + return KernelSearchResults( + results=self._get_result_strings(results) + if output_type is str + else self._get_text_search_results(results) + if output_type is TextSearchResult + else self._get_exa_results(results), + total_count=self._get_total_count(results, options), + metadata=self._get_metadata(results), + ) + + async def _get_result_strings(self, response: ExaSearchResponse) -> AsyncIterable[str]: + for result in response.results: + yield self._get_content(result) + + async def _get_text_search_results(self, response: ExaSearchResponse) -> AsyncIterable[TextSearchResult]: + for result in response.results: + yield TextSearchResult( + name=result.title, + value=self._get_content(result), + link=result.url, + ) + + async def _get_exa_results(self, response: ExaSearchResponse) -> AsyncIterable[ExaSearchResult]: + for result in response.results: + yield result + + @staticmethod + def _get_content(result: ExaSearchResult) -> str: + if result.highlights: + return " ".join(result.highlights) + return result.summary or result.text or "" + + def _get_metadata(self, response: ExaSearchResponse) -> dict[str, Any]: + return { + "request_id": response.request_id, + "search_time": response.search_time, + "cost_dollars": response.cost_dollars, + } + + def _get_total_count(self, response: ExaSearchResponse, options: SearchOptions) -> int | None: + if options.include_total_count: + return len(response.results) + return None + + def _get_options(self, **kwargs: Any) -> SearchOptions: + try: + return SearchOptions(**kwargs) + except ValidationError: + return SearchOptions() + + async def _inner_search(self, query: str, options: SearchOptions) -> ExaSearchResponse: + self._validate_options(options) + + logger.info(f"Received request for exa web search with params:\nnum_results: {options.top}") + + url = self._get_url() + payload = self._build_request_payload(query, options) + + logger.info(f"Sending POST request to {url}") + + headers = { + "x-api-key": self.settings.api_key.get_secret_value(), + "Content-Type": "application/json", + "user_agent": SEMANTIC_KERNEL_USER_AGENT, + "x-exa-integration": "microsoft/semantic-kernel-integration", + } + try: + async with AsyncClient(timeout=30) as client: + response = await client.post(url, headers=headers, json=payload) + response.raise_for_status() + return ExaSearchResponse.model_validate_json(response.text) + except HTTPStatusError as ex: + logger.error(f"Failed to get search results: {ex}") + raise ServiceInvalidRequestError("Failed to get search results.") from ex + except RequestError as ex: + logger.error(f"Client error occurred: {ex}") + raise ServiceInvalidRequestError("A client error occurred while getting search results.") from ex + except Exception as ex: + logger.error(f"An unexpected error occurred: {ex}") + raise ServiceInvalidRequestError("An unexpected error occurred while getting search results.") from ex + + def _validate_options(self, options: SearchOptions) -> None: + if options.top > MAX_TOP: + raise ServiceInvalidRequestError(f"numResults value must be less than or equal to {MAX_TOP}.") + if options.skip: + raise ServiceInvalidRequestError("Exa search does not support skip/offset pagination.") + + def _get_url(self) -> str: + return DEFAULT_URL + + def _parse_filter_lambda(self, filter_lambda: Callable | str) -> list[dict[str, str]]: + """Parse a string lambda or string expression into a list of {field: value} dicts using AST.""" + expr = filter_lambda if isinstance(filter_lambda, str) else getsource(filter_lambda).strip() + tree = ast.parse(expr, mode="eval") + node = tree.body + visitor = SearchLambdaVisitor(valid_parameters=QUERY_PARAMETERS) + visitor.visit(node) + return visitor.filters + + def _build_request_payload(self, query: str, options: SearchOptions) -> dict[str, Any]: + payload: dict[str, Any] = { + "query": query or "", + "type": "auto", + "numResults": options.top, + "contents": {"highlights": True}, + } + if not options.filter: + return payload + filters = options.filter + if not isinstance(filters, list): + filters = [filters] + for f in filters: + try: + for d in self._parse_filter_lambda(f): + for field, value in d.items(): + decoded = unquote_plus(value) + if field in LIST_PARAMETERS: + payload.setdefault(field, []).append(decoded) + else: + payload[field] = decoded + except Exception as exc: + logger.warning(f"Failed to parse filter lambda: {f}, ignoring this filter. Error: {exc}") + continue + return payload diff --git a/python/semantic_kernel/connectors/search.py b/python/semantic_kernel/connectors/search.py index 25ac66dcd64b..5a1e82cc963c 100644 --- a/python/semantic_kernel/connectors/search.py +++ b/python/semantic_kernel/connectors/search.py @@ -13,6 +13,10 @@ "BraveWebPages": ".brave", "BraveWebPage": ".brave", "BraveSearchResponse": ".brave", + "ExaSearch": ".exa", + "ExaSettings": ".exa", + "ExaSearchResult": ".exa", + "ExaSearchResponse": ".exa", } diff --git a/python/semantic_kernel/connectors/search.pyi b/python/semantic_kernel/connectors/search.pyi index 167cf21789e1..3168611d675c 100644 --- a/python/semantic_kernel/connectors/search.pyi +++ b/python/semantic_kernel/connectors/search.pyi @@ -1,6 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. from .brave import BraveSearch, BraveSearchResponse, BraveSettings, BraveWebPage, BraveWebPages +from .exa import ExaSearch, ExaSearchResponse, ExaSearchResult, ExaSettings from .google_search import ( GoogleSearch, GoogleSearchInformation, @@ -15,6 +16,10 @@ __all__ = [ "BraveSettings", "BraveWebPage", "BraveWebPages", + "ExaSearch", + "ExaSearchResponse", + "ExaSearchResult", + "ExaSettings", "GoogleSearch", "GoogleSearchInformation", "GoogleSearchResponse", diff --git a/python/tests/unit/connectors/conftest.py b/python/tests/unit/connectors/conftest.py index bd9111a70c55..0de5fb0d9c2c 100644 --- a/python/tests/unit/connectors/conftest.py +++ b/python/tests/unit/connectors/conftest.py @@ -125,6 +125,28 @@ def brave_unit_test_env(monkeypatch, exclude_list, override_env_param_dict): return env_vars +@fixture() +def exa_unit_test_env(monkeypatch, exclude_list, override_env_param_dict): + """Fixture to set environment variables for ExaConnector.""" + if exclude_list is None: + exclude_list = [] + + if override_env_param_dict is None: + override_env_param_dict = {} + + env_vars = {"EXA_API_KEY": "test_api_key"} + + env_vars.update(override_env_param_dict) + + for key, value in env_vars.items(): + if key not in exclude_list: + monkeypatch.setenv(key, value) + else: + monkeypatch.delenv(key, raising=False) + + return env_vars + + @fixture() def google_search_unit_test_env(monkeypatch, exclude_list, override_env_param_dict): """Fixture to set environment variables for the Google Search Connector.""" diff --git a/python/tests/unit/connectors/search/test_exa_search.py b/python/tests/unit/connectors/search/test_exa_search.py new file mode 100644 index 000000000000..0877e1499238 --- /dev/null +++ b/python/tests/unit/connectors/search/test_exa_search.py @@ -0,0 +1,207 @@ +# Copyright (c) Microsoft. All rights reserved. + +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +from semantic_kernel.connectors.exa import ExaSearch, ExaSearchResponse, ExaSearchResult +from semantic_kernel.data.text_search import KernelSearchResults, TextSearchResult +from semantic_kernel.exceptions import ServiceInitializationError, ServiceInvalidRequestError + + +@pytest.fixture +def exa_search(exa_unit_test_env): + """Set up the fixture to configure the Exa Search for these tests.""" + return ExaSearch() + + +@pytest.fixture +def async_client_mock(): + """Set up the fixture to mock AsyncClient.""" + async_client_mock = AsyncMock() + with patch("semantic_kernel.connectors.exa.AsyncClient.__aenter__", return_value=async_client_mock): + yield async_client_mock + + +async def test_exa_search_init_success(exa_search): + """Test that ExaSearch initializes successfully with valid env.""" + assert exa_search.settings.api_key.get_secret_value() == "test_api_key" + + +@pytest.mark.parametrize("exclude_list", [["EXA_API_KEY"]], indirect=True) +async def test_exa_search_init_validation_error(exa_unit_test_env): + """Test that ExaSearch raises ServiceInitializationError if ExaSettings creation fails.""" + with pytest.raises(ServiceInitializationError): + ExaSearch(env_file_path="invalid.env") + + +async def test_search_success(exa_unit_test_env, async_client_mock): + """Test that search returns KernelSearchResults with the highlights joined.""" + mock_response = ExaSearchResponse( + requestId="req-1", + results=[ + ExaSearchResult( + title="Result Name", + url="https://example.com", + highlights=["First highlight", "second highlight"], + ) + ], + searchTime=12.5, + ) + async_client_mock.post.return_value = MagicMock() + + with patch.object(ExaSearchResponse, "model_validate_json", return_value=mock_response): + search_instance = ExaSearch() + kernel_results: KernelSearchResults[str] = await search_instance.search("Test query", include_total_count=True) + + results_list = [res async for res in kernel_results.results] + + assert results_list == ["First highlight second highlight"] + assert kernel_results.total_count == 1 + assert kernel_results.metadata == {"request_id": "req-1", "search_time": 12.5, "cost_dollars": None} + + +async def test_search_falls_back_to_text(exa_unit_test_env, async_client_mock): + """Test that results without highlights fall back to the summary or full text.""" + mock_response = ExaSearchResponse( + results=[ + ExaSearchResult(title="Summarized", url="https://example.com/1", summary="A summary"), + ExaSearchResult(title="Full text", url="https://example.com/2", text="Page text"), + ExaSearchResult(title="Empty", url="https://example.com/3"), + ] + ) + async_client_mock.post.return_value = MagicMock() + + with patch.object(ExaSearchResponse, "model_validate_json", return_value=mock_response): + kernel_results: KernelSearchResults[str] = await ExaSearch().search("Test query") + + assert [res async for res in kernel_results.results] == ["A summary", "Page text", ""] + + +async def test_get_text_search_results_success(exa_unit_test_env, async_client_mock): + """Test that search with output_type=TextSearchResult returns TextSearchResults.""" + mock_response = ExaSearchResponse( + results=[ + ExaSearchResult(title="Result Name", url="https://example.com", highlights=["Test snippet"]), + ] + ) + async_client_mock.post.return_value = MagicMock() + + with patch.object(ExaSearchResponse, "model_validate_json", return_value=mock_response): + kernel_results: KernelSearchResults[TextSearchResult] = await ExaSearch().search( + "Test query", include_total_count=True, output_type=TextSearchResult + ) + + results_list = [res async for res in kernel_results.results] + + assert len(results_list) == 1 + assert isinstance(results_list[0], TextSearchResult) + assert results_list[0].name == "Result Name" + assert results_list[0].value == "Test snippet" + assert results_list[0].link == "https://example.com" + assert kernel_results.total_count == 1 + + +async def test_get_search_results_success(exa_unit_test_env, async_client_mock): + """Test that search with output_type="Any" returns the raw ExaSearchResults.""" + mock_response = ExaSearchResponse( + results=[ExaSearchResult(title="Result Name", url="https://example.com", publishedDate="2026-01-01")] + ) + async_client_mock.post.return_value = MagicMock() + + with patch.object(ExaSearchResponse, "model_validate_json", return_value=mock_response): + kernel_results: KernelSearchResults[ExaSearchResult] = await ExaSearch().search("Test query", output_type="Any") + + results_list = [res async for res in kernel_results.results] + + assert len(results_list) == 1 + assert isinstance(results_list[0], ExaSearchResult) + assert results_list[0].published_date == "2026-01-01" + + +async def test_search_http_status_error(exa_unit_test_env, async_client_mock): + """Test that search raises ServiceInvalidRequestError on HTTPStatusError.""" + mock_response = MagicMock() + mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( + "Error", request=MagicMock(), response=MagicMock() + ) + async_client_mock.post.return_value = mock_response + + with pytest.raises(ServiceInvalidRequestError) as exc_info: + await ExaSearch().search("Test query") + assert "Failed to get search results." in str(exc_info.value) + + +async def test_search_request_error(exa_unit_test_env, async_client_mock): + """Test that search raises ServiceInvalidRequestError on RequestError.""" + async_client_mock.post.side_effect = httpx.RequestError("Client error") + + with pytest.raises(ServiceInvalidRequestError) as exc_info: + await ExaSearch().search("Test query") + assert "A client error occurred while getting search results." in str(exc_info.value) + + +async def test_search_generic_exception(exa_unit_test_env, async_client_mock): + """Test that search raises ServiceInvalidRequestError on unexpected exception.""" + async_client_mock.post.side_effect = Exception("Something unexpected") + + with pytest.raises(ServiceInvalidRequestError) as exc_info: + await ExaSearch().search("Test query") + assert "An unexpected error occurred while getting search results." in str(exc_info.value) + + +async def test_validate_options_raises_error_for_large_top(exa_search): + """Test that _validate_options raises when top exceeds the API maximum.""" + with pytest.raises(ServiceInvalidRequestError) as exc_info: + await exa_search.search("test", top=101) + assert "numResults value must be less than or equal to 100." in str(exc_info.value) + + +async def test_validate_options_raises_error_for_nonzero_skip(exa_search): + """Test that nonzero skip is rejected because Exa has no offset pagination.""" + with pytest.raises(ServiceInvalidRequestError) as exc_info: + await exa_search.search("test", skip=10) + assert "does not support skip/offset pagination" in str(exc_info.value) + + +def test_build_request_payload_nests_contents(exa_search): + """Test that content extraction options are nested under `contents`.""" + payload = exa_search._build_request_payload("query", exa_search._get_options(top=3)) + + assert payload == { + "query": "query", + "type": "auto", + "numResults": 3, + "contents": {"highlights": True}, + } + + +def test_build_request_payload_with_filters(exa_search): + """Test that filter lambdas map onto Exa query parameters, collecting list-valued ones.""" + payload = exa_search._build_request_payload( + "query", + exa_search._get_options( + top=5, + filter=[ + 'lambda x: x.category == "news"', + 'lambda x: x.includeDomains == "arxiv.org"', + 'lambda x: x.includeDomains == "nature.com"', + 'lambda x: x.includeText == "new york"', + ], + ), + ) + + assert payload["category"] == "news" + assert payload["includeDomains"] == ["arxiv.org", "nature.com"] + # SearchLambdaVisitor quote_plus-encodes; JSON body should get decoded values. + assert payload["includeText"] == ["new york"] + + +def test_build_request_payload_ignores_invalid_filter(exa_search): + """Test that an unparsable filter is ignored instead of failing the request.""" + payload = exa_search._build_request_payload( + "query", exa_search._get_options(top=5, filter="lambda x: x.not_a_parameter == 'value'") + ) + + assert "not_a_parameter" not in payload