Edit on GitHub

agent_search_gateway.providers.contracts

Provider-facing contracts and candidate value objects.

View Source

  1"""Provider-facing contracts and candidate value objects."""
  2
  3from collections.abc import Mapping, Sequence
  4from dataclasses import dataclass
  5from datetime import date
  6from typing import Protocol, runtime_checkable
  7
  8from ..models import LLMInvocation, OAResolution
  9from ..url_normalization import NormalizedURL
 10
 11ChatMessage = Mapping[str, str]
 12
 13
 14@dataclass(frozen=True, slots=True)
 15class KeywordSearchHit:
 16    url: str
 17    title: str = ""
 18    snippet: str = ""
 19    raw_content: str = ""
 20    content: str = ""
 21
 22
 23@dataclass(frozen=True, slots=True)
 24class URLFetchCandidate:
 25    raw_content: str
 26    content: str = ""
 27
 28
 29@dataclass(frozen=True, slots=True)
 30class PaperSearchHit:
 31    source: str
 32    source_id: str
 33    title: str
 34    authors: tuple[str, ...] = ()
 35    abstract: str = ""
 36    doi: str = ""
 37    arxiv_id: str = ""
 38    published_date: date | None = None
 39    updated_date: date | None = None
 40    url: str = ""
 41    pdf_url: str = ""
 42    venue: str = ""
 43    topics: tuple[str, ...] = ()
 44    citation_count: int | None = None
 45    is_open_access: bool | None = None
 46    oa_status: str = ""
 47    license: str = ""
 48
 49
 50@dataclass(frozen=True, slots=True)
 51class ProviderCapabilities:
 52    search: bool
 53    fetch: bool
 54
 55
 56@runtime_checkable
 57class KeywordSearchProvider(Protocol):
 58    name: str
 59
 60    async def search(self, query: str) -> list[KeywordSearchHit]: ...
 61
 62
 63@runtime_checkable
 64class URLFetchProvider(Protocol):
 65    name: str
 66
 67    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: ...
 68
 69
 70@runtime_checkable
 71class AcademicSearchProvider(Protocol):
 72    name: str
 73
 74    async def search(self, query: str) -> list[PaperSearchHit]: ...
 75
 76
 77@runtime_checkable
 78class OAResolver(Protocol):
 79    name: str
 80
 81    async def resolve(self, doi: str) -> OAResolution | None: ...
 82
 83
 84@runtime_checkable
 85class LLMClient(Protocol):
 86    name: str
 87
 88    async def complete_text(
 89        self,
 90        invocation: LLMInvocation,
 91        messages: Sequence[ChatMessage],
 92    ) -> str: ...
 93
 94    async def complete_json(
 95        self,
 96        invocation: LLMInvocation,
 97        messages: Sequence[ChatMessage],
 98    ) -> Mapping[str, object]: ...
 99
100    async def aclose(self) -> None: ...

ChatMessage = collections.abc.Mapping[str, str]

@dataclass(frozen=True, slots=True)

class KeywordSearchHit: View Source

15@dataclass(frozen=True, slots=True)
16class KeywordSearchHit:
17    url: str
18    title: str = ""
19    snippet: str = ""
20    raw_content: str = ""
21    content: str = ""

KeywordSearchHit( url: str, title: str = '', snippet: str = '', raw_content: str = '', content: str = '')

url: str

title: str

snippet: str

raw_content: str

content: str

@dataclass(frozen=True, slots=True)

class URLFetchCandidate: View Source

24@dataclass(frozen=True, slots=True)
25class URLFetchCandidate:
26    raw_content: str
27    content: str = ""

URLFetchCandidate(raw_content: str, content: str = '')

raw_content: str

content: str

@dataclass(frozen=True, slots=True)

class PaperSearchHit: View Source

30@dataclass(frozen=True, slots=True)
31class PaperSearchHit:
32    source: str
33    source_id: str
34    title: str
35    authors: tuple[str, ...] = ()
36    abstract: str = ""
37    doi: str = ""
38    arxiv_id: str = ""
39    published_date: date | None = None
40    updated_date: date | None = None
41    url: str = ""
42    pdf_url: str = ""
43    venue: str = ""
44    topics: tuple[str, ...] = ()
45    citation_count: int | None = None
46    is_open_access: bool | None = None
47    oa_status: str = ""
48    license: str = ""

