rag-path

# Example

```python [10-14|16-17|19-26|28-36]
import subprocess
from pydantic import BaseModel
from openai import OpenAI
from dotenv import find_dotenv, load_dotenv
from markdownify import markdownify as md
import pandas as pd

load_dotenv(find_dotenv())

#  先用 singlefile 抓下網頁
html_content = subprocess.run(
    ["docker", "run", "singlefile", 
    "https://espn.com/soccer/commentary/_/gameId/704489"], 
    capture_output=True)

# 將網頁變成 Markdown
content = md(html_content.stdout.decode(), strip=["a", "img"])

# 定義結構
class Commentary(BaseModel):
    time: str
    text: str

class Match(BaseModel):
    commentaries: list[Commentary]

# 取得結果
completion = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "Extract commentaries from the given passage. Do not fabricate any information."},
        {"role": "user", "content": f"Passage:{content}"},
    ],
    response_format=Match, #依照 class 結構擷取資料
)

event = completion.choices[0].message.parsed

commentaries = pd.DataFrame(event.dict()['commentaries'])

```

### 擷取 youtube 字幕

```python [6|32-36|92-105|111-126]
from urllib.parse import parse_qs
from urllib.parse import urlparse
from openai import OpenAI
from yt_dlp import YoutubeDL
import timeout_decorator
from youtube_transcript_api import YouTubeTranscriptApi
from dotenv import find_dotenv, load_dotenv

DEFAULT_LANGUAGES = ["zh-TW", "zh-Hant", "zh", "zh-Hans", "ja", "en", "ko"]

ALLOWED_NETLOCS = {
    "youtu.be",
    "m.youtube.com",
    "youtube.com",
    "www.youtube.com",
    "www.youtube-nocookie.com",
    "vid.plus",
}

class YoutubeLoader:
    def __init__(self, languages: list[str] | None = None) -> None:
        self.languages = languages or DEFAULT_LANGUAGES

@timeout_decorator.timeout(20)
    def load(self, url: str) -> str:
 
        video_id = parse_video_id(url)

# 擷取字幕
        transcript_pieces: list[dict[str, str | float]] = (
            YouTubeTranscriptApi().get_transcript(
                video_id, self.languages)
        )

lines = []
        for transcript_piece in transcript_pieces:
            text = str(transcript_piece.get("text", "")).strip()
            if text:
                lines.append(text)
        return "\n".join(lines)

def parse_video_id(url: str) -> str:

parsed_url = urlparse(url)

if parsed_url.scheme not in {"http", "https"}:
        raise f"unsupported URL scheme: {parsed_url.scheme}"

if parsed_url.netloc not in ALLOWED_NETLOCS:
        raise f"unsupported URL netloc: {parsed_url.netloc}"

path = parsed_url.path

if path.endswith("/watch"):
        query = parsed_url.query
        parsed_query = parse_qs(query)
        if "v" in parsed_query:
            ids = parsed_query["v"]
            video_id = ids if isinstance(ids, str) else ids[0]
        else:
            raise f"no video found in URL: {url}"
    else:
        path = parsed_url.path.lstrip("/")
        video_id = path.split("/")[-1]

if len(video_id) != 11:  # Video IDs are 11 characters long
        raise f"invalid video ID: {video_id}"

return video_id

def get_youtube_videos(channel_url):

ydl_opts = {
        "extract_flat": True,
        "quiet": True,
    }
    with YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(channel_url, download=False)

video_list = [
        {"title": entry["title"], "url": entry["url"]}
        for entry in info.get("entries", [])
    ]
    return video_list

# 定義食譜的內容
class Ingredient(BaseModel):
    name: str
    quantity: str
    unit: str
    preparation: str

class Step(BaseModel):
    description: str

class Recipe(BaseModel):
    title: str
    ingredients: list[Ingredient]
    steps: Step

def main():

load_dotenv(find_dotenv())

client = OpenAI()
    yt_loader = YoutubeLoader()
    url = "https://youtu.be/smD_6Ranb4g"
    content = yt_loader.load(url)
    prompt = """
    從字幕中抽取食譜資訊，不要捏造任何資訊。
    抽取後請務必將所有內容翻譯成台灣繁體中文。"""
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini-2024-07-18",
        messages=[
            {"role": "system", "content": prompt},
            {"role": "user", "content": f"字幕：{content}"},
        ],
        response_format=Recipe,
    )
    markdown_recipe = completion.choices[0].message.content

if __name__ == "__main__":

main()

```

