LangChain Agents¶

Agents are often useful in the RAG setting to retrieve real-time information to be used for question answering.

This example utilizes the openai functions agent to reliably call and return structured responses from particular tools. Certain OpenAI models have been fine-tuned for this capability to detect when a particular function should be called and respond with the inputs required for that function. Compared to a ReACT framework that generates reasoning and actions in an interleaving manner, this strategy can often be more reliable and consistent.

In either case - as the questions change over time, different agents may be needed to retrieve the most useful context. In this example you will create a langchain agent and use TruLens to identify gaps in tool coverage. By quickly identifying this gap, we can quickly add the missing tools to the application and improve the quality of the answers.

Import from LangChain and TruLens¶

Install additional packages¶

In addition to trulens and langchain, we will also need additional packages: yfinance and google-search-results.

In [ ]:

# !pip install trulens trulens-apps-langchain trulens-providers-openai langchain>=0.0.248 openai>=1.0 yfinance>=0.2.27 google-search-results>=2.4.2

# !pip install trulens trulens-apps-langchain trulens-providers-openai langchain>=0.0.248 openai>=1.0 yfinance>=0.2.27 google-search-results>=2.4.2

In [ ]:

from datetime import datetime
from datetime import timedelta
from typing import Type

from langchain import SerpAPIWrapper
from langchain.agents import AgentType
from langchain.agents import Tool
from langchain.agents import initialize_agent
from langchain.chat_models import ChatOpenAI
from langchain.tools import BaseTool
from pydantic import BaseModel
from pydantic import Field
from trulens.core import Feedback
from trulens.core import TruSession
from trulens.apps.langchain import TruChain
from trulens.providers.openai import OpenAI as fOpenAI
import yfinance as yf

session = TruSession()

from datetime import datetime from datetime import timedelta from typing import Type from langchain import SerpAPIWrapper from langchain.agents import AgentType from langchain.agents import Tool from langchain.agents import initialize_agent from langchain.chat_models import ChatOpenAI from langchain.tools import BaseTool from pydantic import BaseModel from pydantic import Field from trulens.core import Feedback from trulens.core import TruSession from trulens.apps.langchain import TruChain from trulens.providers.openai import OpenAI as fOpenAI import yfinance as yf session = TruSession()

Setup¶

Add API keys¶

For this quickstart you will need Open AI and SERP API keys.

In [ ]:

import os

os.environ["OPENAI_API_KEY"] = "sk-..."
os.environ["SERPAPI_API_KEY"] = "..."

import os os.environ["OPENAI_API_KEY"] = "sk-..." os.environ["SERPAPI_API_KEY"] = "..."

Create agent with search tool¶

In [ ]:

search = SerpAPIWrapper()
search_tool = Tool(
    name="Search",
    func=search.run,
    description="useful for when you need to answer questions about current events",
)

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

tools = [search_tool]

agent = initialize_agent(
    tools, llm, agent=AgentType.OPENAI_FUNCTIONS, verbose=True
)

search = SerpAPIWrapper() search_tool = Tool( name="Search", func=search.run, description="useful for when you need to answer questions about current events", ) llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) tools = [search_tool] agent = initialize_agent( tools, llm, agent=AgentType.OPENAI_FUNCTIONS, verbose=True )

Set up Evaluation¶

In [ ]:

class OpenAI_custom(fOpenAI):
    def no_answer_feedback(self, question: str, response: str) -> float:
        return (
            float(
                self.endpoint.client.chat.completions.create(
                    model="gpt-3.5-turbo",
                    messages=[
                        {
                            "role": "system",
                            "content": "Does the RESPONSE provide an answer to the QUESTION? Rate on a scale of 1 to 10. Respond with the number only.",
                        },
                        {
                            "role": "user",
                            "content": f"QUESTION: {question}; RESPONSE: {response}",
                        },
                    ],
                )
                .choices[0]
                .message.content
            )
            / 10
        )


custom = OpenAI_custom()

# No answer feedback (custom)
f_no_answer = Feedback(custom.no_answer_feedback).on_input_output()

class OpenAI_custom(fOpenAI): def no_answer_feedback(self, question: str, response: str) -> float: return ( float( self.endpoint.client.chat.completions.create( model="gpt-3.5-turbo", messages=[ { "role": "system", "content": "Does the RESPONSE provide an answer to the QUESTION? Rate on a scale of 1 to 10. Respond with the number only.", }, { "role": "user", "content": f"QUESTION: {question}; RESPONSE: {response}", }, ], ) .choices[0] .message.content ) / 10 ) custom = OpenAI_custom() # No answer feedback (custom) f_no_answer = Feedback(custom.no_answer_feedback).on_input_output()

In [ ]:

tru_agent = TruChain(agent, app_name="Search_Agent", app_version="v1", feedbacks=[f_no_answer])

tru_agent = TruChain(agent, app_name="Search_Agent", app_version="v1", feedbacks=[f_no_answer])

In [ ]:

prompts = [
    "What company acquired MosaicML?",
    "What's the best way to travel from NYC to LA?",
    "How did the change in the exchange rate during 2021 affect the stock price of US based companies?",
    "Compare the stock performance of Google and Microsoft",
    "What is the highest market cap airline that flies from Los Angeles to New York City?",
    "I'm interested in buying a new smartphone from the producer with the highest stock price. Which company produces the smartphone I should by and what is their current stock price?",
]

with tru_agent as recording:
    for prompt in prompts:
        agent(prompt)