PaperSearchHit( source: str, source_id: str, title: str, authors: tuple[str, ...] = (), abstract: str = '', doi: str = '', arxiv_id: str = '', published_date: datetime.date | None = None, updated_date: datetime.date | None = None, url: str = '', pdf_url: str = '', venue: str = '', topics: tuple[str, ...] = (), citation_count: int | None = None, is_open_access: bool | None = None, oa_status: str = '', license: str = '')

source: str

source_id: str

title: str

authors: tuple[str, ...]

abstract: str

doi: str

arxiv_id: str

published_date: datetime.date | None

updated_date: datetime.date | None

url: str

pdf_url: str

venue: str

topics: tuple[str, ...]

citation_count: int | None

is_open_access: bool | None

oa_status: str

license: str

@dataclass(frozen=True, slots=True)

class ProviderCapabilities: View Source

51@dataclass(frozen=True, slots=True)
52class ProviderCapabilities:
53    search: bool
54    fetch: bool

ProviderCapabilities(search: bool, fetch: bool)

search: bool

fetch: bool

@runtime_checkable

class KeywordSearchProvider(typing.Protocol): View Source

57@runtime_checkable
58class KeywordSearchProvider(Protocol):
59    name: str
60
61    async def search(self, query: str) -> list[KeywordSearchHit]: ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

KeywordSearchProvider(*args, **kwargs) View Source

1739def _no_init_or_replace_init(self, *args, **kwargs):
1740    cls = type(self)
1741
1742    if cls._is_protocol:
1743        raise TypeError('Protocols cannot be instantiated')
1744
1745    # Already using a custom `__init__`. No need to calculate correct
1746    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1747    if cls.__init__ is not _no_init_or_replace_init:
1748        return
1749
1750    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1751    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1752    # searches for a proper new `__init__` in the MRO. The new `__init__`
1753    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1754    # instantiation of the protocol subclass will thus use the new
1755    # `__init__` and no longer call `_no_init_or_replace_init`.
1756    for base in cls.__mro__:
1757        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1758        if init is not _no_init_or_replace_init:
1759            cls.__init__ = init
1760            break
1761    else:
1762        # should not happen
1763        cls.__init__ = object.__init__
1764
1765    cls.__init__(self, *args, **kwargs)

async def search( self, query: str) -> list[KeywordSearchHit]: View Source

61    async def search(self, query: str) -> list[KeywordSearchHit]: ...

@runtime_checkable

class URLFetchProvider(typing.Protocol): View Source

64@runtime_checkable
65class URLFetchProvider(Protocol):
66    name: str
67
68    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

URLFetchProvider(*args, **kwargs) View Source

1739def _no_init_or_replace_init(self, *args, **kwargs):
1740    cls = type(self)
1741
1742    if cls._is_protocol:
1743        raise TypeError('Protocols cannot be instantiated')
1744
1745    # Already using a custom `__init__`. No need to calculate correct
1746    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1747    if cls.__init__ is not _no_init_or_replace_init:
1748        return
1749
1750    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1751    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1752    # searches for a proper new `__init__` in the MRO. The new `__init__`
1753    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1754    # instantiation of the protocol subclass will thus use the new
1755    # `__init__` and no longer call `_no_init_or_replace_init`.
1756    for base in cls.__mro__:
1757        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1758        if init is not _no_init_or_replace_init:
1759            cls.__init__ = init
1760            break
1761    else:
1762        # should not happen
1763        cls.__init__ = object.__init__
1764
1765    cls.__init__(self, *args, **kwargs)

async def fetch( self, url: agent_search_gateway.url_normalization.NormalizedURL) -> URLFetchCandidate: View Source

68    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: ...

@runtime_checkable

class AcademicSearchProvider(typing.Protocol): View Source

71@runtime_checkable
72class AcademicSearchProvider(Protocol):
73    name: str
74
75    async def search(self, query: str) -> list[PaperSearchHit]: ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

AcademicSearchProvider(*args, **kwargs) View Source

