Source code for camel.toolkits.ask_news_toolkit

# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
import os
from datetime import datetime
from typing import List, Literal, Optional, Tuple, Union

from camel.toolkits import FunctionTool
from camel.toolkits.base import BaseToolkit


def _process_response(
    response, return_type: str
) -> Union[str, dict, Tuple[str, dict]]:
    r"""Process the response based on the specified return type.

    This helper method processes the API response and returns the content
    in the specified format, which could be a string, a dictionary, or
    both.

    Args:
        response: The response object returned by the API call.
        return_type (str): Specifies the format of the return value. It
            can be "string" to return the response as a string, "dicts" to
            return it as a dictionary, or "both" to return both formats as
            a tuple.

    Returns:
        Union[str, dict, Tuple[str, dict]]: The processed response,
            formatted according to the return_type argument. If "string",
            returns the response as a string. If "dicts", returns the
            response as a dictionary. If "both", returns a tuple
            containing both formats.

    Raises:
        ValueError: If the return_type provided is invalid.
    """
    if return_type == "string":
        return response.as_string
    elif return_type == "dicts":
        return response.as_dicts
    elif return_type == "both":
        return (response.as_string, response.as_dicts)
    else:
        raise ValueError(f"Invalid return_type: {return_type}")


[docs] class AskNewsToolkit(BaseToolkit): r"""A class representing a toolkit for interacting with the AskNews API. This class provides methods for fetching news, stories, and other content based on user queries using the AskNews API. """ def __init__(self): r"""Initialize the AskNewsToolkit with API clients.The API keys and credentials are retrieved from environment variables. """ from asknews_sdk import AskNewsSDK client_id = os.environ.get("ASKNEWS_CLIENT_ID") client_secret = os.environ.get("ASKNEWS_CLIENT_SECRET") self.asknews_client = AskNewsSDK(client_id, client_secret)
[docs] def get_news( self, query: str, n_articles: int = 10, return_type: Literal["string", "dicts", "both"] = "string", method: Literal["nl", "kw"] = "kw", ) -> Union[str, dict, Tuple[str, dict]]: r"""Fetch news or stories based on a user query. Args: query (str): The search query for fetching relevant news. n_articles (int): Number of articles to include in the response. (default: :obj:`10`) return_type (Literal["string", "dicts", "both"]): The format of the return value. (default: :obj:`"string"`) method (Literal["nl", "kw"]): The search method, either "nl" for natural language or "kw" for keyword search. (default: :obj:`"kw"`) Returns: Union[str, dict, Tuple[str, dict]]: A string, dictionary, or both containing the news or story content, or error message if the process fails. """ try: response = self.asknews_client.news.search_news( query=query, n_articles=n_articles, return_type=return_type, method=method, ) return _process_response(response, return_type) except Exception as e: return f"Got error: {e}"
[docs] def get_stories( self, query: str, categories: List[ Literal[ 'Politics', 'Economy', 'Finance', 'Science', 'Technology', 'Sports', 'Climate', 'Environment', 'Culture', 'Entertainment', 'Business', 'Health', 'International', ] ], reddit: int = 3, expand_updates: bool = True, max_updates: int = 2, max_articles: int = 10, ) -> Union[dict, str]: r"""Fetch stories based on the provided parameters. Args: query (str): The search query for fetching relevant stories. categories (list): The categories to filter stories by. reddit (int): Number of Reddit threads to include. (default: :obj:`3`) expand_updates (bool): Whether to include detailed updates. (default: :obj:`True`) max_updates (int): Maximum number of recent updates per story. (default: :obj:`2`) max_articles (int): Maximum number of articles associated with each update. (default: :obj:`10`) Returns: Unio[dict, str]: A dictionary containing the stories and their associated data, or error message if the process fails. """ try: response = self.asknews_client.stories.search_stories( query=query, categories=categories, reddit=reddit, expand_updates=expand_updates, max_updates=max_updates, max_articles=max_articles, ) # Collect only the headline and story content from the updates stories_data = { "stories": [ { "headline": story.updates[0].headline, "updates": [ { "headline": update.headline, "story": update.story, } for update in story.updates[:max_updates] ], } for story in response.stories ] } return stories_data except Exception as e: return f"Got error: {e}"
[docs] def search_reddit( self, keywords: List[str], n_threads: int = 5, return_type: Literal["string", "dicts", "both"] = "string", method: Literal["nl", "kw"] = "kw", ) -> Union[str, dict, Tuple[str, dict]]: r"""Search Reddit based on the provided keywords. Args: keywords (List[str]): The keywords to search for on Reddit. n_threads (int): Number of Reddit threads to summarize and return. (default: :obj:`5`) return_type (Literal["string", "dicts", "both"]): The format of the return value. (default: :obj:`"string"`) method (Literal["nl", "kw"]): The search method, either "nl" for natural language or "kw" for keyword search. (default: :obj:`"kw"`) Returns: Union[str, dict, Tuple[str, dict]]: The Reddit search results as a string, dictionary, or both, or error message if the process fails. """ try: response = self.asknews_client.news.search_reddit( keywords=keywords, n_threads=n_threads, method=method ) return _process_response(response, return_type) except Exception as e: return f"Got error: {e}"
[docs] def query_finance( self, asset: Literal[ 'bitcoin', 'ethereum', 'cardano', 'uniswap', 'ripple', 'solana', 'polkadot', 'polygon', 'chainlink', 'tether', 'dogecoin', 'monero', 'tron', 'binance', 'aave', 'tesla', 'microsoft', 'amazon', ], metric: Literal[ 'news_positive', 'news_negative', 'news_total', 'news_positive_weighted', 'news_negative_weighted', 'news_total_weighted', ] = "news_positive", return_type: Literal["list", "string"] = "string", date_from: Optional[datetime] = None, date_to: Optional[datetime] = None, ) -> Union[list, str]: r"""Fetch asset sentiment data for a given asset, metric, and date range. Args: asset (Literal): The asset for which to fetch sentiment data. metric (Literal): The sentiment metric to analyze. return_type (Literal["list", "string"]): The format of the return value. (default: :obj:`"string"`) date_from (datetime, optional): The start date and time for the data in ISO 8601 format. date_to (datetime, optional): The end date and time for the data in ISO 8601 format. Returns: Union[list, str]: A list of dictionaries containing the datetime and value or a string describing all datetime and value pairs for providing quantified time-series data for news sentiment on topics of interest, or an error message if the process fails. """ try: response = self.asknews_client.analytics.get_asset_sentiment( asset=asset, metric=metric, date_from=date_from, date_to=date_to, ) time_series_data = response.data.timeseries if return_type == "list": return time_series_data elif return_type == "string": header = ( f"This is the sentiment analysis for '{asset}' based " + f"on the '{metric}' metric from {date_from} to {date_to}" + ". The values reflect the aggregated sentiment from news" + " sources for each given time period.\n" ) descriptive_text = "\n".join( [ f"On {entry.datetime}, the sentiment value was " f"{entry.value}." for entry in time_series_data ] ) return header + descriptive_text except Exception as e: return f"Got error: {e}"
[docs] def get_tools(self) -> List[FunctionTool]: r"""Returns a list of FunctionTool objects representing the functions in the toolkit. Returns: List[FunctionTool]: A list of FunctionTool objects representing the functions in the toolkit. """ return [ FunctionTool(self.get_news), FunctionTool(self.get_stories), FunctionTool(self.get_web_search), FunctionTool(self.search_reddit), FunctionTool(self.query_finance), ]
[docs] class AsyncAskNewsToolkit(BaseToolkit): r"""A class representing a toolkit for interacting with the AskNews API asynchronously. This class provides methods for fetching news, stories, and other content based on user queries using the AskNews API. """ def __init__(self): r"""Initialize the AsyncAskNewsToolkit with API clients.The API keys and credentials are retrieved from environment variables. """ from asknews_sdk import AsyncAskNewsSDK # type: ignore[import] client_id = os.environ.get("ASKNEWS_CLIENT_ID") client_secret = os.environ.get("ASKNEWS_CLIENT_SECRET") self.asknews_client = AsyncAskNewsSDK(client_id, client_secret)
[docs] async def get_news( self, query: str, n_articles: int = 10, return_type: Literal["string", "dicts", "both"] = "string", method: Literal["nl", "kw"] = "kw", ) -> Union[str, dict, Tuple[str, dict]]: r"""Fetch news or stories based on a user query. Args: query (str): The search query for fetching relevant news or stories. n_articles (int): Number of articles to include in the response. (default: :obj:10) return_type (Literal["string", "dicts", "both"]): The format of the return value. (default: :obj:"string") method (Literal["nl", "kw"]): The search method, either "nl" for natural language or "kw" for keyword search. (default: :obj:"kw") Returns: Union[str, dict, Tuple[str, dict]]: A string, dictionary, or both containing the news or story content, or error message if the process fails. """ try: response = await self.asknews_client.news.search_news( query=query, n_articles=n_articles, return_type=return_type, method=method, ) return _process_response(response, return_type) except Exception as e: return f"Got error: {e}"
[docs] async def get_stories( self, query: str, categories: List[ Literal[ 'Politics', 'Economy', 'Finance', 'Science', 'Technology', 'Sports', 'Climate', 'Environment', 'Culture', 'Entertainment', 'Business', 'Health', 'International', ] ], reddit: int = 3, expand_updates: bool = True, max_updates: int = 2, max_articles: int = 10, ) -> Union[dict, str]: r"""Fetch stories based on the provided parameters. Args: query (str): The search query for fetching relevant stories. categories (list): The categories to filter stories by. reddit (int): Number of Reddit threads to include. (default: :obj:`3`) expand_updates (bool): Whether to include detailed updates. (default: :obj:`True`) max_updates (int): Maximum number of recent updates per story. (default: :obj:`2`) max_articles (int): Maximum number of articles associated with each update. (default: :obj:`10`) Returns: Unio[dict, str]: A dictionary containing the stories and their associated data, or error message if the process fails. """ try: response = await self.asknews_client.stories.search_stories( query=query, categories=categories, reddit=reddit, expand_updates=expand_updates, max_updates=max_updates, max_articles=max_articles, ) # Collect only the headline and story content from the updates stories_data = { "stories": [ { "headline": story.updates[0].headline, "updates": [ { "headline": update.headline, "story": update.story, } for update in story.updates[:max_updates] ], } for story in response.stories ] } return stories_data except Exception as e: return f"Got error: {e}"
[docs] async def search_reddit( self, keywords: List[str], n_threads: int = 5, return_type: Literal["string", "dicts", "both"] = "string", method: Literal["nl", "kw"] = "kw", ) -> Union[str, dict, Tuple[str, dict]]: r"""Search Reddit based on the provided keywords. Args: keywords (list): The keywords to search for on Reddit. n_threads (int): Number of Reddit threads to summarize and return. (default: :obj:5) return_type (Literal["string", "dicts", "both"]): The format of the return value. (default: :obj:"string") method (Literal["nl", "kw"]): The search method, either "nl" for natural language or "kw" for keyword search. (default: :obj:"kw") Returns: Union[str, dict, Tuple[str, dict]]: The Reddit search results as a string, dictionary, or both, or error message if the process fails. """ try: response = await self.asknews_client.news.search_reddit( keywords=keywords, n_threads=n_threads, method=method ) return _process_response(response, return_type) except Exception as e: return f"Got error: {e}"
[docs] async def query_finance( self, asset: Literal[ 'bitcoin', 'ethereum', 'cardano', 'uniswap', 'ripple', 'solana', 'polkadot', 'polygon', 'chainlink', 'tether', 'dogecoin', 'monero', 'tron', 'binance', 'aave', 'tesla', 'microsoft', 'amazon', ], metric: Literal[ 'news_positive', 'news_negative', 'news_total', 'news_positive_weighted', 'news_negative_weighted', 'news_total_weighted', ] = "news_positive", return_type: Literal["list", "string"] = "string", date_from: Optional[datetime] = None, date_to: Optional[datetime] = None, ) -> Union[list, str]: r"""Fetch asset sentiment data for a given asset, metric, and date range. Args: asset (Literal): The asset for which to fetch sentiment data. metric (Literal): The sentiment metric to analyze. return_type (Literal["list", "string"]): The format of the return value. (default: :obj:`"string"`) date_from (datetime, optional): The start date and time for the data in ISO 8601 format. date_to (datetime, optional): The end date and time for the data in ISO 8601 format. Returns: Union[list, str]: A list of dictionaries containing the datetime and value or a string describing all datetime and value pairs for providing quantified time-series data for news sentiment on topics of interest, or an error message if the process fails. """ try: response = await self.asknews_client.analytics.get_asset_sentiment( asset=asset, metric=metric, date_from=date_from, date_to=date_to, ) time_series_data = response.data.timeseries if return_type == "list": return time_series_data elif return_type == "string": header = ( f"This is the sentiment analysis for '{asset}' based " + f"on the '{metric}' metric from {date_from} to {date_to}" + ". The values reflect the aggregated sentiment from news" + " sources for each given time period.\n" ) descriptive_text = "\n".join( [ f"On {entry.datetime}, the sentiment value was " f"{entry.value}." for entry in time_series_data ] ) return header + descriptive_text except Exception as e: return f"Got error: {e}"
[docs] def get_tools(self) -> List[FunctionTool]: r"""Returns a list of FunctionTool objects representing the functions in the toolkit. Returns: List[FunctionTool]: A list of FunctionTool objects representing the functions in the toolkit. """ return [ FunctionTool(self.get_news), FunctionTool(self.get_stories), FunctionTool(self.get_web_search), FunctionTool(self.search_reddit), FunctionTool(self.query_finance), ]