梅竹黑客松2023範例專案
from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
import datetime
import urllib


class GoogleCalendarGeneratorInput(BaseModel):
    """Input for Google Calendar Generator."""

    dates: str = Field(
        ...,
        description=f"Datetime symbol if text contained. format should be 'YYYYMMDDTHHMMSS/YYYYMMDDTHHMMSS'. Current time is {datetime.date.today()}")
    title: str = Field(
        ...,
        description="Calendar Title symbol for reserve schedule.")
    description: str = Field(
        ...,
        description="Calendar Summary text symbol for schedule description.")
    location: str = Field(
        ...,
        description="Calendar location symbol for reservation.")


class CalendarTool(BaseTool):
    name = "google_calendar_reservation"
    description = f"""
Generate Google Calendar url from user text first when containing time, date.
"""

    @staticmethod
    def create_gcal_url(
            title='看到這個..請重生',
            date='20230524T180000/20230524T220000',
            location='那邊',
            description=''):
        base_url = "https://www.google.com/calendar/render?action=TEMPLATE"
        event_url = f"{base_url}&text={urllib.parse.quote(title)}&dates={date}&location={urllib.parse.quote(location)}&details={urllib.parse.quote(description)}"
        return event_url+"&openExternalBrowser=1"

    def _run(self, dates: str, title: str, description: str, location: str):
        print('Google Calendar')
        print('時間：'+dates)
        print('標題：'+title)
        print('描述：'+description)
        print('地點：'+location)
        result = self.create_gcal_url(title, dates, location, description)

        return result

    args_schema: Optional[Type[BaseModel]] = GoogleCalendarGeneratorInput
g_cal_url.py
louis70109/linebot-find-some
from http.client import HTTPException
import logging
import os
import sys
from g_cal_url import CalendarTool

if os.getenv('API_ENV') != 'production':
    from dotenv import load_dotenv

    load_dotenv()

import uvicorn
from fastapi import FastAPI, Request
from linebot.v3.webhook import WebhookParser
from linebot.v3.messaging import (
    AsyncApiClient,
    AsyncMessagingApi,
    Configuration,
    ReplyMessageRequest,
    PushMessageRequest,
    TextMessage
)
from linebot.v3.exceptions import (
    InvalidSignatureError
)
from linebot.v3.webhooks import (
    MessageEvent,
    TextMessageContent
)

from stock_peformace import StockPercentageChangeTool, StockGetBestPerformingTool
from stock_price import StockPriceTool
from langchain.schema import HumanMessage
from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
from langchain.chat_models import ChatOpenAI
from wikipedia import WikiTool
from youtube_restaurant import FindYoutubeVideoTool

logging.basicConfig(level=os.getenv('LOG', 'WARNING'))
logger = logging.getLogger(__file__)


# get channel_secret and channel_access_token from your environment variable
channel_secret = os.getenv('LINE_CHANNEL_SECRET', None)
channel_access_token = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', None)
if channel_secret is None:
    print('Specify LINE_CHANNEL_SECRET as environment variable.')
    sys.exit(1)
if channel_access_token is None:
    print('Specify LINE_CHANNEL_ACCESS_TOKEN as environment variable.')
    sys.exit(1)

configuration = Configuration(
    access_token=channel_access_token
)

app = FastAPI()
async_api_client = AsyncApiClient(configuration)
line_bot_api = AsyncMessagingApi(async_api_client)
parser = WebhookParser(channel_secret)


# Langchain (you must use 0613 model to use OpenAI functions.)
model = ChatOpenAI(model="gpt-3.5-turbo-0613")
tools = [
    StockPriceTool(), StockPercentageChangeTool(), StockGetBestPerformingTool(),
    CalendarTool(),
    FindYoutubeVideoTool(),
    WikiTool()
]
open_ai_agent = initialize_agent(
    tools,
    model,
    agent=AgentType.OPENAI_FUNCTIONS,
    verbose=False)


