agent.py

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 this query relates to. An integer between {min(years)} and {max(years)}.")
        ticker: str = Field(..., description=f"The company ticker this query relates to. 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)
    summarizer = 'vectara-experimental-summary-ext-2023-12-11-med-omni'
    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 = summarizer,
        include_citations = True,
    )

    tools_factory = ToolsFactory()
    return (
            [tools_factory.create_tool(tool) for tool in
                [
                    get_company_info,
                    get_valid_years,
                    get_income_statement,
                ]
            ] +
            tools_factory.financial_tools() +
            [ask_transcripts]
    )

def initialize_agent(_cfg, agent_progress_callback=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, strategy, 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 provided in the output of a tool you use.
    - 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,
        agent_progress_callback=agent_progress_callback,
    )
    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