1739def _no_init_or_replace_init(self, *args, **kwargs):
1740    cls = type(self)
1741
1742    if cls._is_protocol:
1743        raise TypeError('Protocols cannot be instantiated')
1744
1745    # Already using a custom `__init__`. No need to calculate correct
1746    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1747    if cls.__init__ is not _no_init_or_replace_init:
1748        return
1749
1750    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1751    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1752    # searches for a proper new `__init__` in the MRO. The new `__init__`
1753    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1754    # instantiation of the protocol subclass will thus use the new
1755    # `__init__` and no longer call `_no_init_or_replace_init`.
1756    for base in cls.__mro__:
1757        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1758        if init is not _no_init_or_replace_init:
1759            cls.__init__ = init
1760            break
1761    else:
1762        # should not happen
1763        cls.__init__ = object.__init__
1764
1765    cls.__init__(self, *args, **kwargs)

async def search( self, query: str) -> list[PaperSearchHit]: View Source

75    async def search(self, query: str) -> list[PaperSearchHit]: ...

@runtime_checkable

class OAResolver(typing.Protocol): View Source

78@runtime_checkable
79class OAResolver(Protocol):
80    name: str
81
82    async def resolve(self, doi: str) -> OAResolution | None: ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

OAResolver(*args, **kwargs) View Source

1739def _no_init_or_replace_init(self, *args, **kwargs):
1740    cls = type(self)
1741
1742    if cls._is_protocol:
1743        raise TypeError('Protocols cannot be instantiated')
1744
1745    # Already using a custom `__init__`. No need to calculate correct
1746    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1747    if cls.__init__ is not _no_init_or_replace_init:
1748        return
1749
1750    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1751    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1752    # searches for a proper new `__init__` in the MRO. The new `__init__`
1753    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1754    # instantiation of the protocol subclass will thus use the new
1755    # `__init__` and no longer call `_no_init_or_replace_init`.
1756    for base in cls.__mro__:
1757        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1758        if init is not _no_init_or_replace_init:
1759            cls.__init__ = init
1760            break
1761    else:
1762        # should not happen
1763        cls.__init__ = object.__init__
1764
1765    cls.__init__(self, *args, **kwargs)

async def resolve(self, doi: str) -> agent_search_gateway.models.OAResolution | None: View Source

82    async def resolve(self, doi: str) -> OAResolution | None: ...

@runtime_checkable

class LLMClient(typing.Protocol): View Source

 85@runtime_checkable
 86class LLMClient(Protocol):
 87    name: str
 88
 89    async def complete_text(
 90        self,
 91        invocation: LLMInvocation,
 92        messages: Sequence[ChatMessage],
 93    ) -> str: ...
 94
 95    async def complete_json(
 96        self,
 97        invocation: LLMInvocation,
 98        messages: Sequence[ChatMessage],
 99    ) -> Mapping[str, object]: ...
100
101    async def aclose(self) -> None: ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

LLMClient(*args, **kwargs) View Source

1739def _no_init_or_replace_init(self, *args, **kwargs):
1740    cls = type(self)
1741
1742    if cls._is_protocol:
1743        raise TypeError('Protocols cannot be instantiated')
1744
1745    # Already using a custom `__init__`. No need to calculate correct
1746    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1747    if cls.__init__ is not _no_init_or_replace_init:
1748        return
1749
1750    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1751    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1752    # searches for a proper new `__init__` in the MRO. The new `__init__`
1753    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1754    # instantiation of the protocol subclass will thus use the new
1755    # `__init__` and no longer call `_no_init_or_replace_init`.
1756    for base in cls.__mro__:
1757        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1758        if init is not _no_init_or_replace_init:
1759            cls.__init__ = init
1760            break
1761    else:
1762        # should not happen
1763        cls.__init__ = object.__init__
1764
1765    cls.__init__(self, *args, **kwargs)

async def complete_text( self, invocation: agent_search_gateway.models.LLMInvocation, messages: Sequence[Mapping[str, str]]) -> str: View Source

89    async def complete_text(
90        self,
91        invocation: LLMInvocation,
92        messages: Sequence[ChatMessage],
93    ) -> str: ...

async def complete_json( self, invocation: agent_search_gateway.models.LLMInvocation, messages: Sequence[Mapping[str, str]]) -> Mapping[str, object]: View Source

95    async def complete_json(
96        self,
97        invocation: LLMInvocation,
98        messages: Sequence[ChatMessage],
99    ) -> Mapping[str, object]: ...

async def aclose(self) -> None: View Source

101    async def aclose(self) -> None: ...