Skip to content

symfonic.core.capabilities.openai_search

OpenAISearcher -- web search via OpenAI API.

OpenAISearcher

OpenAISearcher(api_key: str | None = None)

Search the web via OpenAI's search API.

Requires: pip install openai (optional dependency)

Source code in src/symfonic/core/capabilities/openai_search.py
def __init__(self, api_key: str | None = None) -> None:
    self._api_key = api_key

search async

search(
    query: str, *, limit: int = 10, timeout: float = 30.0
) -> list[SearchResultEvent]

Search the web. Raises CapabilityError on failure.

Source code in src/symfonic/core/capabilities/openai_search.py
async def search(
    self,
    query: str,
    *,
    limit: int = 10,
    timeout: float = 30.0,
) -> list[SearchResultEvent]:
    """Search the web. Raises CapabilityError on failure."""
    validate_query(query)
    try:
        return await asyncio.wait_for(
            self._do_search(query, limit), timeout=timeout
        )
    except TimeoutError as exc:
        raise CapabilityError(
            capability="OpenAISearcher",
            cause=f"Timeout after {timeout}s searching '{query}'",
            code="timeout",
            metadata={"query": query},
        ) from exc
    except ImportError as exc:
        raise CapabilityError(
            capability="OpenAISearcher",
            cause="openai is not installed. pip install openai",
            code="missing_dependency",
        ) from exc
    except CapabilityError:
        raise
    except Exception as e:
        raise CapabilityError(
            capability="OpenAISearcher",
            cause=str(e),
            code="search_error",
            metadata={"query": query},
        ) from e