prompts = [ "What company acquired MosaicML?", "What's the best way to travel from NYC to LA?", "How did the change in the exchange rate during 2021 affect the stock price of US based companies?", "Compare the stock performance of Google and Microsoft", "What is the highest market cap airline that flies from Los Angeles to New York City?", "I'm interested in buying a new smartphone from the producer with the highest stock price. Which company produces the smartphone I should by and what is their current stock price?", ] with tru_agent as recording: for prompt in prompts: agent(prompt)

After running the first set of prompts, we notice that our agent is struggling with questions around stock performance.

In response, we can create some custom tools that use yahoo finance to get stock performance information.

Define custom functions¶

In [ ]:

def get_current_stock_price(ticker):
    """Method to get current stock price"""

    ticker_data = yf.Ticker(ticker)
    recent = ticker_data.history(period="1d")
    return {
        "price": recent.iloc[0]["Close"],
        "currency": ticker_data.info["currency"],
    }


def get_stock_performance(ticker, days):
    """Method to get stock price change in percentage"""

    past_date = datetime.today() - timedelta(days=days)
    ticker_data = yf.Ticker(ticker)
    history = ticker_data.history(start=past_date)
    old_price = history.iloc[0]["Close"]
    current_price = history.iloc[-1]["Close"]
    return {"percent_change": ((current_price - old_price) / old_price) * 100}

def get_current_stock_price(ticker): """Method to get current stock price""" ticker_data = yf.Ticker(ticker) recent = ticker_data.history(period="1d") return { "price": recent.iloc[0]["Close"], "currency": ticker_data.info["currency"], } def get_stock_performance(ticker, days): """Method to get stock price change in percentage""" past_date = datetime.today() - timedelta(days=days) ticker_data = yf.Ticker(ticker) history = ticker_data.history(start=past_date) old_price = history.iloc[0]["Close"] current_price = history.iloc[-1]["Close"] return {"percent_change": ((current_price - old_price) / old_price) * 100}

Make custom tools¶

In [ ]:

class CurrentStockPriceInput(BaseModel):
    """Inputs for get_current_stock_price"""

    ticker: str = Field(description="Ticker symbol of the stock")


class CurrentStockPriceTool(BaseTool):
    name = "get_current_stock_price"
    description = """
        Useful when you want to get current stock price.
        You should enter the stock ticker symbol recognized by the yahoo finance
        """
    args_schema: Type[BaseModel] = CurrentStockPriceInput

    def _run(self, ticker: str):
        price_response = get_current_stock_price(ticker)
        return price_response


current_stock_price_tool = CurrentStockPriceTool()


class StockPercentChangeInput(BaseModel):
    """Inputs for get_stock_performance"""

    ticker: str = Field(description="Ticker symbol of the stock")
    days: int = Field(
        description="Timedelta days to get past date from current date"
    )


class StockPerformanceTool(BaseTool):
    name = "get_stock_performance"
    description = """
        Useful when you want to check performance of the stock.
        You should enter the stock ticker symbol recognized by the yahoo finance.
        You should enter days as number of days from today from which performance needs to be check.
        output will be the change in the stock price represented as a percentage.
        """
    args_schema: Type[BaseModel] = StockPercentChangeInput

    def _run(self, ticker: str, days: int):
        response = get_stock_performance(ticker, days)
        return response


stock_performance_tool = StockPerformanceTool()

class CurrentStockPriceInput(BaseModel): """Inputs for get_current_stock_price""" ticker: str = Field(description="Ticker symbol of the stock") class CurrentStockPriceTool(BaseTool): name = "get_current_stock_price" description = """ Useful when you want to get current stock price. You should enter the stock ticker symbol recognized by the yahoo finance """ args_schema: Type[BaseModel] = CurrentStockPriceInput def _run(self, ticker: str): price_response = get_current_stock_price(ticker) return price_response current_stock_price_tool = CurrentStockPriceTool() class StockPercentChangeInput(BaseModel): """Inputs for get_stock_performance""" ticker: str = Field(description="Ticker symbol of the stock") days: int = Field( description="Timedelta days to get past date from current date" ) class StockPerformanceTool(BaseTool): name = "get_stock_performance" description = """ Useful when you want to check performance of the stock. You should enter the stock ticker symbol recognized by the yahoo finance. You should enter days as number of days from today from which performance needs to be check. output will be the change in the stock price represented as a percentage. """ args_schema: Type[BaseModel] = StockPercentChangeInput def _run(self, ticker: str, days: int): response = get_stock_performance(ticker, days) return response stock_performance_tool = StockPerformanceTool()

Give our agent the new finance tools¶

In [ ]:

tools = [search_tool, current_stock_price_tool, stock_performance_tool]

agent = initialize_agent(
    tools, llm, agent=AgentType.OPENAI_FUNCTIONS, verbose=True
)

tools = [search_tool, current_stock_price_tool, stock_performance_tool] agent = initialize_agent( tools, llm, agent=AgentType.OPENAI_FUNCTIONS, verbose=True )

Set up Tracking + Eval¶

In [ ]:

tru_agent = TruChain(agent, app_name="Search_Agent", app_version="v2", feedbacks=[f_no_answer])

tru_agent = TruChain(agent, app_name="Search_Agent", app_version="v2", feedbacks=[f_no_answer])

Test the new agent¶

In [ ]:

# wrapped agent can act as context manager
with tru_agent as recording:
    for prompt in prompts:
        agent(prompt)

# wrapped agent can act as context manager with tru_agent as recording: for prompt in prompts: agent(prompt)

Explore in a Dashboard¶

In [ ]:

from trulens.dashboard import run_dashboard

run_dashboard(session)  # open a local streamlit app to explore

# session.stop_dashboard(session) # stop if needed

from trulens.dashboard import run_dashboard run_dashboard(session) # open a local streamlit app to explore # session.stop_dashboard(session) # stop if needed