@app.post("/webhooks/line")
async def handle_callback(request: Request):
    signature = request.headers['X-Line-Signature']

    # get request body as text
    body = await request.body()
    body = body.decode()

    try:
        events = parser.parse(body, signature)
    except InvalidSignatureError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    for event in events:
        print(event)
        if not isinstance(event, MessageEvent):
            continue
        if not isinstance(event.message, TextMessageContent):
            continue

        await line_bot_api.reply_message(
            ReplyMessageRequest(
                reply_token=event.reply_token,
                messages=[TextMessage(text='請稍候...查詢中')]
            )
        )

        tool_result = open_ai_agent.run(event.message.text)

        await line_bot_api.push_message(push_message_request=PushMessageRequest(
            to=event.source.user_id,
            messages=[TextMessage(
                text=tool_result,
                quoteToken=event.message.quote_token)],
        ))
    return 'OK'


if __name__ == "__main__":
    port = int(os.environ.get('PORT', default=8080))
    debug = True if os.environ.get(
        'API_ENV', default='develop') == 'develop' else False
    logging.info('Application will start...')
    uvicorn.run("main:app", host="0.0.0.0", port=port, reload=debug)

main.py
louis70109/linebot-find-some
from typing import List
from langchain.tools import BaseTool
from langchain.agents import AgentType
from typing import Optional, Type
from pydantic import BaseModel, Field
from yf_tool import get_stock_price, calculate_performance, get_price_change_percent, get_best_performing


class StockChangePercentageCheckInput(BaseModel):
    """Input for Stock ticker check. for percentage check"""

    stockticker: str = Field(...,
                             description="Ticker symbol for stock or index")
    days_ago: int = Field(..., description="Int number of days to look back")


class StockPercentageChangeTool(BaseTool):
    name = "get_price_change_percent"
    description = "Useful for when you need to find out the percentage change in a stock's value. You should input the stock ticker used on the yfinance API and also input the number of days to check the change over"

    def _run(self, stockticker: str, days_ago: int):
        price_change_response = get_price_change_percent(stockticker, days_ago)

        return price_change_response

    def _arun(self, stockticker: str, days_ago: int):
        raise NotImplementedError("This tool does not support async")

    args_schema: Optional[Type[BaseModel]] = StockChangePercentageCheckInput


# the best performing

class StockBestPerformingInput(BaseModel):
    """Input for Stock ticker check. for percentage check"""

    stocktickers: List[str] = Field(...,
                                    description="Ticker symbols for stocks or indices")
    days_ago: int = Field(..., description="Int number of days to look back")


class StockGetBestPerformingTool(BaseTool):
    name = "get_best_performing"
    description = "Useful for when you need to the performance of multiple stocks over a period. You should input a list of stock tickers used on the yfinance API and also input the number of days to check the change over"

    def _run(self, stocktickers: List[str], days_ago: int):
        price_change_response = get_best_performing(stocktickers, days_ago)

        return price_change_response

    def _arun(self, stockticker: List[str], days_ago: int):
        raise NotImplementedError("This tool does not support async")

    args_schema: Optional[Type[BaseModel]] = StockBestPerformingInput

stock_peformace.py
louis70109/linebot-find-some
from langchain.tools import BaseTool
from langchain.agents import AgentType
from typing import Optional, Type
from pydantic import BaseModel, Field
from yf_tool import get_stock_price, calculate_performance, get_price_change_percent, get_best_performing


class StockPriceCheckInput(BaseModel):
    """Input for Stock price check."""

    stockticker: str = Field(...,
                             description="Ticker symbol for stock or index")


class StockPriceTool(BaseTool):
    name = "get_stock_ticker_price"
    description = "Useful for when you need to find out the price of stock. You should input the stock ticker used on the yfinance API"

    def _run(self, stockticker: str):
        # print("i'm running")
        price_response = get_stock_price(stockticker)

        return price_response

    def _arun(self, stockticker: str):
        raise NotImplementedError("This tool does not support async")

    args_schema: Optional[Type[BaseModel]] = StockPriceCheckInput

stock_price.py
louis70109/linebot-find-some
from langchain.tools import BaseTool
from langchain.agents import AgentType
from typing import Optional, Type
from pydantic import BaseModel, Field

import yfinance as yf


def get_stock_price(symbol):
    ticker = yf.Ticker(symbol)
    todays_data = ticker.history(period='1d')
    return round(todays_data['Close'][0], 2)


