DeepSeek V4.1 Flash × Search API検索で根拠をつかみ、自分で調べるエージェントをつくる。

思考モードの出力を見る

"""最初の Chat Completions 呼び出し。

思考モードの reasoning_content（思考の過程）と content（最終回答）、
usage（トークン使用量）を表示する。
"""

from __future__ import annotations

import argparse
import sys
from typing import Any

from openai import APIError

from ds_client import MODEL, chat_client, format_chat_error

DEFAULT_QUESTION = "DeepSeek の Search API と Fetch API の違いを 2 文で説明してください。"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="DeepSeek Chat Completions の最初の呼び出し")
    parser.add_argument("question", nargs="?", default=DEFAULT_QUESTION, help="質問文（省略可）")
    parser.add_argument("--no-thinking", action="store_true", help="思考モードを無効にする")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    client = chat_client()

    messages = [
        {"role": "system", "content": "あなたは日本語で簡潔に答えるアシスタントです。"},
        {"role": "user", "content": args.question},
    ]

    # 公式ガイド（Thinking Mode）: OpenAI SDK では thinking を extra_body に入れる。
    # 無効にするときは reasoning_effort を付けない。
    kwargs: dict[str, Any] = {}
    if args.no_thinking:
        kwargs["extra_body"] = {"thinking": {"type": "disabled"}}
    else:
        kwargs["reasoning_effort"] = "high"
        kwargs["extra_body"] = {"thinking": {"type": "enabled"}}

    try:
        response = client.chat.completions.create(model=MODEL, messages=messages, **kwargs)
    except APIError as exc:
        print(format_chat_error(exc), file=sys.stderr)
        raise SystemExit(1)
    message = response.choices[0].message

    reasoning_content = getattr(message, "reasoning_content", None)
    print("[思考]")
    print(reasoning_content or "（思考なし）")
    print()
    print("[回答]")
    print(message.content)
    print()

    usage = response.usage
    print("[usage]")
    print(f"prompt_tokens: {usage.prompt_tokens}")
    print(f"completion_tokens: {usage.completion_tokens}")
    # ディスクキャッシュのヒット/ミス（公式ガイドの KV Cache）。応答に無い場合がある。
    hit = getattr(usage, "prompt_cache_hit_tokens", None)
    miss = getattr(usage, "prompt_cache_miss_tokens", None)
    if hit is not None:
        print(f"prompt_cache_hit_tokens: {hit}")
    if miss is not None:
        print(f"prompt_cache_miss_tokens: {miss}")


if __name__ == "__main__":
    main()

python 01_hello.py
python 01_hello.py --no-thinking

[思考]
「DeepSeek の Search API と Fetch API の違いを 2」について考えます。

[回答]
「DeepSeek の Search API と Fetch API の違いを 2 文で説明してください。」についてお答えします（モック応答）。

[usage]
prompt_tokens: 72
completion_tokens: 62
prompt_cache_hit_tokens: 0
prompt_cache_miss_tokens: 72

思考モードでは、答え（content）の前にモデルの考えた過程が reasoning_content として返ります。--no-thinking を付けると思考が無効になり、reasoning_content は返りません。

検索して、結果の質を数値で判定する

"""Search API の呼び出しサンプル。

使い方:
    python 02_search.py "DeepSeek site:github.com after:2026" --count 5 [--content]
"""

from __future__ import annotations

import argparse
import sys

from ds_client import DeepSeekToolError, format_tool_error, is_authoritative, is_high_quality, search


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="DeepSeek Search API を呼び出す")
    parser.add_argument("query", help="検索クエリ（検索演算子を含められる）")
    parser.add_argument("--count", type=int, default=5, help="取得件数（1〜50、既定 5）")
    parser.add_argument("--content", action="store_true", help="本文（content）も取得する")
    return parser.parse_args()


def main() -> None:
    args = parse_args()

    try:
        results = search(args.query, count=args.count, need_content=args.content)
    except DeepSeekToolError as exc:
        # count=51 のような不正な引数のときも、トレースバックではなく短いメッセージで終わる。
        print(format_tool_error(exc), file=sys.stderr)
        raise SystemExit(1)

    if not results:
        print("該当なし（空の結果は課金されません）")
        return

    high_quality = 0
    high_authority = 0
    for result in results:
        marks = ""
        if is_high_quality(result):
            marks += "★"
            high_quality += 1
        if is_authoritative(result):
            marks += "◆"
            high_authority += 1
        date = result.get("date") or "-"
        print(f"{marks:<2} rerank_score={result['rerank_score']:.1f} authority={result['authority']} date={date}")
        print(f"   {result['title']}")
        print(f"   {result['url']}")
        print(f"   {result['snippet']}")
        if args.content:
            print(f"   本文: {result['content'][:200]}")
        print()

    print(f"高品質 {high_quality} 件 / 高権威 {high_authority} 件 / 合計 {len(results)} 件")


