Spaces:
Running
Running
| import os | |
| import pandas as pd | |
| import requests | |
| from pydantic import Field, BaseModel | |
| from omegaconf import OmegaConf | |
| from vectara_agentic.agent import Agent | |
| from vectara_agentic.tools import ToolsFactory, VectaraToolFactory | |
| from dotenv import load_dotenv | |
| load_dotenv(override=True) | |
| tickers = { | |
| "AAPL": "Apple Computer", | |
| "GOOG": "Google", | |
| "AMZN": "Amazon", | |
| "SNOW": "Snowflake", | |
| "TEAM": "Atlassian", | |
| "TSLA": "Tesla", | |
| "NVDA": "Nvidia", | |
| "MSFT": "Microsoft", | |
| "AMD": "Advanced Micro Devices", | |
| "INTC": "Intel", | |
| "NFLX": "Netflix", | |
| } | |
| years = [2020, 2021, 2022, 2023, 2024] | |
| initial_prompt = "How can I help you today?" | |
| def create_assistant_tools(cfg): | |
| def get_company_info() -> list[str]: | |
| """ | |
| Returns a dictionary of companies you can query about. Always check this before using any other tool. | |
| The output is a dictionary of valid ticker symbols mapped to company names. | |
| You can use this to identify the companies you can query about, and their ticker information. | |
| """ | |
| return tickers | |
| def get_valid_years() -> list[str]: | |
| """ | |
| Returns a list of the years for which financial reports are available. | |
| Always check this before using any other tool. | |
| """ | |
| return years | |
| # Tool to get the income statement for a given company and year using the FMP API | |
| def get_income_statement( | |
| ticker: str = Field(description="the ticker symbol of the company."), | |
| year: int = Field(description="the year for which to get the income statement."), | |
| ) -> str: | |
| """ | |
| Get the income statement for a given company and year using the FMP (https://financialmodelingprep.com) API. | |
| Returns a dictionary with the income statement data. All data is in USD, but you can convert it to more compact form like K, M, B. | |
| """ | |
| fmp_api_key = os.environ.get("FMP_API_KEY", None) | |
| if fmp_api_key is None: | |
| return "FMP_API_KEY environment variable not set. This tool does not work." | |
| url = f"https://financialmodelingprep.com/api/v3/income-statement/{ticker}?apikey={fmp_api_key}" | |
| response = requests.get(url) | |
| if response.status_code == 200: | |
| data = response.json() | |
| income_statement = pd.DataFrame(data) | |
| if len(income_statement) == 0 or "date" not in income_statement.columns: | |
| return "No data found for the given ticker symbol." | |
| income_statement["date"] = pd.to_datetime(income_statement["date"]) | |
| income_statement_specific_year = income_statement[ | |
| income_statement["date"].dt.year == int(year) | |
| ] | |
| values_dict = income_statement_specific_year.to_dict(orient="records")[0] | |
| return f"Financial results: {', '.join([f'{key}: {value}' for key, value in values_dict.items() if key not in ['date', 'cik', 'link', 'finalLink']])}" | |
| return f"FMP API returned error {response.status_code}. This tool does not work." | |
| class QueryTranscriptsArgs(BaseModel): | |
| query: str = Field(..., description="The user query.") | |
| year: int = Field(..., description=f"The year. An integer between {min(years)} and {max(years)}.") | |
| ticker: str = Field(..., description=f"The company ticker. Must be a valid ticket symbol from the list {tickers.keys()}.") | |
| vec_factory = VectaraToolFactory(vectara_api_key=cfg.api_key, | |
| vectara_customer_id=cfg.customer_id, | |
| vectara_corpus_id=cfg.corpus_id) | |
| tools_factory = ToolsFactory() | |
| ask_transcripts = vec_factory.create_rag_tool( | |
| tool_name = "ask_transcripts", | |
| tool_description = """ | |
| Given a company name and year, responds to a user question about the company, based on analyst call transcripts about the company's financial reports for that year. | |
| You can ask this tool any question about the company including risks, opportunities, financial performance, competitors and more. | |
| """, | |
| tool_args_schema = QueryTranscriptsArgs, | |
| reranker = "multilingual_reranker_v1", rerank_k = 100, | |
| n_sentences_before = 2, n_sentences_after = 2, lambda_val = 0.005, | |
| summary_num_results = 10, | |
| vectara_summarizer = 'vectara-summary-ext-24-05-med-omni', | |
| include_citations = True, | |
| ) | |
| return ( | |
| [tools_factory.create_tool(tool) for tool in | |
| [ | |
| get_company_info, | |
| get_valid_years, | |
| get_income_statement, | |
| ] | |
| ] + | |
| tools_factory.standard_tools() + | |
| tools_factory.financial_tools() + | |
| [ask_transcripts] | |
| ) | |
| def initialize_agent(_cfg, update_func=None): | |
| financial_bot_instructions = """ | |
| - You are a helpful financial assistant, with expertise in financial reporting, in conversation with a user. | |
| - Use the ask_transcripts tool to answer most questions about the company's financial performance, risks, opportunities, competitors, and more. | |
| - responses from ask_transcripts are summarized. You don't need to further summarize them. | |
| - Use the get_income_statement tool only to receive financial data like revenue, net income, and other financial metrics for a specific company and year. | |
| - Respond in a compact format by using appropriate units of measure (e.g., K for thousands, M for millions, B for billions). | |
| Do not report the same number twice (e.g. $100K and 100,000 USD). | |
| - Always check the get_company_info and get_valid_years tools to validate company and year are valid. | |
| - Do not include URLS unless they are from one of the tools. | |
| - When querying a tool for a numeric value or KPI, use a concise and non-ambiguous description of what you are looking for. | |
| - If you calculate a metric, make sure you have all the necessary information to complete the calculation. Don't guess. | |
| """ | |
| agent = Agent( | |
| tools=create_assistant_tools(_cfg), | |
| topic="Financial data, annual reports and 10-K filings", | |
| custom_instructions=financial_bot_instructions, | |
| update_func=update_func, | |
| ) | |
| agent.report() | |
| return agent | |
| def get_agent_config() -> OmegaConf: | |
| companies = ", ".join(tickers.values()) | |
| cfg = OmegaConf.create({ | |
| 'customer_id': str(os.environ['VECTARA_CUSTOMER_ID']), | |
| 'corpus_id': str(os.environ['VECTARA_CORPUS_ID']), | |
| 'api_key': str(os.environ['VECTARA_API_KEY']), | |
| 'examples': os.environ.get('QUERY_EXAMPLES', None), | |
| 'demo_name': "finance-chat", | |
| 'demo_welcome': "Welcome to the Financial Assistant demo.", | |
| 'demo_description': f"This assistant can help you with any questions about the financials of several companies:\n\n **{companies}**.\n" | |
| }) | |
| return cfg | |