Google AI studio ✕ chat bot

import streamlit as st 
import os 
from dotenv import load_dotenv 
import nest_asyncio 

from langchain_community.document_loaders import PyPDFLoader, UnstructuredHTMLLoader 
from langchain.text_splitter import RecursiveCharacterTextSplitter 
from langchain_google_genai import GoogleGenerativeAIEmbeddings, ChatGoogleGenerativeAI 
from langchain_community.vectorstores import FAISS 
from langchain.chains import RetrievalQA 

# Streamlit上でasyncioエラー回避 
nest_asyncio.apply() 

# .envからAPIキー読み込み 
load_dotenv() 
gemini_key = os.getenv("GOOGLE_API_KEY") 
if gemini_key is None: 
st.error(".envに GOOGLE_API_KEY が見つかりません") 
st.stop() 
os.environ["GOOGLE_API_KEY"] = gemini_key 

st.title("📄 PDF / HTML Q&A チャットボット") 

# セッション状態の初期化 
if "messages" not in st.session_state: 
st.session_state.messages = [] 
if "qa_chain" not in st.session_state: 
st.session_state.qa_chain = None 

# 既存メッセージの表示 
for message in st.session_state.messages: 
with st.chat_message(message["role"]): 
st.markdown(message["content"]) 

# ファイルアップロード 
uploaded_file = st.file_uploader( 
"Q&Aを行いたいPDFまたはHTMLファイルをアップロードしてください", 
type=["pdf", "html", "htm"] 
) 

def setup_qa_chain(file_path: str, file_type: str): 
# PDF or HTML の読み込み 
if file_type == "pdf": 
loader = PyPDFLoader(file_path) 
elif file_type in ["html", "htm"]: 
loader = UnstructuredHTMLLoader(file_path) 
else: 
raise ValueError("対応していないファイル形式です") 

documents = loader.load() 
if not documents: 
st.error(f"{file_type.upper()}からテキストを抽出できませんでした。") 
st.stop() 

# テキスト分割 
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) 
texts = text_splitter.split_documents(documents) 

# 埋め込みとベクトルストア 
embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001") 
vector_store = FAISS.from_documents(texts, embeddings) 
retriever = vector_store.as_retriever() 

# GeminiモデルとQAチェーン 
model = ChatGoogleGenerativeAI(model="gemini-1.5-flash") 
qa_chain = RetrievalQA.from_chain_type( 
llm=model, retriever=retriever, return_source_documents=True 
) 
return qa_chain 

# ファイルがアップロードされ、QAチェーンが未作成の場合 
if uploaded_file and st.session_state.qa_chain is None: 
with st.spinner("ドキュメントを読み込み中..."): 
try: 
temp_file_path = f"temp_uploaded.{uploaded_file.name.split('.')[-1]}" 
with open(temp_file_path, "wb") as f: 
f.write(uploaded_file.getbuffer()) 

file_type = uploaded_file.name.split(".")[-1].lower() 
st.session_state.qa_chain = setup_qa_chain(temp_file_path, file_type) 
st.success("ドキュメント準備完了！質問を入力してください。") 
os.remove(temp_file_path) 
except Exception as e: 
st.error(f"エラーが発生しました: {e}") 

# QAチェーンが存在する場合にチャット表示 
if st.session_state.qa_chain: 
if prompt := st.chat_input("ドキュメントの内容について質問してください"): 
st.session_state.messages.append({"role": "user", "content": prompt}) 
with st.chat_message("user"): 
st.markdown(prompt) 

with st.spinner("回答を生成中..."): 
try: 
response = st.session_state.qa_chain.invoke({"query": prompt}) 
answer = response['result'] 

st.session_state.messages.append({"role": "assistant", "content": answer}) 
with st.chat_message("assistant"): 
st.markdown(answer) 
except Exception as e: 
st.error(f"質問処理中にエラーが発生しました: {e}")

みたいなのと、.envファイル

GOOGLE_API_KEY = HOGEHOGE;

brew install tesseract 
pip install nest_asyncio 
pip install langchain langchain-community pypdf 
pip install pytesseract pillow

# src/scripts/scrape_site.py 
import requests 
from bs4 import BeautifulSoup 
import json 
import os 
import warnings 
from urllib3.exceptions import NotOpenSSLWarning 
from urllib.parse import urljoin, urlparse 

warnings.filterwarnings("ignore", category=NotOpenSSLWarning) 

BASE_URL = "https://www.HOGEFUGA.co.jp/" 
#引っ張ってくるサイトのurl
OUTPUT_DIR = "scraped_site" 
VISITED = set() 

def fetch_text(url): 
"""1ページのテキストを抽出""" 
try: 
headers = { 
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " 
"AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/114.0.0.0 Safari/537.36" } 
resp = requests.get(url, headers=headers) 
resp.raise_for_status() 
soup = BeautifulSoup(resp.text, 'html.parser') 
text_content = "" 
for tag in soup.find_all(['p', 'h1', 'h2', 'h3', 'li', 'span']): 
cleaned = tag.get_text(separator=' ', strip=True) 
if cleaned: 
text_content += cleaned + " " 
return text_content.strip() 
except requests.exceptions.RequestException: 
return None 

def save_text(url, text): 
"""URL に対応したディレクトリ/JSON に保存""" 
parsed = urlparse(url) 
path = parsed.path.strip("/") 

# index ページなら "index.json" に 
if path == "": 
path = "index" 

dir_path = os.path.join(OUTPUT_DIR, os.path.dirname(path)) 
os.makedirs(dir_path, exist_ok=True) 
file_path = os.path.join(OUTPUT_DIR, path + ".json") 

with open(file_path, "w", encoding="utf-8") as f: 
json.dump({"url": url, "text": text}, f, ensure_ascii=False, indent=2) 
print(f"保存: {file_path}") 

def crawl(url, max_pages=50): 
"""再帰的にサイト内リンクをクロール""" 
if url in VISITED or len(VISITED) >= max_pages: 
return 
VISITED.add(url) 

text = fetch_text(url) 
if text: 
save_text(url, text) 

# サイト内リンクを取得して再帰 
headers = { 
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " 
"AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/114.0.0.0 Safari/537.36" } 
try: 
resp = requests.get(url, headers=headers) 
soup = BeautifulSoup(resp.text, 'html.parser') 
for a in soup.find_all('a', href=True): 
link = urljoin(BASE_URL, a['href']) 
if link.startswith(BASE_URL): 
crawl(link, max_pages) 
except requests.exceptions.RequestException: 
pass 

if __name__ == "__main__": 
crawl(BASE_URL) 
print(f"クロール終了。{len(VISITED)} ページ取得")

こういう感じのコード書いて、スクレイピングしてもいいのかも。