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 
18
19
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 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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=(
                "the OpenAI SDK is not installed. "
                "Install it with: pip install symfonic-core[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