File size: 2,988 Bytes
861dfcc
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
from typing import Optional, Type

from asyncer import asyncify

from lagent.actions.base_action import AsyncActionMixin, BaseAction, tool_api
from lagent.actions.parser import BaseParser, JsonParser
from lagent.schema import ActionReturn, ActionStatusCode


class ArxivSearch(BaseAction):
    """Search information from Arxiv.org. \

Useful for when you need to answer questions about Physics, Mathematics, \

Computer Science, Quantitative Biology, Quantitative Finance, Statistics, \

Electrical Engineering, and Economics from scientific articles on arxiv.org.

    """

    def __init__(

        self,

        top_k_results: int = 3,

        max_query_len: int = 300,

        doc_content_chars_max: int = 1500,

        description: Optional[dict] = None,

        parser: Type[BaseParser] = JsonParser,

    ):
        super().__init__(description, parser)
        self.top_k_results = top_k_results
        self.max_query_len = max_query_len
        self.doc_content_chars_max = doc_content_chars_max

    @tool_api(explode_return=True)
    def get_arxiv_article_information(self, query: str) -> dict:
        """Run Arxiv search and get the article meta information.



        Args:

            query (:class:`str`): the content of search query



        Returns:

            :class:`dict`: article information

                * content (str): a list of 3 arxiv search papers

        """
        import arxiv

        try:
            results = arxiv.Search(  # type: ignore
                query[: self.max_query_len], max_results=self.top_k_results
            ).results()
        except Exception as exc:
            return ActionReturn(errmsg=f'Arxiv exception: {exc}', state=ActionStatusCode.HTTP_ERROR)
        docs = [
            f'Published: {result.updated.date()}\nTitle: {result.title}\n'
            f'Authors: {", ".join(a.name for a in result.authors)}\n'
            f'Summary: {result.summary[:self.doc_content_chars_max]}'
            for result in results
        ]
        if docs:
            return {'content': '\n\n'.join(docs)}
        return {'content': 'No good Arxiv Result was found'}


class AsyncArxivSearch(AsyncActionMixin, ArxivSearch):
    """Search information from Arxiv.org. \

Useful for when you need to answer questions about Physics, Mathematics, \

Computer Science, Quantitative Biology, Quantitative Finance, Statistics, \

Electrical Engineering, and Economics from scientific articles on arxiv.org.

    """

    @tool_api(explode_return=True)
    @asyncify
    def get_arxiv_article_information(self, query: str) -> dict:
        """Run Arxiv search and get the article meta information.



        Args:

            query (:class:`str`): the content of search query



        Returns:

            :class:`dict`: article information

                * content (str): a list of 3 arxiv search papers

        """
        return super().get_arxiv_article_information(query)