if __name__ == "__main__":
    main()

python 02_search.py "DeepSeek site:github.com after:2026" --count 5
python 02_search.py "DeepSeek V4.1 Flash" --count 3 --content

★◆ rerank_score=2.6 authority=4 date=2026-10-10T00:00:00+00:00
   モック結果 1: DeepSeek site:github.com after:2026
   https://example.com/mock/1
   「DeepSeek site:github.com after:2026」に関する短い要約（モック、1 件目）。
…
高品質 3 件 / 高権威 3 件 / 合計 5 件

★ は rerank_score が 1.5 以上、◆ は authority が 3 以上の結果です。このサンプルの --count の既定は 5 です（API 自体の既定は 10）。2 行目の --content を付けると要求に need_content: true が入り、API が各結果の content に本文を入れて返します（取得できなかったページは空のまま）。サンプルは各結果の下に 本文: として先頭 200 文字を表示します。モックは count が 6 を超えても 6 件までしか返しません（実 API は最大 count 件）。検索語を変えて、★ と ◆ の数がどう変わるか見てください。

ページを Markdown として読む

"""Fetch API の呼び出しサンプル。

使い方:
    python 03_fetch.py https://api-docs.deepseek.com/guides/search --chars 600
"""

from __future__ import annotations

import argparse
import sys

from ds_client import DeepSeekToolError, fetch, format_tool_error


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="DeepSeek Fetch API を呼び出す")
    parser.add_argument("url", help="取得する URL")
    parser.add_argument("--chars", type=int, default=600, help="本文を表示する先頭文字数（既定 600）")
    return parser.parse_args()


def main() -> None:
    args = parse_args()

    try:
        page = fetch(args.url)
    except DeepSeekToolError as exc:
        if exc.status == 403:
            print("セキュリティポリシーで拒否されました。", file=sys.stderr)
        elif exc.status == 404:
            print("使える本文を取得できなかった（URL 不可・サイト方針・未対応形式など）。", file=sys.stderr)
        else:
            print(format_tool_error(exc), file=sys.stderr)
        raise SystemExit(1)

    print(f"[タイトル] {page['title'] or '(タイトルなし)'}")
    print("[本文]")
    print(page["content"][: args.chars])


if __name__ == "__main__":
    main()

python 03_fetch.py https://api-docs.deepseek.com/guides/search --chars 600
python 03_fetch.py https://blocked.example/page    # モックで 403 を体験
python 03_fetch.py https://missing.example/page    # モックで 404 を体験

モックでは、ホスト名が blocked.example なら 403、missing.example なら 404 を返します。エラー時の表示と終了コードを確かめてください。

出典つきで答えさせる

"""検索結果を文脈に入れて出典つきで回答させる、RAG（検索拡張生成）の最小形。

使い方:
    python 04_grounded_answer.py "質問"
"""

from __future__ import annotations

import argparse
import sys

from openai import APIError

from ds_client import MODEL, DeepSeekToolError, chat_client, format_chat_error, format_tool_error, is_high_quality, search


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="検索結果を根拠に出典つきで回答する")
    parser.add_argument("question", help="質問文")
    return parser.parse_args()


def build_context(selected: list[dict]) -> str:
    """選んだ検索結果を `[番号] タイトル / URL / 日付 / 要約` の形で文脈テキストにする。"""
    blocks = []
    for i, result in enumerate(selected, start=1):
        date = result.get("date") or "-"
        blocks.append(f"[{i}] {result['title']}\nURL: {result['url']}\n日付: {date}\n要約: {result['summary']}")
    return "\n\n".join(blocks)


