--- title: "DSPy MCPサーバー：Bright Dataでウェブ調査エージェントを構築する" slug: dspy-with-bright-data date: 2026-10-22T07:14:13+00:00 modified: 2026-10-22T07:14:14+00:00 permalink: https://brightdata.jp/blog/ai/dspy-with-bright-data type: blog --- [ ブログ ](https://brightdata.jp/blog "ブログ") / [AI](https://brightdata.jp/blog/ai) [AI](https://brightdata.jp/blog/ai) # DSPy MCPサーバー：Bright Dataでウェブ調査エージェントを構築する Bright Data MCPサーバーでDSPyウェブ調査エージェントを構築し、評価し、GEPAで最適化する方法を学びましょう。 14 分読 [ ![Satyam Tripathi](https://media.brightdata.jp/2024/09/Satyam-Tripathi-50x50.png) ](https://brightdata.com/blog/authors/satyam-tripathi) [Satyam Tripathi Technical Writer ](https://brightdata.com/blog/authors/satyam-tripathi) ![DSPy with Bright Data](https://media.brightdata.jp/2026/10/dspy-mcp-server-web-research-agent.png) DSPyは、データを使って言語モデルのプログラムを改善するPythonフレームワークです。DSPyはModel Context Protocol（MCP）サーバーからエージェントツールを読み込みます。Bright Data MCPサーバーを使うと、エージェントはGoogleで検索し、ページをMarkdownとして読み取り、サーバーがボット対策を処理します。 DSPyのMCPチュートリアルのパターンは、単一のエージェント実行ではうまく機能します。DSPyの評価機能とプロンプトオプティマイザーであるGEPAは、ワーカースレッド上でエージェントを実行しますが、テストでは非同期ツール呼び出しが黙って失敗し、エージェントが記憶から回答していました。 このガイドでは、そのエージェントを構築し、そのツールを並行実行し、ウェブデータが何をもたらすかを測定する方法を示します。 ## 要約 - **ウェブデータが正答率を高めました。**Bright Data MCPサーバーを使うと、`gpt-5.4-mini`は60問中54～58問を正しく回答し、14問から向上し、ナレッジカットオフ後の出来事に関する8問全てに正解し、0問から向上しました。 - **2つのツールを選び、エリシテーションコールバックを追加する。**ツールを減らすことでプロンプトを小さく保てます。コールバックは、プロンプトインジェクションの可能性があるとマークされたページを承認します。 - **スレッド実行には同期ツールを使う。**非同期ツールは`dspy.Evaluate`とGEPAで黙って失敗するか、ハングしました。バックグラウンドイベントループで両方とも修正されました。 - **まずツールを修正する。**GEPAはスコアを上げることなく、学習用の回答を指示文にそのままコピーしました。長いページに関する6つの質問では、`find`引数によってプロンプトトークンが半分になりました。 ## 必要なもの エージェントは、ホスト型のBright Data MCPサーバーである[Bright DataのWeb MCP](/ai/mcp-server)にHTTP経由で接続するため、インストールが必要なランタイムはPythonのみです。このサーバーは[Web Unlocker API](/products/web-unlocker)上で動作しており、エージェントのためのプロキシローテーションとボット対策を処理します。コードについては、[関連のGitHubリポジトリ](https://github.com/triposat/dspy-bright-data-research-agent)に、評価ステップ用の`evaluate.py`と`post_cutoff_questions.csv`を含む全てのスクリプトとデータファイルがあります。必要なものは以下の通りです。 - Python 3.10以降、パッケージをインストールするための[uv](https://docs.astral.sh/uv/) - 無料のBright Dataアカウント、および[**アカウント設定**](/cp/setting/users)から取得するそのAPIキー - OpenAIのAPIキー、または`dspy.LM`がツール呼び出しに対応している他のプロバイダーのキー **ユーザーとAPIキー**タブの**APIキー**テーブルからBright Dataのキーをコピーします。 ![アカウント設定の「ユーザーとAPIキー」タブを示すBright Dataのコントロールパネル。強調表示されたAPIキーテーブルには、ユーザー権限、アクティブステータス、「表示」を選択するまで非表示になっているキーの1件が一覧表示されている。アカウントの詳細はぼかされている。](https://media.brightdata.com/2026/10/SkCRHLm3zl.png)*同じキーが、ホスト型サーバーとローカルサーバーの両方で使用できます。* リポジトリをクローンし、テスト済みバージョンに固定したMCPエクストラ付きでDSPyをインストールし、両方のキーをエクスポートします。 ```none git clone https://github.com/triposat/dspy-bright-data-research-agent && cd dspy-bright-data-research-agent uv venv && source .venv/bin/activate uv pip install "dspy[mcp]==3.4.1" "mcp==2.3.0" export BRIGHTDATA_API_TOKEN="your-bright-data-api-key" export OPENAI_API_KEY="your-openai-api-key" ``` DSPy 3.4.1の`dspy[mcp]`エクストラはSDKバージョンを固定していないため、このコマンドはMCP Python SDKをテスト済みの2.3.0リリースに固定します。無料の月間リクエストから始めることができ、[MCPサーバーの料金ページ](/pricing/mcp-server)で現在の割り当てを確認できます。 ## Bright Data MCPサーバーをDSPyに接続する `dspy.Tool.from_mcp_tool`は、MCPサーバーが一覧表示する各ツールをDSPyツールに変換するため、エージェントは各ツールの名前、説明、引数のスキーマを認識します。以下のスクリプトは、リポジトリ内の`agent.py`で、ホスト型のBright Data MCPサーバーに接続し、検索とスクレイピングのツールを読み込みます。その後、モデルのナレッジカットオフ後の出来事について質問します。 ```none import asyncio import os import dspy from mcp import types from mcp.client import Client from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client dspy.configure( lm=dspy.LM("openai/gpt-5.4-mini", temperature=1.0, max_tokens=16000), adapter=dspy.ChatAdapter(use_native_function_calling=True), ) SERVER_URL = "https://mcp.brightdata.com/mcp?tools=search_engine,scrape_as_markdown" HEADERS = {"Authorization": f"Bearer {os.environ['BRIGHTDATA_API_TOKEN']}"} async def approve_page(context, params): # The server asks before returning a page that contains MCP setup # instructions. This agent's tools only read the web, so return it. return types.ElicitResult(action="accept", content={"return_content": True}) class ResearchQuestion(dspy.Signature): """Answer the question with facts found on the web.""" question: str = dspy.InputField() answer: str = dspy.OutputField(desc="a short, direct answer") sources: list[str] = dspy.OutputField(desc="URLs of the pages that support the answer") async def main(): async with create_mcp_http_client(headers=HEADERS) as http: transport = streamable_http_client(SERVER_URL, http_client=http) async with Client(transport, elicitation_callback=approve_page) as client: mcp_tools = (await client.list_tools()).tools tools = [dspy.Tool.from_mcp_tool(client, t) for t in mcp_tools] agent = dspy.ReActV2(ResearchQuestion, tools=tools, max_iters=8) question = "Who won the 2026 FIFA World Cup final, and what was the score?" pred = await agent.acall(question=question) for event in pred.history.messages: for call in getattr(event.get("tool_calls"), "tool_calls", []): print(f"{call.name}: {call.args}") print("\nAnswer:", pred.answer) print("Sources:", pred.sources) asyncio.run(main()) ``` `python agent.py`を実行すると、2回の検索が表示され、その後に最終的な`submit`呼び出しが表示されました。 ```none search_engine: {'query': '2026 FIFA World Cup final winner score', 'engine': 'google', 'cursor': '', 'geo_location': 'us'} search_engine: {'query': '2026 FIFA World Cup final winner score site:en.wikipedia.org OR site:fifa.com', 'engine': 'google', 'cursor': '', 'geo_location': 'us'} submit: {'answer': 'Spain won the 2026 FIFA World Cup final, defeating Argentina 1–0 after extra time.', 'sources': ['https://www.fifa.com/en/tournaments/mens/worldcup/canadamexicousa2026/final', 'https://www.fifa.com/en/tournaments/mens/worldcup/canadamexicousa2026/articles/spain-argentina-final-report-highlights']} Answer: Spain won the 2026 FIFA World Cup final, defeating Argentina 1–0 after extra time. Sources: ['https://www.fifa.com/en/tournaments/mens/worldcup/canadamexicousa2026/final', 'https://www.fifa.com/en/tournaments/mens/worldcup/canadamexicousa2026/articles/spain-argentina-final-report-highlights'] ``` ツールなしでは、`gpt-5.4-mini`は2026年のワールドカップ決勝がナレッジカットオフ後であると述べました。 このスクリプトは、ローカルのデモサーバーを使用する[DSPy MCPチュートリアル](https://dspy.ai/3.4.1/tutorials/mcp/)をベースにしており、ホスト型サーバー用の設定を3つ追加しています。 **APIキーはヘッダーで送信する。**`create_mcp_http_client`は、全てのリクエストでキーを`Authorization`ヘッダーとして送信するため、プロキシやログに記録されるURLにキーが現れません。 **エージェントが必要とするツールを選ぶ。**Bright Data MCPサーバーは、検索、スクレイピング、バッチジョブ、ブラウザ自動化、および[Amazon](/ai/mcp-server/amazon)や[LinkedIn](/ai/mcp-server/linkedin)などのサイト向けに構造化されたJSONを返す事前構築済みスクレイパーのツールを提供します。`tools`URLパラメータでエージェントが認識するツールを選択し、[MCPサーバーのツールリファレンス](https://docs.brightdata.com/products/mcp-server/tools)には`search_engine_batch`や`scraping_browser_navigate`など各名称が一覧表示されています。選択した各ツールのスキーマは全てのモデル呼び出しと共に送信されるため、このエージェントではプロンプトを小さく保つために`search_engine`と`scrape_as_markdown`のみを選択します。 **サーバーのエリシテーション要求に回答する。**Bright Data MCPサーバーにはプロンプトインジェクション対策機能が含まれています。スクレイピングされたページにMCP設定を変更する指示が含まれている場合、サーバーはそのページを返す前に、[MCPエリシテーション](https://modelcontextprotocol.io/specification/2026-07-28/client/elicitation)を通じてクライアントに確認を求めます。DSPyクライアントが回答できるよう、`elicitation_callback`引数を使ってエリシテーションコールバックを`Client`に渡します。 `approve_page`は、このエージェントがウェブを読み取るだけなので、全てのリクエストを承認します。ファイルを書き込んだり自身の設定を変更したりできるエージェントの場合は、リクエストを人間に送って承認を得てください。 代わりに[サーバーをローカルで実行する](https://docs.brightdata.com/products/mcp-server/integrations/overview)には、Node.jsをインストールし、同じ`Client`に`StdioServerParameters(command="npx", args=["-y", "@brightdata/mcp@latest"], env=...)`を渡し、`env`に`API_TOKEN`と`TOOLS`を指定します。`@latest`を使うと、npxは起動のたびに`latest`タグが指すリリースを取得します。再現可能な評価のためには、これを[Bright Data MCPリリース](https://github.com/brightdata/brightdata-mcp/releases)からテスト済みのバージョンに置き換えてください。 Jupyterノートブックでは、すでにイベントループが実行されているため、`asyncio.run(main())`を`await main()`に置き換えます。 このスクリプトは`dspy.ReActV2`を使用しており、[DSPy ReActのドキュメント](https://dspy.ai/3.4.1/diving-deeper/react/)には、DSPy 3.5以降`dspy.ReAct`が使用する実装として記載されており、3.6ではReActV2という名称は削除されています。ReActV2は`trajectory`の代わりに`history`を返し、`submit`ツールで終了します。アップグレードする際は、`dspy.ReActV2`を`dspy.ReAct`に変更し、評価を再実行してください。 同じサーバーは、[AutoGen](/blog/ai/autogen-agentchat-with-web-mcp)、[LangGraph](/blog/ai/langgraph-with-web-mcp)、[Pydantic AI](/blog/ai/pydantic-ai-with-web-mcp)など、他のエージェントフレームワークとも動作します。 ## スレッドとGEPAでDSPy MCPツールを実行する DSPy 3.4.1では、`dspy.Evaluate`はワーカースレッド上でプログラムを同期的に実行し、[GEPA](https://dspy.ai/3.4.1/api/optimizers/GEPA/overview/)はそれを使って各候補の指示文を評価します。`from_mcp_tool`によるツールは非同期関数であり、MCPセッションを開いたイベントループに紐付いています。テストでは、これらを同期ワーカースレッドから呼び出すと、見落としやすい2つの方法で失敗しました。 DSPyのMCPチュートリアル通りにツールを読み込んだ場合、`Evaluate`は例外を出さずに完了しました。全てのツール呼び出しはサーバーに到達する前に失敗し、Pythonはコルーチンが一度もawaitされなかったという警告のみを出力しました。エージェントは記憶から回答し、それでも`Evaluate`はスコアを報告しました。DSPyの`allow_tool_async_sync_conversion`設定を有効にすると、実行はハングし、9分経っても待機し続けていました。 この修正では、単一のMCPセッションをバックグラウンドのイベントループ上で開いたままにし、各呼び出しをそのループに送る同期関数をDSPyに与えます。 ![BrightDataTools内の単一イベントループにツール呼び出しを送信する3つの同期呼び出し元（agent()、dspy.Evaluate、GEPAスレッド）の図。このループは、Bright Data MCPサーバーへの単一のMCPセッションを開いたまま保持する。](https://media.brightdata.com/2026/10/SJ1x8UQhfx.png)*各同期呼び出しは`run_coroutine_threadsafe`によってループに送信されるため、MCPセッションを開いたイベントループのみがそのセッションを使用します。* このクラスは、評価スクリプトと最適化スクリプトがインポートするモジュールである`brightdata_tools.py`にあります。 ```none import asyncio import json import os import re import threading from urllib.parse import urlparse import dspy from mcp import StdioServerParameters, types from mcp.client import Client from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client SERVER_URL = "https://mcp.brightdata.com/mcp?tools=search_engine,scrape_as_markdown" # The Bright Data MCP server wraps results in a security notice and two markers WRAPPED = re.compile(r"(=====UNTRUSTED_\w+_BEGIN=====\n)(.*)(\n=====UNTRUSTED_\w+_END=====)", re.S) async def approve_page(context, params): # Read-only tools, so pages with MCP setup text are returned as data return types.ElicitResult(action="accept", content={"return_content": True}) class BrightDataTools: """One MCP session on a background event loop, exposed as sync dspy.Tools.""" def __init__(self, country="us", max_chars=20_000, retries=2, blocked_domains=(), timeout=180, cache=False, remote=True): self.country = country self.max_chars = max_chars self.retries = retries self.blocked = tuple(blocked_domains) self.timeout = timeout self.cache = {} # pages always, searches too when cache=True self.keep_all = cache # reuse search results while optimizing self.lock = threading.Lock() self.stats = {"calls": 0, "sent": 0, "errors": 0} self.error = None # why the MCP session failed to start or stopped self.remote = remote self.loop = asyncio.new_event_loop() self.ready = threading.Event() self.thread = threading.Thread(target=self.loop.run_until_complete, args=(self._serve(),), daemon=True) self.thread.start() self.ready.wait() if self.error: raise RuntimeError("Could not connect to the MCP server") from self.error async def _serve(self): try: await self._connect() except Exception as e: # report it instead of waiting forever self.error = e finally: self.ready.set() async def _connect(self): token = os.environ["BRIGHTDATA_API_TOKEN"] if self.remote: # hosted server over streamable HTTP async with create_mcp_http_client(headers={"Authorization": f"Bearer {token}"}) as http: async with Client(streamable_http_client(SERVER_URL, http_client=http), elicitation_callback=approve_page) as client: await self._hold(client) else: # local server, latest release local = StdioServerParameters( command="npx", args=["-y", "@brightdata/mcp@latest"], env={"API_TOKEN": token, "TOOLS": "search_engine,scrape_as_markdown", "PATH": os.environ["PATH"]}) async with Client(local, elicitation_callback=approve_page) as client: await self._hold(client) async def _hold(self, client): self.client = client self.mcp_tools = (await client.list_tools()).tools self.stop = asyncio.Event() self.ready.set() await self.stop.wait() def close(self): self.loop.call_soon_threadsafe(self.stop.set) self.thread.join(timeout=30) # let the session shut down cleanly def _is_blocked(self, url): host = urlparse(url).netloc.lower() return any(host == d or host.endswith("." + d) for d in self.blocked) def call(self, name, args): start = int(args.pop("start", 0) or 0) # where to resume a long page find = args.pop("find", "") or "" # phrase to look for in a long page if name == "search_engine": args = {**args, "geo_location": self.country} if name == "scrape_as_markdown" and self._is_blocked(args.get("url", "")): return "This domain is excluded for this task. Use another source." key = (name, json.dumps(args, sort_keys=True)) with self.lock: self.stats["calls"] += 1 text = self.cache.get(key) if text is None: try: text = self._send(name, args) except Exception: with self.lock: self.stats["errors"] += 1 # the agent sees an error, not a page raise with self.lock: self.stats["sent"] += 1 if self.keep_all or name == "scrape_as_markdown": self.cache[key] = text # pages stay cached, so reading on is free wrapped = WRAPPED.search(text) head, body, tail = (text[:wrapped.start(2)], wrapped.group(2), text[wrapped.end(2):] ) if wrapped else ("", text, "") if name == "search_engine" and self.blocked: data = json.loads(body) data["organic"] = [r for r in data.get("organic", []) if not self._is_blocked(r["link"])] body = json.dumps(data, indent=1) if find: spans = [(max(0, m.start() - 1500), m.end() + 1500) for m in re.finditer(re.escape(find), body, re.I)][:6] body = ("\n...\n".join(body[a:b] for a, b in spans) or f"[{find!r} not found in the {len(body)}-character page]") end = start + self.max_chars if len(body) > end: body = (body[start:end] + f"\n[showing characters {start} to {end} of {len(body)}. " f"Call {name} again with start={end} to read on, or with " "find='a word or phrase' to get matching passages from the whole page]") elif start: body = body[start:] + f"\n[end of page: characters {start} to {len(body)}]" return head + body + tail def _send(self, name, args): if self.error or not self.thread.is_alive(): raise RuntimeError("The MCP session is closed") from self.error for _ in range(self.retries + 1): future = asyncio.run_coroutine_threadsafe( self.client.call_tool(name, args), self.loop) try: result = future.result(self.timeout) finally: future.cancel() # a timed-out call stops instead of running on text = "\n".join(c.text for c in result.content if hasattr(c, "text")) wrapped = WRAPPED.search(text) body = wrapped.group(2) if wrapped else text if not result.is_error and body.strip(): return text raise RuntimeError(text[:500] or "empty result") def tools(self): out = [] for t in self.mcp_tools: spec = dspy.Tool.from_mcp_tool(self.client, t) # name, description, args args, types, desc = dict(spec.args), dict(spec.arg_types), spec.desc if t.name == "scrape_as_markdown": # let the agent read beyond the character limit args["start"] = {"type": "integer", "default": 0, "description": "Character offset to read from, for long pages"} args["find"] = {"type": "string", "default": "", "description": "Word or phrase to find; returns matching passages"} types["start"], types["find"] = int, str desc += (f" Returns {self.max_chars} characters at a time; pass find to search" " the whole page for a word or phrase.") def run(_name=t.name, **kwargs): return self.call(_name, kwargs) out.append(dspy.Tool(run, name=spec.name, desc=desc, args=args, arg_types=types)) return out ``` `BrightDataTools().tools()`は、サーバーのツール名、説明、引数を保持した2つの同期ツールを返します。スクレイピングツールには、`start`と`find`という2つの追加引数があり、`max_chars`が設定する文字数制限を超えてエージェントがテキストを読めるようにします。Bright Data MCPサーバーは、ページテキストにマークを付けてモデルにデータとして扱うよう伝えるセキュリティ通知で結果をラップします。このクラスは、マークされたテキストのみをフィルタリングし切り詰めるため、モデルは通知全体を受け取ります。 コンストラクタの設定： - `remote=False`はホスト型サーバーの代わりにローカルサーバーを使用します。 - `country`は全ての検索で`geo_location`を設定するため、全ての実行で同じ国の結果が得られます。 - `max_chars`は、完全なWikipedia記事など非常に長いページを公開しているサイトがあるため、各結果を20,000文字に制限します。この制限がない場合、あるモデルリクエストでは491,564トークンとなり、モデルのコンテキストウィンドウを超えました。これがあると、2回の評価実行で、どの質問も58,651プロンプトトークン以上を使用しませんでした。ページはキャッシュされたままなので、`start`や`find`を使った呼び出しは新しいリクエストを送信しません。 - `retries`と`timeout`は、長時間の評価実行を完了させるのに役立ちます。エラーを返した呼び出しは最大2回まで再送信され、`timeout`秒経っても実行中の呼び出しはバックグラウンドで継続する代わりにキャンセルされます。それでも失敗した呼び出しは`stats["errors"]`でカウントされ、`evaluate.py`と`optimize.py`がそれを表示します。エージェントはツールが失敗すると記憶から回答するため、スコアを信頼する前にこれを確認してください。 - `blocked_domains`は選択したサイトを検索結果から除外し、それらのサイトのスクレイピングをブロックします。評価中に使用し、本番環境では空のままにしてください。 - `cache=True`は検索結果も再利用し、`optimize.py`でこれを有効にします。 `BrightDataTools`は、`dspy.Evaluate`やGEPAのような同期・スレッド化されたジョブ向けに構築されており、Jupyterノートブックでも変更なしに動作します。FastAPIエンドポイントなどの非同期ウェブサービスでは、`BrightDataTools`上に構築されたエージェントを`await asyncio.to_thread(agent, question=question)`でワーカースレッド内で実行し、イベントループが他のリクエストの処理を続けられるようにしてください。 ## SimpleQA Verifiedでエージェントを評価する GEPAはメトリクスを使ってプログラムを評価することで改善するため、エージェントにはまず正解がわかっている質問が必要です。[SimpleQA Verified](https://huggingface.co/datasets/google/simpleqa-verified)がこれを提供します。これは、検証済みの回答を持つ1,000の短い事実に関する質問からなる、MITライセンスの公開セットです。 質問をリポジトリフォルダにダウンロードします。 ```none curl -L -o simpleqa_verified.csv https://huggingface.co/datasets/google/simpleqa-verified/resolve/main/simpleqa_verified.csv ``` `research.py`には、エージェント、`grader`という名のSimpleQAスタイルのLLM判定器、GEPAが使用するメトリクスが含まれています。 ```none import csv import random import dspy # Hosts that publish the benchmark answers, excluded while evaluating EVAL_BLOCKLIST = ("huggingface.co", "github.com", "raw.githubusercontent.com", "kaggle.com", "openaipublic.blob.core.windows.net") AGENT_LM = dspy.LM("openai/gpt-5.4-mini", temperature=1.0, max_tokens=16000) JUDGE_LM = dspy.LM("openai/gpt-5.4-mini", temperature=1.0, max_tokens=16000) class ResearchQuestion(dspy.Signature): """Answer the question with facts found on the web.""" question: str = dspy.InputField() answer: str = dspy.OutputField(desc="a short, direct answer") sources: list[str] = dspy.OutputField(desc="URLs of the pages that support the answer") def make_agent(tools): return dspy.ReActV2(ResearchQuestion, tools=tools, max_iters=8) def load_examples(path="simpleqa_verified.csv", seed=7): rows = list(csv.DictReader(open(path))) random.Random(seed).shuffle(rows) return [dspy.Example(question=r["problem"], gold=r["answer"]).with_inputs("question") for r in rows] class Grade(dspy.Signature): """Grade a predicted answer against the gold answer, SimpleQA style. CORRECT: the prediction contains the gold answer's meaning and does not contradict it. Small spelling or formatting differences are fine; for numbers and dates the value must match. INCORRECT: the prediction states a different or contradictory answer, or hedges between several. NOT_ATTEMPTED: the prediction does not give an answer.""" question: str = dspy.InputField() gold_answer: str = dspy.InputField() predicted_answer: str = dspy.InputField() grade: str = dspy.OutputField(desc="CORRECT, INCORRECT, or NOT_ATTEMPTED") grader = dspy.Predict(Grade) def tool_calls(pred): calls = [] history = getattr(pred, "history", None) for event in (history.messages if history else []): for call in getattr(event.get("tool_calls"), "tool_calls", []): if call.name != "submit": calls.append((call.name, call.args)) return calls def metric(gold, pred, trace=None, pred_name=None, pred_trace=None): """Score for GEPA, plus feedback that never shows the reference answer.""" answer = getattr(pred, "answer", "") or "" with dspy.context(lm=JUDGE_LM): grade = grader(question=gold.question, gold_answer=gold.gold, predicted_answer=answer).grade.strip().upper() correct = grade.startswith("CORRECT") calls = tool_calls(pred) searches = [args.get("query") for name, args in calls if name == "search_engine"] pages = [args.get("url") for name, args in calls if name == "scrape_as_markdown"] feedback = (f"Grade: {'CORRECT' if correct else 'INCORRECT'}. Agent answered: {answer!r}. " f"Searches ({len(searches)}): {searches}. Pages opened ({len(pages)}): {pages}.") if not correct and not pages: feedback += " The agent answered without opening a page." return dspy.Prediction(score=float(correct), feedback=feedback) def leaked_answers(instruction, examples): """Reference answers (4+ characters) that an optimized instruction repeats.""" text = instruction.lower() return sorted({ex.gold for ex in examples if len(ex.gold) >= 4 and ex.gold.lower() in text}) ``` 判定器は、正解・不正解・未回答というSimpleQAの3つの評価を割り当てるため、エージェント自身の小型モデルを使用します。メトリクスはスコアと、エージェントの検索やページを一覧表示するが参照用の正解は示さない書面でのフィードバックを返します。評価中、`EVAL_BLOCKLIST`はベンチマークの正解を公開しているサイトをブロックするため、エージェントは公開ウェブ上で各正解を見つける必要があります。 リポジトリ内の`evaluate.py`は、シャッフルされた質問の一部をエージェントに対して一度に8問ずつ実行し、全ての回答を`results.json`に保存します。デフォルトでは、60のテスト質問を評価します。`--no-tools`はモデル単体を評価し、`--local`はローカルサーバーを使用します。実行結果は次の通りです。 セットアップ（全て`gpt-5.4-mini`）同じ60のSimpleQA Verified質問のうち正解数モデル単体、ツールなし14（1回実行）Bright Dataツールを使用したエージェント、ローカルサーバー、`find`なし58、55、58、54（4回実行）Bright Dataツールを使用したエージェント、ホスト型サーバー、`find`あり57（1回実行）モデル単体による46件の誤りは全て具体的な誤答で、例えば、ネブラスカ州ビアトリス出身の画家に対してオマハと答えるようなものでした。モデルの誤った回答は、正しい回答のように見えました。 同じエージェントを繰り返し実行すると、単一のスコアのうちどれだけがノイズであるかがわかります。エージェントの4回の実行結果が異なったのは、ランダムなモデルサンプリングと判定器の採点が一部原因でした。エージェントは同じジョン・ケージの楽曲について実行ごとに異なる楽器名を挙げ、判定器は同じチェスのスコアに対して「4½ポイント」を正解と、「4.5」を不正解と採点しました。 60問中4問までの変化はノイズとして扱い、プログラムの比較は合計点ではなく、保存された結果の質問ごとに行ってください。DSPyはデフォルトでモデル呼び出しをキャッシュするため、実行を繰り返す前に両方のLMで`cache=False`を設定してください。 特に判定器がエージェントと同じモデルである場合は、判定器の採点も抜き取り確認してください。別の1回の実行を手動でレビューしたところ、正解と照合した全60件の採点が正しいものでした。 **実行コスト。**60問の実行は平均で1質問あたり2.3回のツール呼び出し、Bright Dataへのリクエストは200件未満、エージェント用の`gpt-5.4-mini`呼び出しコストは0.29ドルでした。評価全体が無料の月間リクエスト枠に収まります。 ## モデルのナレッジカットオフ後の出来事についてエージェントをテストする SimpleQA Verifiedの安定した質問群は、モデルのナレッジカットオフ後の事実をテストするようには設計されていません。これらの事実については、リポジトリ内の`post_cutoff_questions.csv`に、Wikipediaと照合して確認された、2025年と2026年の出来事に関する14の質問が含まれています。`python evaluate.py --csv post_cutoff_questions.csv --start 0 --n 14`を実行し、次に`--no-tools`を付けて再度実行します。各セットアップは1回ずつ実行しました。 対象となる質問（各1回実行）`gpt-5.4-mini`単体Bright Dataツールを使用した`gpt-5.4-mini`、ホスト型サーバー2025年の出来事6問中5問6問中6問2026年の出来事8問中0問8問中8問ツールなしでは、モデルは2026年のいくつかの出来事がまだ起きていないと答えました。ナレッジカットオフは2024年6月だと述べていたにもかかわらず、2025年の事実については6問中5問を把握していました。モデルが述べるナレッジカットオフを信頼するのではなく、モデルが何を知っているかをテストし、最近公開されたページにのみ答えが掲載されている、自分自身のユースケースからの質問を追加してください。 ## GEPAでエージェントを最適化する GEPAはDSPyの振り返り型プロンプトオプティマイザーです。学習用質問の小さなバッチでプログラムを実行します。その後、失敗した実行結果とメトリクスのフィードバックを、これらの実行では`gpt-5.5`である、より強力な振り返りモデルに提示します。GEPAは、検証用セットで最も良い成績を出す新しい指示文を保持します。 **GEPAは学習するために誤答を必要とします。**エージェントがサンプリングされた全ての学習用質問に正しく回答すると、GEPAは新しい指示文を提案しません。 そのため、エージェントが間違える質問から学習用セットを構築してください。テストセット外の質問に対して`python evaluate.py --start 108 --n 400 --out pool.json`を実行します。次に`optimize.py`は、12件の誤答と12件の正答を組み合わせて学習用セットとし、同じ方法で検証用セットを構築します。 ```none import json import random import dspy from brightdata_tools import BrightDataTools from research import AGENT_LM, EVAL_BLOCKLIST, leaked_answers, make_agent, metric dspy.configure(lm=AGENT_LM, adapter=dspy.ChatAdapter(use_native_function_calling=True)) # Half wrong, half correct answers from an evaluate.py run outside the test set pool = json.load(open("pool.json")) misses = [r for r in pool if not r["correct"] and not r.get("error")] hits = [r for r in pool if r["correct"]] random.Random(0).shuffle(hits) def to_ex(rows): return [dspy.Example(question=r["question"], gold=r["gold"]).with_inputs("question") for r in rows] half = min(12, len(misses) // 2) trainset = to_ex(misses[:half] + hits[:half]) valset = to_ex(misses[half:2 * half] + hits[half:2 * half]) bd = BrightDataTools(blocked_domains=EVAL_BLOCKLIST, cache=True) agent = make_agent(bd.tools()) gepa = dspy.GEPA( metric=metric, reflection_lm=dspy.LM("openai/gpt-5.5", temperature=1.0, max_tokens=32000), max_metric_calls=300, num_threads=8, track_stats=True, log_dir="gepa_logs", ) optimized = gepa.compile(agent, trainset=trainset, valset=valset) optimized.save("optimized_agent.json") instruction = optimized.react.signature.instructions print(instruction) print("Training answers in the instruction:", leaked_answers(instruction, trainset + valset)) print("Tool calls:", bd.stats["calls"], "| requests sent:", bd.stats["sent"], "| errors:", bd.stats["errors"]) bd.close() ``` `python optimize.py`で実行します。GEPAは同じ質問を何度も尋ねるため、`BrightDataTools`内のキャッシュがここで重要になります。あるGEPAの実行では、994回のツール呼び出しのうち524回がキャッシュによって新しいリクエストなしで回答されました。 以下の4回のGEPA実行では、それぞれ24の学習用質問、24の検証用質問、同じ60のテスト質問を使用しました。 学習用セットフィードバックに期待される回答を含む検証スコア（24問中）指示文にコピーされた学習用の回答テストスコア（60問中）ランダムな24の質問あり21～23355、52ランダムな24の質問なし24、新しい指示文なし0未測定誤答12件、正答12件あり18～21357、55誤答12件、正答12件なし20～213未測定**GEPAは学習用の回答を指示文に書き込むことがあります。**最初の実行による指示文は次のように終わっていました。 ```none Known facts that may be useful for repeated questions: - Galip Ulsoy received the Charles Russ Richards Memorial Award from ASME and Pi Tau Sigma in 2013. ... - The Koshi/Kosi River basin size is commonly listed as about 74,500 km². ... - For the April 25, 2015 Nepal/Gorkha earthquake, a Mw 6.6 aftershock occurred about 34 minutes after the initial Mw 7.8 event ... ``` 各行が1つの学習用質問に答えており、その実行でGEPAが提案した全ての指示文には同様のブロックがありました。期待される回答を隠してもこれを止めることはできませんでした。すると振り返りモデルは、2000年国勢調査におけるオハイオ州ウェストンの世帯数など、エージェント自身の正答した実行から回答を取り込みました。 **学習に使う前に誤答を再実行する。**誤答から構築された検証用セットでは、ベースラインのエージェントは既に24問中18～20問のスコアを出していました。つまり、その12件の誤答のうち少なくとも6件は、再実行すると正解でした。複数回の実行で間違えた質問のみを残してください。 **保留されたテスト質問で結果を測定する。**新しい指示文を提案した全ての実行で検証スコアは上昇しました。60のテスト質問において、最適化されたプログラムは52～57点、Bright Dataツールを使用したベースラインのエージェントは54～58点でした。GEPAの指示文変更はスコアを測定可能なほど高めることはなく、最初の実行だけで`gpt-5.4-mini`エージェントと`gpt-5.5`振り返りモデルのモデル呼び出しコストが3.31ドルかかりました。 **指示文にコピーされた回答がないか確認する。**`research.py`内の`leaked_answers`は、指示文が繰り返している4文字以上の参照用回答を検出します。これらの実行では、コピーされた回答の大部分を検出できましたが、文字列一致では短い数字や言い換えられた回答を検出できません。指示文全体を読み、特定の学習用質問に関する行があれば削除してください。 ## エラー分析でツールの修正点を見つける GEPAは指示文のみを書き換えるため、ツール自体の問題を修正することはできません。誤答に対するエラー分析は、これらの問題を見つけます。ここでは、誤答は日付と数字に集中していました。3回の実行、519の採点済み質問を通して、エージェントは日付と数字に関する質問の約15%を間違え、人物や場所に関する質問では7%でした。 それらの実行では、エージェント自身の`max_chars`設定によって各ページが20,000文字に制限されていました。62件の誤答のうち24件で、エージェントはその制限より長いページを開いていました。全文を再度取得すると、それらのページのうち6件で、制限を超えた箇所に答えがありました。 エージェントが開いたページ答えが始まる文字位置（エージェントの制限は20,000）エージェントの回答正しい回答Wikipedia、Equinox38,604equinoxequiluxWikipedia、ChromeOS24,709October 2011August 2011Wikipedia、Cannibal Corpse25,566March 1993February 1993Aseprite release notes59,987v1.2.22v1.3-rc8Baseball-Reference、Tim Wakefield65,6632.2512.27RuPaul’s Drag Race Wiki、Season 199,250“I Will Survive”“Stronger”全6件において、エージェントは正しい回答が含まれているページを引用していました。引用は、エージェントがどこを見たかを示すものであり、そのページに何が書いてあるかを示すものではありません。 ラッパークラスである`BrightDataTools`では、`start`が次の20,000文字を返し、`find`は単語やフレーズの一致箇所ごとに約3,000文字の最大6つの文章を返します。切り詰められたページの末尾にある注記は、エージェントに両方の引数について伝えます。ページング機能だけでは十分ではありませんでした。`start`のみを使った別のテストでは、エージェントが18回の実行中3回しかページの次の部分を要求しませんでした。 各セットアップは質問ごとに8回実行しました。 セットアップ（同じ6つの長いページの質問）正解数（48回中）実行あたりのプロンプトトークン数`gpt-5.4-mini`のコスト（全48回実行）文字数制限のみ2423,0230.52ドル`start`と`find`を使った文字数制限3311,9080.35ドルエージェントは48回の実行のうち28回で`find`を呼び出しました。正答率の向上は有望ですが、バージョンあたり48回の実行では統計的にはまだ有意ではありません（p ≈ 0.10）。 情報源をユーザーに示す前に確認してください。エージェントは開いたことのない検索結果をよく引用していました。ローカルでの実行では、引用したURLのうち開いたものは4分の1未満でした。情報源を確認するには、`find`と回答の特徴的な部分を使って`BrightDataTools`のスクレイピングツールを呼び出してください。 ## MLflowで全てのツール呼び出しをトレースする トレースは、エージェントが実行した全ての検索と読み取った全てのページを表示します。これにより、エージェントが各回答をどのように生成したかを確認し、誤りの原因となったステップを素早く見つけることができます。[MLflow DSPy連携](https://mlflow.org/docs/3.17.1/genai/tracing/integrations/listing/dspy)は、各モジュール、モデル呼び出し、ツール呼び出しをスパンとして記録します。テスト済みのMLflowリリースをインストールします。 ```none uv pip install "mlflow==3.17.1" ``` 次に、エージェントが実行される前に、これらの行を`agent.py`に追加します。 ```none import mlflow mlflow.set_tracking_uri("sqlite:///mlflow.db") mlflow.dspy.autolog() ``` MLflow 3.17にはデータベースバックエンドが必要なため、トラッキングURIはローカルのSQLiteファイルに設定されています。トレースを表示するには`mlflow ui --backend-store-uri sqlite:///mlflow.db`を実行してください。このスパンツリーは1つの質問から得られたものです。 ```none CHAIN ReActV2.forward LLM Predict.forward x6 CHAT_MODEL LM.__call__ x6 PARSER ChatAdapter.format x6 TOOL Tool.search_engine x4 TOOL Tool.scrape_as_markdown x1 TOOL Tool.submit x1 ``` 各`TOOL`スパンは、モデルが送信した引数と、Bright Data MCPサーバーが返したテキストを保存します。MLflow UIで最初の検索スパンを選択すると、その呼び出しに関する両方が表示されます。 ![左側のナビゲーションの「Traces」から開いた、単一のDSPyエージェント実行に対するMLflow 3.17のトレース表示。スパンツリーには、ネストされたPredict.forward、ChatAdapter.format、LM.__call__、Tool.search_engineの各スパンを伴うReActV2.forwardが一覧表示されており、最初のTool.search_engineスパンが選択されている。詳細ペインには、入力のquery、engine google、空のcursor、geo_location usが表示され、その後にリンク、タイトル、説明を伴うオーガニック検索結果が表示されている。](https://media.brightdata.com/2026/10/rkm-IIm3fx.png)*誤答について、この表示はエージェントのクエリが正しい回答を含むページを見つけたかどうかを示します。* ## まとめ Bright Data MCPサーバーを使うことで、このDSPyエージェントは60のベンチマーク質問のうち54～58問に正しく回答し、14問から向上し、ナレッジカットオフ後の出来事に関する8問全てにも正解しました。これを評価・最適化するには、`dspy.Evaluate`とGEPAがワーカースレッドからツールを呼び出すため、MCPセッションを専用のイベントループ上に保ってください。ここでの誤答6件は、エージェント自身の文字数制限を超えた箇所に答えがあり、`find`によってエージェントはページ全体を検索できるため、最適化する前に誤答を確認してください。GEPAは、誤答が指示文で修正できる共通の原因を共有している場合に最も役立ち、コピーされた回答がないか出力を確認する必要があります。無料のBright Dataアカウントを作成し、モデルが記憶からは答えられない質問で`agent.py`を実行し、自分自身の質問で`evaluate.py`を実行してみてください。 ## よくある質問 ### DSPyはMCPサーバーをサポートしていますか？ はい。`dspy.Tool.from_mcp_tool`とMCP Python SDKのクライアントを使うことで、DSPyはBright Data MCPサーバーなどのMCPサーバーのツールをDSPyツールに変換します。`pip install "dspy[mcp]"`でMCPエクストラをインストールしてください。DSPy 3.4.1では、MCPツールは非同期であるため、エージェントは`acall`で呼び出してください。ワーカースレッドでは、バックグラウンドのイベントループ上で同期ラッパーを使用してください。 ### DSPyはMCPクライアントですか？ いいえ、それ単体ではありません。DSPyはMCP Python SDKのクライアントを使用し、`dspy.Tool.from_mcp_tool`がそのクライアントのツールをDSPyツールに変換します。DSPyはエージェントプログラムを構築・最適化し、MCPはエージェントをBright Data MCPサーバーなどのサーバーに接続します。 ### DSPyエージェントにウェブ検索を追加するにはどうすればよいですか？ Bright Data MCPサーバーに接続し、その検索・スクレイピングツールを、DSPy 3.5以降`dspy.ReAct`が使用する実装である`dspy.ReActV2`に渡します。エージェントはGoogleで検索し、ページをMarkdownとして読み取り、情報源のURLを返します。検索結果をJSONとして返すBright Data[SERP API](/products/serp-api)を呼び出すPython関数も、DSPyツールとして機能します。 ### どのウェブ検索ツールがDSPyで動作しますか？ Bright Data MCPサーバーは、単一の接続を通じてDSPyエージェントにGoogleの検索結果と完全なページをMarkdownとして提供し、サーバーがボット対策を処理します。60のSimpleQA Verified質問において、`gpt-5.4-mini`はこれを使うと54～58問正解し、使わない場合の14問から向上しました。他の検索ツールも動作し、[AIエージェント向け検索ツールの比較](/blog/ai/best-search-tools-for-ai-agents)でそれらを取り上げています。 ### DSPy ReActV2とは何ですか？ `dspy.ReActV2`は、ネイティブのツール呼び出しに基づいて構築された実験的なDSPyエージェントモジュールです。構造化された履歴を保持し、1ターンで複数のツール呼び出しを行うことができ、`submit`ツールで終了します。DSPyのドキュメントには、DSPy 3.5以降`dspy.ReAct`が使用する実装として記載されており、3.6ではReActV2という名称は削除されています。 ### DSPyは今でも使う価値がありますか？ はい、メトリクスで評価できるプログラムにおいては有用です。DSPyはタスクをプロンプトから分離して保持するため、モデルを切り替え、メトリクスとGEPAなどのオプティマイザーで再最適化できます。デフォルト設定では、GEPAはプログラムが一部の学習用質問にまだ正しく答えられていない場合にのみ変更を提案するため、まずプログラムを評価してください。 ### DSPyは本番環境向けですか？ はい、バージョンを固定し、ツール呼び出しを堅牢にすれば可能です。`dspy`およびMCPサーバーを固定し、ツール出力のサイズを制限し、エージェントがその制限を超えてテキストを検索できるようにし、ツール呼び出しに再試行、タイムアウト、エラー数を追加し、MLflowで実行をトレースしてください。ReActV2などの実験的なモジュールはマイナーリリース間で変更される可能性があります。 ### DSPyはプロンプト最適化に使用できますか？ はい。GEPAなどのDSPyオプティマイザーは、評価済みの例に基づいてプログラムの指示文を書き換えます。GEPAは評価済みの実行とその書面でのフィードバックをレビューし、新しい指示文を提案し、その後、検証用セットで最も良い成績を出すものを保持します。GEPAは学習用の回答を指示文にコピーすることがあるため、デプロイする前に結果をレビューしてください。 お問い合わせ無料トライアル![google social icon](/wp-content/themes/brightdata/assets/images/ic_google.svg) 目次 日本企業向けデータソリューション 日本のトップ企業が信頼するスケーラブルなウェブデータプラットフォーム。日本語24時間サポート、GDPR・個人情報保護法準拠、日本企業20社以上の導入実績。 [詳細を見る](https://brightdata.jp/solutions/data-solutions-for-japanese-companies "詳細を見る") [ ](https://news.ycombinator.com/submitlink?t=DSPy+MCP%E3%82%B5%E3%83%BC%E3%83%90%E3%83%BC%EF%BC%9ABright+Data%E3%81%A7%E3%82%A6%E3%82%A7%E3%83%96%E8%AA%BF%E6%9F%BB%E3%82%A8%E3%83%BC%E3%82%B8%E3%82%A7%E3%83%B3%E3%83%88%E3%82%92%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B&u=https://brightdata.jp/blog/ai/dspy-with-bright-data) [ ](https://www.linkedin.com/shareArticle?mini=true&title=DSPy+MCP%E3%82%B5%E3%83%BC%E3%83%90%E3%83%BC%EF%BC%9ABright+Data%E3%81%A7%E3%82%A6%E3%82%A7%E3%83%96%E8%AA%BF%E6%9F%BB%E3%82%A8%E3%83%BC%E3%82%B8%E3%82%A7%E3%83%B3%E3%83%88%E3%82%92%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B&url=https://brightdata.jp/blog/ai/dspy-with-bright-data) [ ](http://www.reddit.com/submit?title=DSPy+MCP%E3%82%B5%E3%83%BC%E3%83%90%E3%83%BC%EF%BC%9ABright+Data%E3%81%A7%E3%82%A6%E3%82%A7%E3%83%96%E8%AA%BF%E6%9F%BB%E3%82%A8%E3%83%BC%E3%82%B8%E3%82%A7%E3%83%B3%E3%83%88%E3%82%92%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B&url=https://brightdata.jp/blog/ai/dspy-with-bright-data) ## あなたは下記にもご興味がおありかもしれません [ ![Cherry Studio MCP with Bright Data](https://media.brightdata.jp/2026/10/cherry-studio-mcp-bright-data-web-search-scraping.png) ](https://brightdata.jp/blog/ai/cherry-studio-with-bright-data "Cherry Studio MCPチュートリアル：Bright DataをWeb検索とスクレイピングに追加") [AI ![Satyam Tripathi](https://media.brightdata.jp/2024/09/Satyam-Tripathi-50x50.png) Satyam Tripathi Technical Writer ### Cherry Studio MCPチュートリアル：Bright DataをWeb検索とスクレイピングに追加 Bright DataのMCPサーバーをCherry Studioに接続し、アシスタントごとに有効化し、モデルにライブなウェブ情報源を引用させる方法を学びます。 20-Oct-2026 5 分読 ](https://brightdata.jp/blog/ai/cherry-studio-with-bright-data) [ ![Bright Data CLI vs Bright Data Python SDK](https://media.brightdata.jp/2026/10/bright-data-cli-vs-python-sdk.png) ](https://brightdata.jp/blog/%e6%af%94%e8%bc%83%e3%81%99%e3%82%8b/bright-data-cli-vs-python-sdk "Bright Data CLI対Bright Data Python SDK:ウェブデータに適したツールの選び方") [比較する ![Antonello Zanini](https://media.brightdata.jp/2022/12/Antonello-Zanini-2-50x50.jpg) Antonello Zanini Technical Writer ### Bright Data CLI対Bright Data Python SDK:ウェブデータに適したツールの選び方 Bright DataのCLIはターミナルワークフローとAIエージェントに適しており、Python SDKはパイプラインに適しています。一般的なスクレイピングタスクでの比較をご覧ください。 20-Oct-2026 5 分読 ](https://brightdata.jp/blog/%e6%af%94%e8%bc%83%e3%81%99%e3%82%8b/bright-data-cli-vs-python-sdk) [ ![Microsoft Agent Framework MCP Server](https://media.brightdata.jp/2026/10/microsoft-agent-framework-mcp-server.png) ](https://brightdata.jp/blog/ai/microsoft-agent-framework-with-web-mcp "Microsoft Agent Framework MCPサーバー：Bright Dataでライブウェブ検索とスクレイピングを追加") [AI ![Satyam Tripathi](https://media.brightdata.jp/2024/09/Satyam-Tripathi-50x50.png) Satyam Tripathi Technical Writer ### Microsoft Agent Framework MCPサーバー：Bright Dataでライブウェブ検索とスクレイピングを追加 Microsoft Agent Frameworkのエージェントに、Bright Data MCPを通じてライブウェブ検索とスクレイピング機能を追加する方法を、PythonとC#の両方で解説します。 20-Oct-2026 12 分読 ](https://brightdata.jp/blog/ai/microsoft-agent-framework-with-web-mcp)