<pre style="white-space:pre;overflow-x:auto;line-height:normal;font-family:Menlo,'DejaVu Sans Mono',consolas,'Courier New',monospace">Recipe(
 title='和風鰤魚＋大根',
 ingredients=[
 Ingredient(name='鰤魚粗', quantity='適量', unit='', preparation='清洗並去除腥味'),
 Ingredient(name='大根', quantity='適量', unit='', preparation='去皮並切成半月形'),
 Ingredient(name='長蔥', quantity='適量', unit='', preparation='切成白髮蔥與青蔥部分分開備用'),
 Ingredient(name='生薑', quantity='適量', unit='', preparation='清洗並切成薄片與細絲'),
 Ingredient(name='清酒', quantity='大量', unit='', preparation=''),
 Ingredient(name='水', quantity='適量', unit='', preparation=''),
 Ingredient(name='中皿糖', quantity='適量', unit='', preparation='主要用於去腥'),
 Ingredient(name='醬油', quantity='適量', unit='', preparation='')
 ],
 steps=Step(
 description='1. 大根去皮並切成約2-3cm的半月切，備用。\n2. 
清洗鰤魚粗，去除表面血水，並以熱水燙過表面再立即用清水沖洗，去除腥味。\n3. 生薑切成兩種形式：薄片與細絲，備用。\n4.
長蔥切成白髮蔥與青蔥部分分開備用，青蔥部分稍微壓碎以增香。\n5. 用寬型的鍋或平底鍋排放大根，加入清酒與水煮沸。\n6. 
將火調小，加入鰤魚粗，以及長蔥青蔥部分和生薑薄片同煮，以去腥。\n7. 加入中皿糖後繼續煮，去除浮沫。\n8. 
當煮沸後加入醬油，不蓋鍋蓋煮至水分減少，並不時將湯汁澆在食材上。\n9. 味道喜好程度煮至濃稠度適中，盛盤上桌。\n10. 
最後加入白髮蔥、生薑絲，喜好者可加入芥末增味，完成美味鰤大根。'
 )
)
</pre>

### `streamlit sample code`

```python [1-3|5-9|11-18|20-39]
# set select box for dataset
dataset_name = st.sidebar.selectbox(
    "請選擇要查詢的資料集名稱", options=datasets)

# session state
if "chat_history" not in st.session_state:
    st.session_state.chat_history = [
        AIMessage(content="Hello, I am a bot. How can I help you?"),
    ]

# conversation
for message in st.session_state.chat_history:
    if isinstance(message, AIMessage):
        with st.chat_message("AI"):
            st.write(message.content)
    elif isinstance(message, HumanMessage):
        with st.chat_message("Human"):
            st.write(message.content)

# user input
user_query = st.chat_input("Type your message here...")
if user_query is not None and user_query != "":
    st.session_state.chat_history.append(
        HumanMessage(content=user_query))

with st.chat_message("Human"):
        st.markdown(user_query)
    with st.chat_message("AI"):
        response = st.write_stream(
            get_response(
                user_query=user_query,
                chat_history=st.session_state.chat_history[-10:],
                collection_name="bootcamp",
                dataset_name=dataset_name,
            )
        )

st.session_state.chat_history.append(
        AIMessage(content=response))

```

### 後端回答的機器人

```python [4-17|18-26|27-33]
def get_response(
    user_query, chat_history, collection_name
):
    generator_llm = AzureChatOpenAI()
    system_prompt = (
        "你是一位專門根據文件回答問題的 AI 助手。"
        "如果你無法從文件得到答案，請說你不知道。"
        "請根據以下參考資料回答問題："
        "歷史紀錄：{chat_history}"
        "參考資料：{context}"
    )
    prompt = ChatPromptTemplate.from_messages(
        [("system", system_prompt), ("human", "{input}")]
    )
    question_answer_chain = create_stuff_documents_chain(
        generator_llm, prompt)
    
    embedding_llm = AzureOpenAIEmbeddings()
    client = QdrantClient(url="http://localhost:6333")
    qdrant = QdrantVectorStore(
        client=client, 
        collection_name=collection_name, 
        embedding=embedding_llm
    )
    retriever = qdrant.as_retriever(
        search_kwargs={"k": 3})

rag_chain = create_retrieval_chain(
        retriever, question_answer_chain)
    chain = rag_chain.pick("answer")
    return chain.stream({
        "input": user_query, 
        "chat_history": chat_history})

```