def main() -> None:
    args = parse_args()

    if not args.question.strip():
        print("質問が空です。1 文字以上の質問を指定してください。", file=sys.stderr)
        raise SystemExit(2)

    # 公式ガイド（Search & Fetch）: query / ori_question はどちらも 1〜128 文字の制限があるため、
    # 質問が長い場合は先頭 128 文字に切り詰めてから渡す（チャットに渡す質問文そのものは切らない）。
    query = args.question[:128]
    try:
        results = search(query, count=8, ori_question=query)
    except DeepSeekToolError as exc:
        print(format_tool_error(exc), file=sys.stderr)
        raise SystemExit(1)
    if not results:
        print("検索結果が 0 件でした（空の結果は課金されません）。質問を変えて再実行してください。")
        raise SystemExit(1)

    high_quality = [r for r in results if is_high_quality(r)]
    if high_quality:
        selected = high_quality[:3]
    else:
        print("警告: 高品質（rerank_score >= 1.5）の結果が無いため、上位 3 件をそのまま使います。")
        selected = results[:3]

    context = build_context(selected)
    messages = [
        {
            "role": "system",
            "content": (
                "与えた資料だけを根拠に日本語で答え、文末に [番号] で出典を示してください。"
                "資料に無いことは「資料からは分からない」と書いてください。\n\n" + context
            ),
        },
        {"role": "user", "content": args.question},
    ]

    client = chat_client()
    # 出典つきの短い要約タスクなので思考モードを無効にし、応答を速く・安くする
    # （公式ガイド（Thinking Mode）の切替の実演。04 は 01/05 と対照的に非思考モードを使う）。
    try:
        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            extra_body={"thinking": {"type": "disabled"}},
        )
    except APIError as exc:
        print(format_chat_error(exc), file=sys.stderr)
        raise SystemExit(1)

    print("[回答]")
    print(response.choices[0].message.content)
    print()
    print("[出典]")
    for i, result in enumerate(selected, start=1):
        print(f"[{i}] {result['title']} ({result['url']})")


if __name__ == "__main__":
    main()

python 04_grounded_answer.py "DeepSeek の Search API の料金体系は？"

回答の文末に [1] のような番号が付き、その下に出典一覧が表示されれば成功です。モックは count が 6 を超えても 6 件までしか返しません（実 API は最大 count 件）。実 API では、検索語を変えて出典の顔ぶれがどう変わるか確かめてください。

自分で調べるエージェントを動かす

"""web_search / web_fetch を function tool としてモデルに渡すツール呼び出しエージェント。

使い方:
    python 05_search_agent.py "質問" [--max-steps 8] [--follow-up "2 問目"]
"""

from __future__ import annotations

import argparse
import json
import sys
from typing import Any

from openai import APIError, OpenAI

from ds_client import MODEL, DeepSeekToolError, chat_client, format_chat_error
from ds_client import fetch as ds_fetch
from ds_client import search as ds_search

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "web_search",
            "description": "Web を検索して関連ページの一覧を返す。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "検索クエリ（検索演算子を使える）"},
                    "count": {"type": "integer", "description": "取得件数（1〜10）", "minimum": 1, "maximum": 10},
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "web_fetch",
            "description": "指定した URL のページ本文を Markdown として取得する。",
            "parameters": {
                "type": "object",
                "properties": {"url": {"type": "string", "description": "取得する URL"}},
                "required": ["url"],
            },
        },
    },
]

FETCH_CONTENT_LIMIT = 6000  # 文脈節約のため、取得した本文は先頭 6000 文字に切る


def assistant_message(msg: Any) -> dict[str, Any]:
    """公式サンプルの `messages.append(response.choices[0].message)` と等価な辞書を作る。

    tools 付きの要求では、過去の assistant ターンの reasoning_content をすべて返送する
    （返送しないと 400）。ツールを呼ばなかったターンも同じなので、空文字でもキー自体は入れる。
    tool_calls が無ければキーごと省く。
    """
    message: dict[str, Any] = {
        "role": "assistant",
        "content": msg.content or "",
        "reasoning_content": getattr(msg, "reasoning_content", None) or "",
    }
    if msg.tool_calls:
        message["tool_calls"] = [tc.model_dump() for tc in msg.tool_calls]
    return message


def run_web_search(query: str, count: Any) -> tuple[str, str]:
    """web_search を実行する。戻り値は (モデルに返す JSON 文字列, 経過表示用の要約行)。"""
    # スキーマでは count を 1〜10 に制限しているが、non-strict モードではモデルが
    # 守らないことがある（999 を渡す、文字列で渡す、省略して None になる、など）ので丸める。
    try:
        count = int(count)
    except (TypeError, ValueError):
        count = 5
    count = min(max(count, 1), 10)

    try:
        results = ds_search(query, count=count)
    except DeepSeekToolError as exc:
        return json.dumps({"error": exc.message}, ensure_ascii=False), f"エラー: {exc.message}"
    trimmed = [
        {
            "title": r["title"],
            "url": r["url"],
            "date": r.get("date"),
            "authority": r["authority"],
            "rerank_score": r["rerank_score"],
            "summary": r["summary"],
        }
        for r in results
    ]
    return json.dumps(trimmed, ensure_ascii=False), f"検索結果 {len(trimmed)} 件"


