initial

agent.py

@@ -0,0 +1,112 @@
+from omegaconf import OmegaConf
+import os
+
+from typing import Optional
+from pydantic import Field, BaseModel
+
+from dotenv import load_dotenv
+load_dotenv(override=True)
+
+from vectara_agent.agent import Agent
+from vectara_agent.tools import ToolsFactory, VectaraToolFactory
+
+def create_assistant_tools(cfg):    
+
+    class QueryElectricCars(BaseModel):
+        query: str = Field(description="The user query.")
+
+    vec_factory_1 = VectaraToolFactory(vectara_api_key=cfg.api_keys[0],
+                                        vectara_customer_id=cfg.customer_id, 
+                                        vectara_corpus_id=cfg.corpus_ids[0])
+    
+    ask_vehicles = vec_factory_1.create_rag_tool(
+        tool_name = "ask_vehicles",
+        tool_description = """
+        Given a user query, 
+        returns a response (str) to a user question about electric vehicles based on online resources.
+        You can ask this tool any question about electric cars, including the different types of EVs, how they work, the pros and cons of different models, the environmental impact, and more.
+        """,
+        tool_args_schema = QueryElectricCars,
+        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-sml',
+        include_citations = False,
+    )
+
+    vec_factory_2 = VectaraToolFactory(vectara_api_key=cfg.api_keys[1],
+                                       vectara_customer_id=cfg.customer_id,
+                                       vectara_corpus_id=cfg.corpus_ids[1])
+    
+
+    class QueryEVLaws(BaseModel):
+        query: str = Field(description="The user query")
+        state: Optional[str] = Field(default=None,
+                                     description="The two digit state code. Optional.",
+                                     examples=['CA', 'US', 'WA'])
+        type: Optional[str] = Field(default=None,
+                                    description="The type of policy. Optional",
+                                    examples = ['Laws and Regulations', 'State Incentives', 'Incentives', 'Utility / Private Incentives', 'Programs'])
+        
+    
+
+    ask_policies = vec_factory_2.create_rag_tool(
+        tool_name = "ask_policies",
+        tool_description = """
+        Given a user query,
+        returns a response (str) to a user question about incentives and regulations about electric vehicles in the United States.
+        You can ask this tool any question about laws passed by states or the federal government related to electric vehicles.
+        """,
+        tool_args_schema = QueryEVLaws,
+        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-sml',
+        include_citations = False,
+    )
+
+    tools_factory = ToolsFactory()
+
+    return (tools_factory.standard_tools() + 
+            tools_factory.guardrail_tools() +
+            tools_factory.database_tools(
+                content_description = 'Electric Vehicles',
+                scheme = 'postgresql',
+                host = 'localhost', port = '5432', 
+                user = 'ofer', 
+                password = 'noanoa', 
+                dbname = 'ev_database'
+            ) +
+            [ask_vehicles, ask_policies]
+    )
+
+def initialize_agent(_cfg, update_func):
+    electric_vehicle_bot_instructions = """
+    - You are a helpful research assistant, with expertise in electric vehicles, in conversation with a user.
+    - For a query with multiple sub-questions, break down the query into the sub-questions, 
+      and make separate calls to the ask_vehicles or ask_policies tool to answer each sub-question,
+      then combine the answers to provide a complete response.
+    - Never discuss politics, and always respond politely.
+    """
+
+    agent = Agent(
+        tools=create_assistant_tools(_cfg),
+        topic="Electric vehicles in the United States",
+        custom_instructions=electric_vehicle_bot_instructions,
+        update_func=update_func
+    )
+    return agent
+
+
+def get_agent_config() -> OmegaConf:
+    cfg = OmegaConf.create({
+        'customer_id': str(os.environ['VECTARA_CUSTOMER_ID']),
+        'corpus_ids': str(os.environ['VECTARA_CORPUS_IDS']).split(','),
+        'api_keys': str(os.environ['VECTARA_API_KEYS']).split(','),
+        'examples': os.environ.get('QUERY_EXAMPLES', None),
+        'title': "Electric Vehicles in the United States",
+        'demo_welcome': "Welcome to the EV Assistant demo.",
+        'demo_description': "This assistant can help you learn about electric vehicles in the United States, including how they work, the advantages of purchasing them, and reviews on the top choices.",
+    })
+    return cfg
+

