Spaces:
Runtime error
Runtime error
File size: 2,513 Bytes
ed4d993 |
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 81 |
"""
Tool for the AskNews API.
To use this tool, you must first set your credentials as environment variables:
ASKNEWS_CLIENT_ID
ASKNEWS_CLIENT_SECRET
"""
from typing import Optional, Type
from langchain_core.callbacks import (
AsyncCallbackManagerForToolRun,
CallbackManagerForToolRun,
)
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.tools import BaseTool
from langchain_community.utilities.asknews import AskNewsAPIWrapper
class SearchInput(BaseModel):
"""Input for the AskNews Search tool."""
query: str = Field(
description="Search query to be used for finding real-time or historical news "
"information."
)
hours_back: Optional[int] = Field(
0,
description="If the Assistant deems that the event may have occurred more "
"than 48 hours ago, it estimates the number of hours back to search. For "
"example, if the event was one month ago, the Assistant may set this to 720. "
"One week would be 168. The Assistant can estimate up to on year back (8760).",
)
class AskNewsSearch(BaseTool):
"""Tool that searches the AskNews API."""
name: str = "asknews_search"
description: str = (
"This tool allows you to perform a search on up-to-date news and historical "
"news. If you needs news from more than 48 hours ago, you can estimate the "
"number of hours back to search."
)
api_wrapper: AskNewsAPIWrapper = Field(default_factory=AskNewsAPIWrapper) # type: ignore[arg-type]
max_results: int = 10
args_schema: Type[BaseModel] = SearchInput
def _run(
self,
query: str,
hours_back: int = 0,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> str:
"""Use the tool."""
try:
return self.api_wrapper.search_news(
query,
hours_back=hours_back,
max_results=self.max_results,
)
except Exception as e:
return repr(e)
async def _arun(
self,
query: str,
hours_back: int = 0,
run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
) -> str:
"""Use the tool asynchronously."""
try:
return await self.api_wrapper.asearch_news(
query,
hours_back=hours_back,
max_results=self.max_results,
)
except Exception as e:
return repr(e)
|