生成AIで自社データを学習・連携する方法！社内データを活用した事例も紹介

streamlit==1.31.0pdfplumber==0.10.3scikit-learn==1.3.2openai==1.14.0numpy==1.26.3

import streamlit as stimport pdfplumberimport reimport osimport numpy as npfrom sklearn.feature_extraction.text import TfidfVectorizerfrom sklearn.metrics.pairwise import cosine_similarityimport openaiimport tempfile
st.set_page_config(    page_title="PDF RAG アシスタント",    page_icon="📚",    layout="wide")
st.title("PDF RAG アシスタント")st.markdown("PDFをアップロードして、内容に関する質問をしてください。")
if 'pdfs' not in st.session_state:    st.session_state.pdfs = {}  # {filename: content}if 'vectorizer' not in st.session_state:    st.session_state.vectorizer = Noneif 'doc_vectors' not in st.session_state:    st.session_state.doc_vectors = Noneif 'is_index_built' not in st.session_state:    st.session_state.is_index_built = Falseif 'chat_history' not in st.session_state:    st.session_state.chat_history = []
openai_api_key = st.sidebar.text_input("OpenAI APIキー", type="password")if openai_api_key:    openai.api_key = openai_api_key
def clean_text(text):    text = re.sub(r'\n+', ' ', text)    text = re.sub(r'\s{2,}', ' ', text)    text = re.sub(r'\s+([.,;?!])', r'\1', text)    return text.strip()
def extract_text_from_pdf(pdf_file):    try:        with pdfplumber.open(pdf_file) as pdf:            text = ""            for page in pdf.pages:                try:                    extracted = page.extract_text()                    if extracted:                        text += extracted + "\n"                except Exception as page_error:                    st.warning(f"ページの処理中にエラーが発生しましたが、続行します: {str(page_error)}")                    continue            if not text.strip():                st.warning("PDFからテキストを抽出できませんでした。")                return None                            return clean_text(text)    except Exception as e:        st.error(f"PDFからのテキスト抽出中にエラーが発生しました: {str(e)}")        return None
def build_search_index():    try:        # ドキュメントとファイル名の抽出        documents = list(st.session_state.pdfs.values())        filenames = list(st.session_state.pdfs.keys())                if not documents:            st.warning("インデックスを構築するには、少なくとも1つのPDFファイルをアップロードしてください。")            return False                with st.spinner("検索インデックスを構築中..."):            vectorizer = TfidfVectorizer(                min_df=1,                max_df=1.0,                ngram_range=(1, 1),                sublinear_tf=True            )            doc_vectors = vectorizer.fit_transform(documents)
            st.session_state.vectorizer = vectorizer            st.session_state.doc_vectors = doc_vectors            st.session_state.documents = documents            st.session_state.filenames = filenames                        st.session_state.is_index_built = True            return True    except Exception as e:        st.error(f"検索インデックスの構築中にエラーが発生しました: {str(e)}")        return False
def search_documents(query, top_k=3):    try:        if not st.session_state.is_index_built:            st.warning("先に検索インデックスを構築してください。")            return []        if len(st.session_state.documents) == 1:            return [{                "title": st.session_state.filenames[0],                "content": st.session_state.documents[0],                "score": 1.0            }]                    # クエリをベクトル化        try:            query_vector = st.session_state.vectorizer.transform([query])
            similarity_scores = cosine_similarity(query_vector, st.session_state.doc_vectors).flatten()
            top_indices = similarity_scores.argsort()[-top_k:][::-1]                        results = []            for i in top_indices:                if similarity_scores[i] > 0:  # 類似度が0より大きい場合のみ含める                    results.append({                        "title": st.session_state.filenames[i],                        "content": st.session_state.documents[i],                        "score": float(similarity_scores[i])                    })            if not results and len(st.session_state.documents) > 0:                results.append({                    "title": st.session_state.filenames[0],                    "content": st.session_state.documents[0],                    "score": 0.1  # 暫定的なスコア                })                        return results        except Exception as e:            st.warning(f"検索中に問題が発生しました（フォールバック使用）: {str(e)}")            if len(st.session_state.documents) > 0:                return [{                    "title": st.session_state.filenames[0],                    "content": st.session_state.documents[0],                    "score": 0.1                 }]            return []                except Exception as e:        st.error(f"ドキュメント検索中にエラーが発生しました: {str(e)}")        return []
