commit

app.py

@@ -1,261 +1,142 @@
 import os
-import re
-from dataclasses import dataclass
-from typing import Dict, Iterable, List, Optional, Tuple
+from typing import Any, Dict, List, Tuple
 
 import gradio as gr
-import requests
+from openai import OpenAI
 
 
-class ToolCallError(RuntimeError):
-    """リトライを行ってもツール呼び出しが失敗した場合に発生する例外。"""
-
-
-@dataclass
-class StorageHit:
-    """storage_search が返すヒット1件分の情報を保持する構造体。"""
-
-    file_id: Optional[str]
-    file_name: Optional[str]
-    chunk_id: Optional[str]
-    score: Optional[float]
-    snippet: Optional[str]
-
-
-VECTOR_STORE_ID_RE = re.compile(r"vs_[a-zA-Z0-9]{8,}")
-
-
-class OpenAIStorageClient:
-    """OpenAI Storage の検索・取得APIを呼び出す最小限のクライアント。"""
-
-    def __init__(self, api_key: str, base_url: Optional[str] = None, timeout: float = 15.0):
-        """APIキーと各種設定を初期化する。"""
-        if not api_key:
-            raise ValueError("api_key is required")
-        self.api_key = api_key
-        base = base_url or os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1")
-        self.base_url = base.rstrip("/")
-        self.timeout = timeout
-
-    def _post(self, path: str, payload: Dict) -> Dict:
-        """与えられたパスにPOSTし、JSONレスポンスを辞書で返す。"""
-        url = f"{self.base_url}{path}"
-        headers = {
-            "Authorization": f"Bearer {self.api_key}",
-            "Content-Type": "application/json",
-        }
-        try:
-            response = requests.post(url, json=payload, headers=headers, timeout=self.timeout)
-        except requests.RequestException as exc:
-            raise ToolCallError(str(exc)) from exc
-
-        if response.status_code >= 400:
-            raise ToolCallError(f"Tool call failed with status {response.status_code}: {response.text}")
-
-        try:
-            return response.json()
-        except ValueError as exc:
-            raise ToolCallError("Tool call returned non-JSON response") from exc
-
-    def storage_search(
-        self,
-        query: str,
-        top_k: int = 5,
-        filters: Optional[Dict] = None,
-        retries: int = 1,
-    ) -> List[StorageHit]:
-        """storage_search エンドポイントを呼び出してヒットを整形する。"""
-        payload: Dict[str, object] = {"query": query, "top_k": top_k}
-        if filters:
-            payload["filters"] = filters
-
-        attempts = 0
-        while True:
-            try:
-                raw = self._post("/tools/storage_search", payload)
-                hits = []
-                for item in raw.get("hits", []):
-                    hits.append(
-                        StorageHit(
-                            file_id=item.get("file_id"),
-                            file_name=item.get("file_name"),
-                            chunk_id=item.get("chunk_id"),
-                            score=item.get("score"),
-                            snippet=item.get("snippet"),
-                        )
-                    )
-                return hits
-            except ToolCallError:
-                attempts += 1
-                if attempts > retries:
-                    raise
-
-    def storage_get(
-        self,
-        *,
-        chunk_id: Optional[str] = None,
-        file_id: Optional[str] = None,
-        retries: int = 1,
-    ) -> Optional[str]:
-        """storage_get エンドポイントからチャンクまたはファイルを取得する。"""
-        if not chunk_id and not file_id:
-            raise ValueError("Either chunk_id or file_id must be provided.")
-
-        payload: Dict[str, object] = {}
-        if chunk_id:
-            payload["chunk_id"] = chunk_id
-        if file_id:
-            payload["file_id"] = file_id
-
-        attempts = 0
-        while True:
-            try:
-                raw = self._post("/tools/storage_get", payload)
-                content = raw.get("content")
-                if isinstance(content, str):
-                    return content
-                return None
-            except ToolCallError:
-                attempts += 1
-                if attempts > retries:
-                    raise
-def get_vector_store_ids_from_env() -> List[str]:
-    """環境変数 VECTOR_STORE_ID からベクターストアIDのリストを抽出する。"""
-    raw = os.getenv("VECTOR_STORE_ID", "").strip()
-    if not raw:
-        return []
-
-    ids: List[str] = []
-    for token in re.split(r"[,\s]+", raw):
-        if not token:
-            continue
-        if VECTOR_STORE_ID_RE.fullmatch(token) and token not in ids:
-            ids.append(token)
-    return ids
-
-
-def generate_search_queries(question: str) -> List[str]:
-    """自然言語の質問から最大3件の検索クエリを生成する。"""
-    normalized = question.strip()
-    if not normalized:
-        return []
-
-    queries: List[str] = [normalized]
-
-    ascii_tokens = re.findall(r"[A-Za-z0-9]+", normalized)
-    if ascii_tokens:
-        english_query = " ".join(ascii_tokens)
-        if english_query and english_query.lower() != normalized.lower():
-            queries.append(english_query)
-
-    japanese_clean = re.sub(r"[、。；；，]", " ", normalized)
-    japanese_clean = re.sub(r"\s+", " ", japanese_clean).strip()
-    if japanese_clean and japanese_clean not in queries:
-        queries.append(japanese_clean)
+def get_openai_client() -> OpenAI:
+    """OpenAI API クライアントを環境変数から初期化して返す。"""
+    api_key = os.getenv("OPENAI_API_KEY", "").strip()
+    if not api_key:
+        raise RuntimeError("OPENAI_API_KEY が設定されていません。環境変数を確認してください。")
+
+    base_url = os.getenv("OPENAI_API_BASE", "").strip() or None
+    return OpenAI(api_key=api_key, base_url=base_url)
+
+
+def get_vector_store_id() -> str:
+    """環境変数 VECTOR_STORE_ID から単一のベクターストアIDを取得する。"""
+    value = os.getenv("VECTOR_STORE_ID", "").strip()
+    if not value:
+        raise RuntimeError("VECTOR_STORE_ID が設定されていません。環境変数を確認してください。")
+    return value.split(",")[0].strip()
+
+
+def extract_citations(response_dict: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """Responses API レスポンスから file_search の引用情報を抽出する。"""
+    citations: List[Dict[str, Any]] = []
+    outputs = response_dict.get("output") or []
+
+    for output in outputs:
+        for content in output.get("content", []) or []:
+            annotations = content.get("annotations") or []
+            for annotation in annotations:
+                file_citation = annotation.get("file_citation")
+                if not file_citation:
+                    continue
+                entry = {
+                    "file_id": file_citation.get("file_id"),
+                    "file_name": file_citation.get("file_name"),
+                    "vector_store_id": file_citation.get("vector_store_id"),
+                    "quote": file_citation.get("quote"),
+                }
+                if entry not in citations:
+                    citations.append(entry)
+    return citations
+
+
+def answer_with_file_search(
+    vector_store_id: str,
+    query: str,
+    top_k: int = 5,
+    model: str = "gpt-4o-mini",
+) -> Tuple[str, List[Dict[str, Any]]]:
+    """
+    Responses API + file_search ツールを用いて検索と回答を同時に行い、回答と引用を返す。
+
+    Cookbook の "Integrating search results with LLM in a single API call" を参考に構成。
+    """
+    if not query.strip():
+        return "", []
+
+    client = get_openai_client()
+    system_prompt = (
+        "あなたは取得したドキュメントから根拠が確認できる場合のみ回答するアシスタントです。"
+        "根拠が確認できない場合は「該当データがありません。」と答えてください。"
+    )
 
-    return queries[:3]
+    user_message = f"{query}\n\n検索結果は最大 {top_k} 件まで引用してください。"
+
+    response = client.responses.create(
+        model=model,
+        input=[
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": user_message},
+        ],
+        tools=[
+            {
+                "type": "file_search",
+                "vector_store_ids": [vector_store_id],
+            }
+        ],
+    )
 
+    response_dict = response.to_dict_recursive()  # type: ignore[attr-defined]
+    answer_text = response_dict.get("output_text") or ""
+    citations = extract_citations(response_dict)
+    return answer_text.strip(), citations
 
-def deduplicate_hits(hits_by_query: Iterable[Tuple[str, List[StorageHit]]]) -> List[StorageHit]:
-    """クエリごとのヒットを重複除去しながら結合する。"""
-    seen_keys = set()
-    merged: List[StorageHit] = []
-    for _, hits in hits_by_query:
-        for hit in hits:
-            key = (hit.chunk_id, hit.file_id)
-            if key in seen_keys:
-                continue
-            seen_keys.add(key)
-            merged.append(hit)
-    return merged
 
+def format_answer(answer: str, citations: List[Dict[str, Any]]) -> str:
+    """回答文と引用リストを UI 向けのテキストに整形する。"""
+    if not answer:
+        return "該当データがありません。"
 
-def build_excerpt(text: str, limit: int = 280) -> str:
-    """テキストを指定文字数以内に要約した抜粋を返す。"""
-    collapsed = re.sub(r"\s+", " ", text).strip()
-    if len(collapsed) <= limit:
-        return collapsed
-    return collapsed[: limit - 1] + "…"
+    lines = [answer]
+    if citations:
+        lines.append("")
+        lines.append("引用:")
+        for citation in citations:
+            file_name = citation.get("file_name") or citation.get("file_id") or "unknown"
+            quote = citation.get("quote")
+            if quote:
+                lines.append(f"- {quote} [{file_name}]")
+            else:
+                lines.append(f"- [{file_name}]")
+    return "\n".join(lines)
 
 
 def produce_answer(question: str, top_k: int) -> str:
-    """質問に応じて検索を実行し、仕様に沿った回答テキストを生成する。"""
+    """Gradio UI からの質問を処理し、Responses API の結果を返す。"""
     question = (question or "").strip()
     if not question:
         return "質問が入力されていません。"
 
-    api_key = os.getenv("OPENAI_API_KEY")
-    if not api_key:
-        return "OPENAI_API_KEY が設定されていません。環境変数を確認してください。"
-
-    client = OpenAIStorageClient(api_key=api_key)
-
-    vector_store_ids = get_vector_store_ids_from_env()
-    if not vector_store_ids:
-        return "VECTOR_STORE_ID が設定されていません。環境変数を確認してください。"
-
-    queries = generate_search_queries(question)
-    if not queries:
-        return "該当データがありません。"
-
     try:
-        hits_by_query = []
-        for query in queries:
-            filters = {"vector_store_ids": vector_store_ids}
-            hits = client.storage_search(query=query, top_k=top_k, filters=filters)
-            if hits:
-                hits_by_query.append((query, hits))
-
-        if not hits_by_query:
-            return "該当データがありません。"
-
-        merged_hits = deduplicate_hits(hits_by_query)
-        evidence_lines: List[str] = []
-
-        for hit in merged_hits[:3]:
-            content: Optional[str] = None
-            try:
-                if hit.chunk_id:
-                    content = client.storage_get(chunk_id=hit.chunk_id)
-                if not content and hit.file_id:
-                    content = client.storage_get(file_id=hit.file_id)
-            except ToolCallError:
-                # Propagate after exceeding retry budget according to spec.
-                return "システムエラーが発生しました。時間をおいて再度お試しください。"
-
-            source_text = content or hit.snippet
-            if not source_text:
-                continue
-
-            excerpt = build_excerpt(source_text)
-            if not excerpt:
-                continue
-
-            file_ref = hit.file_name or (hit.file_id or "unknown")
-            chunk_ref = hit.chunk_id or "chunk"
-            evidence_lines.append(f"- {excerpt} [{file_ref}#{chunk_ref}]")
-
-        if not evidence_lines:
-            return "該当データがありません。"
-
-        return "\n".join(evidence_lines)
-
-    except ToolCallError:
+        vector_store_id = get_vector_store_id()
+        answer, citations = answer_with_file_search(vector_store_id, question, top_k=top_k)
+        return format_answer(answer, citations)
+    except RuntimeError as runtime_error:
+        return str(runtime_error)
+    except Exception:
         return "システムエラーが発生しました。時間をおいて再度お試しください。"
 
 
 def respond(question: str, top_k: int = 5) -> str:
-    """Gradio UI から呼び出されるエントリーポイント。"""
+    """Gradio の UI から呼び出されるハンドラ。"""
     try:
-        return produce_answer(question, top_k=int(top_k))
-    except Exception:  # pragma: no cover - defensive fallback for UI
+        return produce_answer(question, int(top_k))
+    except Exception:
         return "システムエラーが発生しました。時間をおいて再度お試しください。"
 
 
 with gr.Blocks() as demo:
     gr.Markdown(
         """
-        ## MCP Storage
+        ## MCP Storage レスポンダ
+        OpenAI Responses API の file_search ツールを利用し、指定した Vector Store から根拠付きで回答します。
+        OPENAI_API_KEY と VECTOR_STORE_ID を環境変数に設定してからご利用ください。
         """
     )