def run_web_fetch(url: str) -> tuple[str, str]:
    """web_fetch を実行する。戻り値は (モデルに返す JSON 文字列, 経過表示用の要約行)。"""
    try:
        page = ds_fetch(url)
    except DeepSeekToolError as exc:
        return json.dumps({"error": exc.message}, ensure_ascii=False), f"エラー: {exc.message}"
    content = page["content"][:FETCH_CONTENT_LIMIT]
    trimmed = {"title": page["title"], "content": content}
    return json.dumps(trimmed, ensure_ascii=False), f"取得: {page['title']}（{len(content)} 文字）"


def summarize_messages(messages: list[dict[str, Any]]) -> str:
    """messages 配列を 1 行に要約する（--show-messages 用）。

    assistant だけは reasoning_content の有無と tool_calls の件数を表示し、
    2 ターン目以降も前のターンの reasoning_content が残っていることを確認できるようにする。
    """
    parts = []
    for message in messages:
        role = message.get("role")
        if role == "assistant":
            has_reasoning = "あり" if message.get("reasoning_content") else "なし"
            tool_call_count = len(message.get("tool_calls") or [])
            parts.append(f"assistant(reasoning_content {has_reasoning}, tool_calls {tool_call_count})")
        else:
            parts.append(str(role))
    return "送信 messages: " + " / ".join(parts)


def call_tool(name: str, arguments: Any) -> tuple[str, str]:
    """モデルが生成した引数を実行する。引数の形が壊れていても例外を投げず、エラーを結果として返す。"""
    if not isinstance(arguments, dict):
        message = "引数が JSON オブジェクトではありません"
        return json.dumps({"error": message}, ensure_ascii=False), f"エラー: {message}"

    if name == "web_search":
        if "query" not in arguments:
            message = "引数に query がありません"
            return json.dumps({"error": message}, ensure_ascii=False), f"エラー: {message}"
        return run_web_search(arguments["query"], arguments.get("count", 5))

    if name == "web_fetch":
        if "url" not in arguments:
            message = "引数に url がありません"
            return json.dumps({"error": message}, ensure_ascii=False), f"エラー: {message}"
        return run_web_fetch(arguments["url"])

    message = f"未知のツール: {name}"
    return json.dumps({"error": message}, ensure_ascii=False), f"エラー: {message}"


def run_loop(
    messages: list[dict[str, Any]],
    client: OpenAI,
    max_steps: int,
    *,
    turn: int,
    show_messages: bool,
) -> bool:
    """ツール呼び出しループを 1 ターン分回す。最終回答に到達したら True を返す。"""
    for step in range(1, max_steps + 1):
        if show_messages:
            # 2 ターン目以降でも、前のターンの assistant(reasoning_content あり) が
            # messages に残ったまま送られていることをここで確認できる。
            print(summarize_messages(messages))

        try:
            response = client.chat.completions.create(
                model=MODEL,
                messages=messages,
                tools=TOOLS,
                reasoning_effort="high",
                extra_body={"thinking": {"type": "enabled"}},
            )
        except APIError as exc:
            print(format_chat_error(exc), file=sys.stderr)
            raise SystemExit(1)
        msg = response.choices[0].message
        messages.append(assistant_message(msg))  # reasoning_content を必ず含める

        reasoning = (getattr(msg, "reasoning_content", None) or "")[:120]
        print(f"[turn {turn} / step {step}] 思考: {reasoning}")

        if not msg.tool_calls:
            print("[回答]")
            print(msg.content)
            return True

        for call in msg.tool_calls:
            try:
                arguments = json.loads(call.function.arguments)
            except json.JSONDecodeError:
                # モデルの出力が壊れていても、例外で落とさずツール結果としてエラーを返す。
                print(f"→ {call.function.name}(引数の JSON 解析に失敗)")
                print("← エラー: 引数を JSON として読めませんでした")
                error_result = json.dumps({"error": "引数を JSON として読めませんでした"}, ensure_ascii=False)
                messages.append({"role": "tool", "tool_call_id": call.id, "content": error_result})
                continue

            print(f"→ {call.function.name}({arguments})")
            result, summary_line = call_tool(call.function.name, arguments)
            print(f"← {summary_line}")
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})

    print("ステップ上限に達しました。")
    return False