def generate_answer(results, query):    if not openai.api_key:        st.error("回答を生成するには、サイドバーにOpenAI APIキーを入力してください。")        return None            try:        context = "\n\n".join([            f"ドキュメント: {r['title']}\n内容: {r['content'][:1500]}"             for r in results        ])                system_prompt = """あなたは提供された文書情報のみに基づいて、正確で役立つ回答を提供するインテリジェントなアシスタントです。        提供された文書に情報が見つからない場合は、その制限を認め、情報を作り上げることは避けてください。日本語で回答してください。"""                user_prompt = f"質問: {query}\n\n以下の文書に基づいて、明確で正確な回答を提供してください:\n\n{context}"
        # OpenAI APIの新しい呼び出し方法（OpenAI 1.0.0以降に対応）        response = openai.chat.completions.create(            model="gpt-3.5-turbo",             messages=[                {"role": "system", "content": system_prompt},                {"role": "user", "content": user_prompt}            ],            max_tokens=500,            temperature=0.5        )                return response.choices[0].message.content.strip()    except Exception as e:        st.error(f"回答生成中にエラーが発生しました: {str(e)}")        # エラーの詳細なログ出力        import traceback        st.error(f"詳細なエラー: {traceback.format_exc()}")        return None
with st.sidebar:    st.header("PDFのアップロード")    uploaded_files = st.file_uploader("PDFファイルを選択", accept_multiple_files=True, type=['pdf'])
    if uploaded_files:        new_files = False        for uploaded_file in uploaded_files:            if uploaded_file.name not in st.session_state.pdfs:                new_files = True                with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp_file:                    tmp_file.write(uploaded_file.getvalue())                    tmp_path = tmp_file.name                                with st.spinner(f"{uploaded_file.name}を処理中..."):                    pdf_text = extract_text_from_pdf(tmp_path)                    os.unlink(tmp_path)                                         if pdf_text:                        st.session_state.pdfs[uploaded_file.name] = pdf_text                        st.success(f"正常に処理されました: {uploaded_file.name}")                    else:                        st.error(f"{uploaded_file.name}の処理に失敗しました")
        if new_files:            st.session_state.is_index_built = False    if st.session_state.pdfs:        st.subheader("アップロードされたPDF")        for filename in st.session_state.pdfs.keys():            st.write(f"- {filename}")                if not st.session_state.is_index_built:            if st.button("検索インデックスを構築"):                success = build_search_index()                if success:                    st.success("検索インデックスが正常に構築されました！")        else:            st.success("検索インデックスが構築済みです")            if st.button("インデックスを再構築"):                success = build_search_index()                if success:                    st.success("検索インデックスが再構築されました！")
st.header("PDFに関する質問")
for i, (q, a) in enumerate(st.session_state.chat_history):    st.markdown(f"**質問 {i+1}:** {q}")    st.markdown(f"**回答:** {a}")    st.markdown("---")
query = st.text_input("質問を入力してください")
if query and st.button("回答を取得"):    if not st.session_state.is_index_built:        st.warning("質問に回答する前に、PDFをアップロードして検索インデックスを構築してください。")    else:        with st.spinner("関連ドキュメントを検索中..."):            results = search_documents(query)                if results:            with st.spinner("回答を生成中..."):                answer = generate_answer(results, query)                                if answer:                    st.markdown("### 回答:")                    st.write(answer)
                    st.session_state.chat_history.append((query, answer))
                    with st.expander("参照された文書"):                        for i, result in enumerate(results):                            st.subheader(f"文書 {i+1}: {result['title']}")                            st.text(f"関連度スコア: {result['score']:.4f}")                            st.text(result['content'][:500] + "..." if len(result['content']) > 500 else result['content'])                else:                    st.error("回答を生成できませんでした。")        else:            st.warning("質問に関連する文書が見つかりませんでした。")
if st.session_state.chat_history and st.button("チャット履歴をクリア"):    st.session_state.chat_history = []    st.experimental_rerun()