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
|