def _positive_int(value: str) -> int:
    """argparse 用: 1 以上の整数のみを受け付ける。"""
    parsed = int(value)
    if parsed < 1:
        raise argparse.ArgumentTypeError("1 以上の整数を指定してください")
    return parsed


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="web_search / web_fetch を使うツール呼び出しエージェント")
    parser.add_argument("question", help="質問文")
    parser.add_argument("--max-steps", type=_positive_int, default=8, help="1 ターンあたりの最大ステップ数")
    parser.add_argument("--follow-up", help="2 問目の質問（指定すると続けて 2 ターン目を実行する）")
    parser.add_argument(
        "--show-messages",
        action="store_true",
        help="各ステップで送信する messages 配列を 1 行要約で表示する",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    client = chat_client()
    messages: list[dict[str, Any]] = [{"role": "user", "content": args.question}]

    if not run_loop(messages, client, args.max_steps, turn=1, show_messages=args.show_messages):
        raise SystemExit(1)

    if args.follow_up:
        # 前のターンの reasoning_content も messages に残したまま送り続ける。
        messages.append({"role": "user", "content": args.follow_up})
        if not run_loop(messages, client, args.max_steps, turn=2, show_messages=args.show_messages):
            raise SystemExit(1)


if __name__ == "__main__":
    main()

python 05_search_agent.py "DeepSeek の Fetch API の QPS 上限は？"
python 05_search_agent.py "DeepSeek の Fetch API の QPS 上限は？" --follow-up "Search API の上限は？" --show-messages

[turn 1 / step 1] 思考: ユーザーの質問に答えるため、まず web_search で情報を集めます。
→ web_search({'query': 'DeepSeek の Fetch API の QPS 上限は？', 'count': 3})
← 検索結果 3 件
[turn 1 / step 2] 思考: 検索結果の中から有力なページを 1 件取得します。
→ web_fetch({'url': 'https://example.com/mock/1'})
← 取得: Mock page（53 文字）
[turn 1 / step 3] 思考: 収集した資料を踏まえて最終回答をまとめます。
[回答]
収集した1 件の資料（例: Mock page）をもとに回答します（モック応答）。

各ステップで、モデルの思考の冒頭・呼んだツール・結果（検索の件数、取得したページのタイトルと文字数、またはエラー）が表示されます。--follow-up を付けると、同じ履歴に 2 問目を足してもう一度ループを回します（表示は turn 2）。--show-messages を付けると、モデルを呼ぶ直前に、送る messages の並びを 送信 messages: で始まる 1 行に要約して表示します。2 行目のコマンドを実行し、[turn 2 / step 1] の直前の行に、1 問目の assistant(reasoning_content あり, …) が残っていることを確かめてください。

送信 messages: user / assistant(reasoning_content あり, tool_calls 1) / tool / assistant(reasoning_content あり, tool_calls 1) / tool / assistant(reasoning_content あり, tool_calls 0) / user

--max-steps は 1 ターンあたりのステップ数の上限で、1 以上の整数を指定します（0 以下は使い方のエラーとして終了コード 2）。

再試行・空の結果・入力エラーを確かめる

python 02_search.py "mock:429 DeepSeek"     # 最初の 2 回は 429、3 回目で成功
python 02_search.py "mock:empty DeepSeek"   # results が空（課金されない）
python 02_search.py "DeepSeek" --count 51   # count が範囲外 → 400、再試行しない

python 02_search.py "mock:402 DeepSeek"     # 残高不足 → 402、再試行しない
python 01_hello.py "mock:chat400 テスト"     # Chat API の 400 をエラー表示で確かめる

モックは検索語に mock:429 を含むと最初の 2 回だけ 429 を返し、mock:empty を含むと空の結果を返します。429 の回数は、モックを起動してからの通算で検索語ごとに数えます。もう一度試すときはモックを再起動してください。1 つ目は待ち時間のあとに結果が表示されれば、再試行が働いています。3 つ目は再試行せずに エラー（HTTP 400）: 'count' must be between 1 and 50 と表示し、終了コード 1 で止まります。エラーの message に、直すべき項目が書かれていることを確かめてください。

ほかにも、検索語に mock:408 を含むと最初の 2 回だけ 408、mock:500 を含むと常に 500（4 回試して打ち切り）、mock:402 を含むと 402 を返します。Chat では、質問に mock:chat400 を含めると 400 になります。「余裕があれば」の 2 つは、それぞれ次のように表示して終了コード 1 で終わります。

エラー（HTTP 402）: 残高不足です。入金してから再試行してください（サーバー: Insufficient Balance）。
エラー（HTTP 400）: forced 400 for testing (mock:chat400)