Upload 2 files

app.py

@@ -0,0 +1,199 @@
+# https://github.com/anthropics/anthropic-cookbook/blob/main/third_party/Brave/web_search_using_brave.ipynb
+import asyncio
+import html
+import json
+import os
+from typing import List
+
+import boto3
+import requests
+import streamlit as st
+from googleapiclient.discovery import build
+
+st.title("Qiitaに聞いた！！")
+
+if "client" not in st.session_state:
+    st.session_state.client = boto3.client("bedrock-runtime")
+client = st.session_state.client
+
+
+# 検索クエリを生成する関数
+def generate_search_queries(question: str) -> List[str]:
+    """
+    Google 検索エンジン用の検索クエリを生成する
+    """
+    GENERATE_QUERIES = """
+        User question: {{question}}
+
+        Format: {"queries": ["query_1", "query_2", "query_3"]}
+        """
+
+    response = client.invoke_model(
+        modelId="anthropic.claude-3-haiku-20240307-v1:0",
+        body=json.dumps(
+            {
+                "anthropic_version": "bedrock-2023-05-31",
+                "max_tokens": 1024,
+                "system": "You are an expert at generating search queries for the Google search engine. Generate two search queries that are relevant to this question in Japanese. Output only valid JSON.",
+                "messages": [
+                    {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "text",
+                                "text": GENERATE_QUERIES.replace(
+                                    "{{question}}", question
+                                ),
+                            }
+                        ],
+                    },
+                ],
+                "temperature": 0,
+            }
+        ),
+    )
+
+    result = json.loads(response.get("body").read())
+    search_queries = result["content"][0]["text"]
+    search_queries = json.loads(search_queries)
+
+    return search_queries
+
+
+# Qiitaを検索する関数
+def search_qiita(search_query: str) -> list:
+    """
+    指定された検索クエリでQiitaを検索する
+    """
+    service = build("customsearch", "v1", developerKey=os.environ.get("GOOGLE_API_KEY"))
+    cse = service.cse()
+    res = cse.list(
+        q=f"{search_query} site:qiita.com",
+        cx=os.environ.get("GOOGLE_CSE_ID"),
+        num=3,
+    ).execute()
+
+    documents = list(
+        map(
+            lambda x: {
+                "title": x["title"],
+                "link": x["link"],
+                "snippet": x["snippet"],
+            },
+            res["items"],
+        )
+    )
+
+    return documents
+
+
+# 検索結果にマークダウンを追加する非同期関数
+async def add_markdown(search_result: dict) -> dict:
+    """
+    検索結果にマークダウンを追加する
+    """
+    url = search_result["link"]
+    response = requests.get(f"{url}.md")
+    markdown = response.text
+    search_result["markdown"] = html.escape(markdown)
+
+    return search_result
+
+
+# 検索結果をXML形式のドキュメントに変換する関数
+def create_xml_documents(documents: list) -> str:
+    """
+    検索結果をXML形式のドキュメントに変換する
+    """
+    xml_documents = ""
+    xml_doc = list(
+        map(
+            lambda x: f'<doc title="{x["title"]}"><link>{x["link"]}</link><markdown>{x["markdown"]}</markdown></doc>',
+            documents,
+        )
+    )
+    xml_documents = f"<documents>{''.join(xml_doc)}</documents>"
+    return xml_documents
+
+
+# 質問に対する回答を生成する関数
+def generate_answer(question: str, documents: dict):
+    """
+    検索結果から質問に対する回答を生成する
+    """
+    xml_docs = create_xml_documents(documents=documents)
+
+    ANSWER_QUESTION = f"""I have provided you with the following search results:
+    {xml_docs}
+
+    Please answer the user's question using only information from the search results. 
+    Keep your answer concise.
+    Answer is olways in Japanese!
+
+    User's question: {question}
+    """
+
+    response = client.invoke_model(
+        modelId="anthropic.claude-3-haiku-20240307-v1:0",
+        body=json.dumps(
+            {
+                "anthropic_version": "bedrock-2023-05-31",
+                "max_tokens": 1024,
+                "messages": [
+                    {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "text",
+                                "text": ANSWER_QUESTION,
+                            }
+                        ],
+                    },
+                ],
+                "temperature": 0.1,
+            }
+        ),
+    )
+
+    result = json.loads(response.get("body").read())
+    return result["content"][0]["text"]
+
+
+# メイン関数
+async def main():
+    with st.form("Form"):
+        question = st.text_input("質問")
+
+        if st.form_submit_button("質問する"):
+            with st.status("処理中...", expanded=True) as status:
+                search_queries = generate_search_queries(question=question)
+                st.write("検索クエリ: " + str(search_queries["queries"]))
+
+                documents = []
+                for search_query in search_queries["queries"]:
+                    search_results = search_qiita(search_query=search_query)
+                    result = await asyncio.gather(
+                        *[add_markdown(x) for x in search_results]
+                    )
+                    documents.extend(result)
+                st.write("検索完了")
+
+                st.write("回答生成中...")
+                answer = generate_answer(question=question, documents=documents)
+
+                status.update(label="complete!", state="complete", expanded=False)
+
+            st.markdown(answer)
+
+            st.divider()
+
+            st.markdown("参照ドキュメント")
+
+            for document in documents:
+                st.markdown(
+                    f'[{document["title"]}]({document["link"]}) by {document["link"].split("/")[3]}'
+                )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

requirements.txt

@@ -0,0 +1,3 @@
+boto3
+streamlit
+google-api-python-client