app.py

@@ -1,148 +1,14 @@
-
-import os
 from PIL import Image
 import sys
-import pandas as pd
-import requests
 
-from omegaconf import OmegaConf
 import streamlit as st
 from streamlit_pills import pills
 
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-from pydantic import Field, BaseModel
-from vectara_agent.agent import Agent, AgentStatusType
-from vectara_agent.tools import ToolsFactory, VectaraToolFactory
+from vectara_agent.agent import AgentStatusType
+from agent import initialize_agent, get_agent_config
 
-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=Field(description="the ticker symbol of the company."),
-        year=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)
-            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']])}"
-        else:
-            return "FMP API returned error. 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 compaany 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 = False,
-    )
-
-    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() + 
-            tools_factory.guardrail_tools() +
-            [ask_transcripts]
-    )
-
-def initialize_agent(_cfg):
-    if 'agent' in st.session_state:
-        return st.session_state.agent
-
-    financial_bot_instructions = """
-    - You are a helpful financial assistant, with expertise in financial reporting, in conversation with a user. 
-    - 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.
-    """
-
-    def update_func(status_type: AgentStatusType, msg: str):
-        if status_type != AgentStatusType.AGENT_UPDATE:
-            output = f"{status_type.value} - {msg}"
-            st.session_state.log_messages.append(output)
-
-    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 toggle_logs():
     st.session_state.show_logs = not st.session_state.show_logs
 
@@ -155,6 +21,11 @@ def show_example_questions():
             return True
     return False
 
+def update_func(status_type: AgentStatusType, msg: str):
+    if status_type != AgentStatusType.AGENT_UPDATE:
+        output = f"{status_type.value} - {msg}"
+        st.session_state.log_messages.append(output)
+
 def launch_bot():
     def reset():
         st.session_state.messages = [{"role": "assistant", "content": initial_prompt, "avatar": "🦖"}]
@@ -162,17 +33,13 @@ def launch_bot():
         st.session_state.log_messages = []
         st.session_state.prompt = None
         st.session_state.ex_prompt = None
-        st.session_state.show_logs = False
         st.session_state.first_turn = True
+        st.session_state.show_logs = False
+        if 'agent' not in st.session_state:
+            st.session_state.agent = initialize_agent(cfg, update_func=update_func)
 
-    st.set_page_config(page_title="Financial Assistant", layout="wide")
     if 'cfg' not in st.session_state:
-        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)
-        })
+        cfg = get_agent_config()
         st.session_state.cfg = cfg
         st.session_state.ex_prompt = None
         example_messages = [example.strip() for example in cfg.examples.split(",")] if cfg.examples else []
@@ -180,18 +47,14 @@ def launch_bot():
         reset()
 
     cfg = st.session_state.cfg
-    if 'agent' not in st.session_state:
-        st.session_state.agent = initialize_agent(cfg)
+    st.set_page_config(page_title=cfg['title'], layout="wide")
 
     # left side content
     with st.sidebar:
         image = Image.open('Vectara-logo.png')
         st.image(image, width=175)
-        st.markdown("## Welcome to the financial assistant demo.\n\n\n")
-        companies = ", ".join(tickers.values())
-        st.markdown(
-            f"This assistant can help you with any questions about the financials of several companies:\n\n **{companies}**.\n"
-        )
+        st.markdown(f"## {cfg['demo_welcome']}")
+        st.markdown(f"{cfg['demo_description']}")
 
         st.markdown("\n\n")
         bc1, _ = st.columns([1, 1])
@@ -206,7 +69,6 @@ def launch_bot():
             "This app was built with [Vectara](https://vectara.com).\n\n"
             "It demonstrates the use of Agentic RAG functionality with Vectara"
         )
-        st.markdown("---")
 
     if "messages" not in st.session_state.keys():
         reset()
@@ -249,8 +111,9 @@ def launch_bot():
             st.markdown(res)
         st.session_state.ex_prompt = None
         st.session_state.prompt = None
+        st.session_state.first_turn = False
         st.rerun()
-    
+
     log_placeholder = st.empty()
     with log_placeholder.container():
         if st.session_state.show_logs:
@@ -264,5 +127,4 @@ def launch_bot():
     sys.stdout.flush()
 
 if __name__ == "__main__":
-    launch_bot()
-    
+    launch_bot()