class StockPriceCheckInput(BaseModel):
    """Input for Stock price check."""

    stockticker: str = Field(...,
                             description="Ticker symbol for stock or index")


class StockPriceTool(BaseTool):
    name = "get_stock_ticker_price"
    description = "Useful for when you need to find out the price of stock. You should input the stock ticker used on the yfinance API"

    def _run(self, stockticker: str):
        # print("i'm running")
        price_response = get_stock_price(stockticker)

        return price_response

    def _arun(self, stockticker: str):
        raise NotImplementedError("This tool does not support async")

    args_schema: Optional[Type[BaseModel]] = StockPriceCheckInput

stock_tool.py
louis70109/linebot-find-some
from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
import datetime
import urllib
from langchain.tools import WikipediaQueryRun
from langchain.utilities import WikipediaAPIWrapper

class WikiInput(BaseModel):
    """Input for Google Calendar Generator."""

    title: str = Field(
        ...,
        description="Wikipedia Title symbol ")
    link: str = Field(
        ...,
        description="Wikipedia url")


class WikiTool(BaseTool):
    name = "find_wikipedia_information"
    description = "Use wikipedia resources to find unknown information."

    def _run(self, title: str, link: str):
        print("Wiki")
        print('標題：'+title)
        print('描述：'+link)

        return title, link

    args_schema: Optional[Type[BaseModel]] = WikiInput
wikipedia.py
louis70109/linebot-find-some
from datetime import datetime, timedelta
import yfinance as yf


def calculate_performance(symbol, days_ago):
    ticker = yf.Ticker(symbol)
    end_date = datetime.now()
    start_date = end_date - timedelta(days=days_ago)
    start_date = start_date.strftime('%Y-%m-%d')
    end_date = end_date.strftime('%Y-%m-%d')
    historical_data = ticker.history(start=start_date, end=end_date)
    old_price = historical_data['Close'].iloc[0]
    new_price = historical_data['Close'].iloc[-1]
    percent_change = ((new_price - old_price) / old_price) * 100
    return round(percent_change, 2)


def get_best_performing(stocks, days_ago):
    best_stock = None
    best_performance = None
    for stock in stocks:
        try:
            performance = calculate_performance(stock, days_ago)
            if best_performance is None or performance > best_performance:
                best_stock = stock
                best_performance = performance
        except Exception as e:
            print(f"Could not calculate performance for {stock}: {e}")
    return best_stock, best_performance


def get_stock_price(symbol):
    ticker = yf.Ticker(symbol)
    todays_data = ticker.history(period='1d')
    return round(todays_data['Close'][0], 2)


def get_price_change_percent(symbol, days_ago):
    ticker = yf.Ticker(symbol)

    # Get today's date
    end_date = datetime.now()

    # Get the date N days ago
    start_date = end_date - timedelta(days=days_ago)

    # Convert dates to string format that yfinance can accept
    start_date = start_date.strftime('%Y-%m-%d')
    end_date = end_date.strftime('%Y-%m-%d')

    # Get the historical data
    historical_data = ticker.history(start=start_date, end=end_date)

    # Get the closing price N days ago and today's closing price
    old_price = historical_data['Close'].iloc[0]
    new_price = historical_data['Close'].iloc[-1]

    # Calculate the percentage change
    percent_change = ((new_price - old_price) / old_price) * 100

    return round(percent_change, 2)

yf_tool.py
louis70109/linebot-find-some
from ast import literal_eval
from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
from langchain.tools import YouTubeSearchTool


class YoutubeDefineInput(BaseModel):
    """Youtube Restaurant recommendation."""

    title: str = Field(
        ...,
        description="Restaurant title which will be recommend to user.")


class FindYoutubeVideoTool(BaseTool):
    name = "find_restaurant_youtube"
    description = "Find recommendation restaurant from Youtube"

    def _run(self, title: str):
        print("Youtube")
        print('標題：'+title)
        tool = YouTubeSearchTool()
        youtube_str = tool.run(title)  # force change str to list
        youtube_list = literal_eval(youtube_str)
        for i in range(len(youtube_list)):
            youtube_list[i] = youtube_list[i]
        return youtube_list

    args_schema: Optional[Type[BaseModel]] = YoutubeDefineInput
youtube_restaurant.py
louis70109/linebot-find-some