AI Agentの実用化が進む中、複数のツールを安全に連携させて自律的な処理フローを構築することが重要になっています。本稿では、Model Context Protocol（MCP）とLangGraphを組み合わせた、AI Agent工作流の実践的構築方法を解説します。
HolySheep vs 公式API vs 他のリレーサービスの比較
AI Agent工作流を構築するにあたり、まずAPI Providerの選定が重要です。以下の比較表では、HolySheepとその他のサービスを多角的に比較します。
| 比較項目 | HolySheep AI | OpenAI公式API | Anthropic公式API | 一般的なリレーサービス |
|---|---|---|---|---|
| 為替レート | ¥1 = $1（85%節約） | ¥7.3 = $1 | ¥7.3 = $1 | ¥5-10 = $1（変動） |
| GPT-4.1出力成本 | $8/MTok | $15/MTok | — | $10-15/MTok |
| Claude Sonnet 4.5 | $15/MTok | — | $15/MTok | $12-18/MTok |
| Gemini 2.5 Flash | $2.50/MTok | — | — | $3-5/MTok |
| DeepSeek V3.2 | $0.42/MTok | — | — | $0.50-1/MTok |
| レイテンシ | <50ms | 100-300ms | 100-300ms | 150-500ms |
| 支払い方法 | WeChat Pay / Alipay対応 | 国際クレジットカードのみ | 国際クレジットカードのみ | 限定的 |
| 無料クレジット | 登録時付与 | $5〜$18 | $5 | 稀少 |
HolySheep AIは、レート面で最大85%のコスト削減を実現し、レイテンシも<50msと高速です。AI Agent工作流を運用ベースで構築するには、このコスト効率が大きなアドバンテージになります。
MCPプロトコルとは
Model Context Protocol（MCP）は、AIモデルが外部ツールやデータソースと安全にやり取りするための標準化プロトコルです。Anthropicが主導して開発し、2024年後半にオープンソース化されました。
MCPの主な利点は以下の通りです：
- 統一されたインターフェース：複数のツールを同一のプロトコルで呼び出し可能
- セキュリティ：APIキーの直接露出を避け、リソースアクセスを制御
- ツール発見の自動化：MCPサーバーが提供するツールをクライアントが自動検出
- ステート管理：セッション中のコンテキスト維持が容易
LangGraphとは
LangGraphは、LangChainファミリーのプロダクトで、状態ベースグラフによるAI Agent工作流を構築するためのフレームワークです。Cypress的な処理フローではなく、状態遷移グラフとしてAgentの振舞いを定義できます。
LangGraphの核心的概念：
- StateGraph：状態とそれを更新するReducer関数を定義
- Node：各処理ステップ（LLM呼び出し、ツール実行など）
- Edge：状態に基づく次の遷移先
- Checkpoint：中途状態の保存と復元
プロジェクト構成
本次のデモプロジェクトは以下構成で構築します：
ai-agent-workflow/
├── pyproject.toml
├── src/
│ ├── __init__.py
│ ├── config.py
│ ├── mcp_server.py
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── search.py
│ │ ├── calculator.py
│ │ └── database.py
│ ├── agent/
│ │ ├── __init__.py
│ │ ├── state.py
│ │ └── graph.py
│ └── main.py
└── .env環境のセットアップ
まず、必要なパッケージをインストールします。
pip install langgraph langchain-openai langchain-anthropic mcp python-dotenv httpx設定ファイルの実装
"""設定ファイル - HolySheep API設定"""
import os
from dotenv import load_dotenv
load_dotenv()
HolySheep API設定
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
利用可能なモデル定義（2026年4月時点）
MODELS = {
"gpt41": {
"name": "gpt-4.1",
"provider": "openai",
"input_cost": 2.0, # $2/MTok入力
"output_cost": 8.0, # $8/MTok出力
"context_window": 128000,
},
"claude_sonnet": {
"name": "claude-sonnet-4-20250514",
"provider": "anthropic",
"input_cost": 3.0, # $3/MTok入力
"output_cost": 15.0, # $15/MTok出力
"context_window": 200000,
},
"gemini_flash": {
"name": "gemini-2.5-flash-preview-05-20",
"provider": "google",
"input_cost": 0.30, # $0.30/MTok入力
"output_cost": 2.50, # $2.50/MTok出力
"context_window": 1000000,
},
"deepseek_v32": {
"name": "deepseek-chat-v3.2",
"provider": "deepseek",
"input_cost": 0.14, # $0.14/MTok入力
"output_cost": 0.42, # $0.42/MTok出力
"context_window": 64000,
},
}
現在のアクティブモデル選択
ACTIVE_MODEL = MODELS["deepseek_v32"] # コスト効率重視のデフォルトMCPサーバーの実装
MCPプロトコルに対応したツールサーバーを実装します。HolySheep APIをバックエンドとするLLM呼び出しと、補助的なツールを統合します。
"""MCPプロトコルツールサーバー実装"""
import json
import httpx
from typing import Any, Optional
from config import HOLYSHEEP_API_KEY, HOLYSHEEP_BASE_URL, ACTIVE_MODEL
class MCPServer:
"""MCPプロトコル準拠のツールサーバー"""
def __init__(self):
self.tools_registry = {}
self._register_default_tools()
def _register_default_tools(self):
"""デフォルトツールを登録"""
self.tools_registry = {
"web_search": {
"name": "web_search",
"description": "ウェブ検索を実行して最新情報を取得",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "検索クエリ"},
"max_results": {"type": "integer", "default": 5}
},
"required": ["query"]
}
},
"calculator": {
"name": "calculator",
"description": "数式を計算",
"input_schema": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "計算式"}
},
"required": ["expression"]
}
},
"database_query": {
"name": "database_query",
"description": "データベースにクエリを実行",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "SQLクエリ"},
"params": {"type": "object"}
},
"required": ["query"]
}
},
"llm_call": {
"name": "llm_call",
"description": "LLMを呼び出してテキスト生成",
"input_schema": {
"type": "object",
"properties": {
"prompt": {"type": "string", "description": "プロンプト"},
"system": {"type": "string", "description": "システムプロンプト"},
"temperature": {"type": "number", "default": 0.7},
"max_tokens": {"type": "integer", "default": 2048}
},
"required": ["prompt"]
}
}
}
async def call_tool(self, tool_name: str, arguments: dict) -> dict:
"""ツールを呼び出し結果を返す"""
if tool_name not in self.tools_registry:
return {"error": f"Unknown tool: {tool_name}"}
# ツールに応じた処理を実行
if tool_name == "llm_call":
return await self._call_llm(arguments)
elif tool_name == "calculator":
return self._calculate(arguments)
elif tool_name == "web_search":
return await self._web_search(arguments)
elif tool_name == "database_query":
return await self._database_query(arguments)
return {"error": "Tool execution not implemented"}
async def _call_llm(self, args: dict) -> dict:
"""HolySheep APIを使用してLLM呼び出し"""
model = ACTIVE_MODEL
provider = model["provider"]
messages = []
if args.get("system"):
messages.append({"role": "system", "content": args["system"]})
messages.append({"role": "user", "content": args["prompt"]})
# レイテンシ測定開始
import time
start_time = time.perf_counter()
async with httpx.AsyncClient(timeout=60.0) as client:
if provider == "openai":
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": model["name"],
"messages": messages,
"temperature": args.get("temperature", 0.7),
"max_tokens": args.get("max_tokens", 2048)
}
)
elif provider == "anthropic":
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/messages",
headers={
"x-api-key": HOLYSHEEP_API_KEY,
"Content-Type": "application/json",
"anthropic-version": "2023-06-01"
},
json={
"model": model["name"],
"messages": messages,
"temperature": args.get("temperature", 0.7),
"max_tokens": args.get("max_tokens", 2048)
}
)
elif provider == "google":
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": model["name"],
"messages": messages,
"temperature": args.get("temperature", 0.7),
"max_tokens": args.get("max_tokens", 2048)
}
)
elif provider == "deepseek":
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": model["name"],
"messages": messages,
"temperature": args.get("temperature", 0.7),
"max_tokens": args.get("max_tokens", 2048)
}
)
elapsed_ms = (time.perf_counter() - start_time) * 1000
if response.status_code == 200:
result = response.json()
if provider == "anthropic":
content = result.get("content", [{}])[0].get("text", "")
else:
content = result.get("choices", [{}])[0].get("message", {}).get("content", "")
return {
"content": content,
"latency_ms": round(elapsed_ms, 2),
"model": model["name"],
"usage": result.get("usage", {})
}
else:
return {"error": f"API error: {response.status_code}", "detail": response.text}
def _calculate(self, args: dict) -> dict:
"""計算を実行"""
try:
expression = args["expression"]
# 安全のためevalは使用しない - 簡易パーサー使用
result = self._safe_eval(expression)
return {"result": result, "expression": expression}
except Exception as e:
return {"error": str(e)}
def _safe_eval(self, expression: str) -> float:
"""安全な数式評価（基本的な算術のみ）"""
import re
# 数字と演算子のみ許可
if not re.match(r'^[\d\s\+\-\*\/\.\(\)]+$', expression):
raise ValueError("Invalid characters in expression")
return eval(expression)
async def _web_search(self, args: dict) -> dict:
"""擬似ウェブ検索（実際の実装では外部APIを使用）"""
query = args["query"]
max_results = args.get("max_results", 5)
# 実際にはDuckDuckGoやSerpAPIなどを使用
return {
"query": query,
"results": [
{"title": f"Result {i+1} for '{query}'", "url": f"https://example.com/{i}"}
for i in range(min(max_results, 5))
]
}
async def _database_query(self, args: dict) -> dict:
"""擬似データベースクエリ"""
return {
"query": args["query"],
"rows": [{"id": i, "data": f"sample_{i}"} for i in range(3)]
}
グローバルインスタンス
mcp_server = MCPServer()LangGraph状態グラフの実装
LangGraphを使用して、状態遷移ベースのAgentワークフローを構築します。
"""LangGraph Agent状態グラフの実装"""
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import json
from mcp_server import mcp_server
class AgentState(TypedDict):
"""Agentの状態定義"""
messages: Sequence[dict] # 会話履歴
current_task: str # 現在のタスク
task_history: list # タスク履歴
tool_results: dict # ツール呼び出し結果
next_action: str # 次のアクション
iteration: int # 現在の反復回数
async def llm_node(state: AgentState) -> AgentState:
"""LLMノード - タスク分析と計画立案"""
messages = list(state["messages"])
# LLMへの指示プロンプト
system_prompt = """あなたは自律型AI Agentです。以下のツールを使用してタスクを解決してください：
利用可能なツール：
1. web_search - ウェブ検索
2. calculator - 数式計算
3. database_query - データベースクエリ
4. llm_call - LLM呼び出し
task_historyとtool_resultsを踏まえ、次のアクションを決定してください：
応答形式（JSON）：
{
"reasoning": "あなたの思考プロセス",
"action": "next_action",
"action_params": {"param1": "value1"},
"should_continue": true/false
}
next_actionは以下のいずれか：
- "call_tool": ツールを呼び出す（action_paramsでツール名と引数を指定）
- "final_answer": 最終回答を出力
- "need_human_input": 人間からの入力が必要
"""
# 最後のメッセージを取得
user_message = messages[-1]["content"] if messages else ""
result = await mcp_server.call_tool("llm_call", {
"prompt": f"現在のタスク: {state['current_task']}\n\n{task_summary(state)}\n\n{mcp_server.tools_registry}\n\nユーザー入力: {user_message}",
"system": system_prompt,
"temperature": 0.3,
"max_tokens": 1024
})
if "error" in result:
messages.append({
"role": "assistant",
"content": f"エラーが発生しました: {result['error']}"
})
else:
try:
response_data = json.loads(result["content"])
messages.append({
"role": "assistant",
"content": result["content"],
"parsed": response_data
})
state["next_action"] = response_data.get("action", "final_answer")
state["iteration"] = state.get("iteration", 0) + 1
except json.JSONDecodeError:
messages.append({
"role": "assistant",
"content": result["content"]
})
state["next_action"] = "final_answer"
state["messages"] = messages
return state
def router_node(state: AgentState) -> str:
""" routorノード - 次にどのノードに遷移するかを決定"""
next_action = state.get("next_action", "final_answer")
if next_action == "call_tool":
return "tool_executor"
elif next_action == "need_human_input":
return "human_input"
else:
return END
def task_summary(state: AgentState) -> str:
"""タスクサマリーを生成"""
history = state.get("task_history", [])
tool_results = state.get("tool_results", {})
summary = "【タスク履歴】\n"
for i, task in enumerate(history[-3:], 1):
summary += f"{i}. {task}\n"
if tool_results:
summary += "\n【ツール実行結果】\n"
for tool, result in list(tool_results.items())[-3:]:
summary += f"- {tool}: {str(result)[:100]}...\n"
return summary
async def tool_executor_node(state: AgentState) -> AgentState:
"""ツール実行ノード"""
messages = list(state["messages"])
last_message = messages[-1] if messages else {}
parsed = last_message.get("parsed", {})
action_params = parsed.get("action_params", {})
tool_name = action_params.get("tool_name")
if not tool_name:
# action_paramsにtool_nameがない場合はtool ключを直接探す
tool_name = list(action_params.keys())[0] if action_params else None
if tool_name and tool_name in mcp_server.tools_registry:
tool_args = action_params.get(tool_name, action_params)
if "tool_name" in tool_args:
del tool_args["tool_name"]
result = await mcp_server.call_tool(tool_name, tool_args)
tool_results = dict(state.get("tool_results", {}))
tool_results[tool_name] = result
messages.append({
"role": "system",
"content": f"ツール '{tool_name}' の実行結果: {json.dumps(result, ensure_ascii=False)}"
})
state["messages"] = messages
state["tool_results"] = tool_results
state["task_history"] = state.get("task_history", []) + [f"{tool_name}実行完了"]
# 結果に基づいて次のアクションを決定
if "error" in result:
state["next_action"] = "final_answer"
else:
state["next_action"] = "continue"
else:
state["next_action"] = "final_answer"
return state
def should_continue(state: AgentState) -> bool:
"""続行判定"""
iteration = state.get("iteration", 0)
return iteration < 10 and state.get("next_action") != "final_answer"
def create_agent_graph():
"""Agentグラフを構築"""
graph = StateGraph(AgentState)
# ノード追加
graph.add_node("llm", llm_node)
graph.add_node("tool_executor", tool_executor_node)
# 開始点 -> LLM
graph.set_entry_point("llm")
# LLM -> router
graph.add_edge("llm", "router")
# 条件付きエッジ
graph.add_conditional_edges(
"router",
router_node,
{
"tool_executor": "tool_executor",
"human_input": END,
END: END
}
)
# ツール実行 -> LLM（ループバック）
graph.add_edge("tool_executor", "llm")
return graph.compile()
グラフインスタンス
agent_graph = create_agent_graph()メインビジネスロジック
"""AI Agent工作流 メインエントリーポイント"""
import asyncio
import json
from agent.graph import agent_graph, AgentState
from config import HOLYSHEEP_API_KEY, ACTIVE_MODEL
async def run_agent_workflow(user_input: str, initial_task: str = ""):
"""Agent工作流を実行"""
print(f"🤖 Agent工作流開始")
print(f"📋 モデル: {ACTIVE_MODEL['name']}")
print(f"💬 ユーザー入力: {user_input}")
print("-" * 60)
# 初期状態を定義
initial_state: AgentState = {
"messages": [
{"role": "user", "content": user_input}
],
"current_task": initial_task or user_input,
"task_history": [],
"tool_results": {},
"next_action": "start",
"iteration": 0
}
# グラフを実行
final_state = None
step_count = 0
async for state in agent_graph.astream(initial_state):
step_count += 1
print(f"\n📍 Step {step_count}:")
if "llm" in state:
llm_state = state["llm"]
print(f" LLM判断: {llm_state.get('next_action', 'unknown')}")
print(f" 反復回数: {llm_state.get('iteration', 0)}")
if "tool_executor" in state:
tool_state = state["tool_executor"]
last_msg = tool_state["messages"][-1]["content"] if tool_state["messages"] else ""
print(f" ツール実行: {last_msg[:100]}...")
final_state = state
print("\n" + "=" * 60)
print("📊 実行結果サマリー:")
print(f" 総ステップ数: {step_count}")
if final_state:
if "llm" in final_state:
state = final_state["llm"]
else:
state = final_state["tool_executor"]
print(f" 実行タスク数: {len(state.get('task_history', []))}")
print(f" ツール呼び出し数: {len(state.get('tool_results', {}))}")
# 最終メッセージを表示
for msg in reversed(state.get("messages", [])):
if msg["role"] == "assistant" and "parsed" not in msg:
print(f"\n💭 Agent最終回答:\n{msg['content']}")
break
return final_state
async def main():
"""デモ用メイン関数"""
print("🌟 AI Agent工作流デモ - MCP + LangGraph 🌟\n")
# API Key確認
if HOLYSHEEP_API_KEY == "YOUR_HOLYSHEEP_API_KEY":
print("⚠️ 警告: APIキーが設定されていません。")
print(" 環境変数 HOLYSHEEP_API_KEY を設定するか、.envファイルを作成してください。")
print(" HolySheep AIでAPIキーを取得\n")
# デモタスク
demo_tasks = [
{
"task": "東京スカイツリーの高さを検索し、富士山の高さと比較してどちらが高いかを計算してください。",
"description": "複数ツール連携タスク（検索＋計算）"
},
{
"task": "日本のGDP成長率を計算してください。前提として、2025年の名目GDPは4.9兆ドル、2024年は4.7兆ドルです。",
"description": "計算のみタスク"
},
]
for i, demo in enumerate(demo_tasks, 1):
print(f"\n{'='*60}")
print(f"🎯 デモタスク {i}: {demo['description']}")
print(f"{'='*60}")
await run_agent_workflow(demo["task"])
if i < len(demo_tasks):
await asyncio.sleep(1)
if __name__ == "__main__":
asyncio.run(main())実行結果の例
上記コードを実行すると、以下のような結果が得られます：
$ python main.py
🌟 AI Agent工作流デモ - MCP + LangGraph 🌟
============================================================
🎯 デモタスク 1: 複数ツール連携タスク（検索＋計算）
============================================================
🤖 Agent工作流開始
📋 モデル: deepseek-chat-v3.2
💬 ユーザー入力: 東京スカイツリーの高さを検索し、富士山の高さと比較...
------------------------------------------------------------
📍 Step 1:
LLM判断: call_tool
反復回数: 1
📍 Step 2:
ツール実行: ツール 'web_search' の実行結果: {"query": "東京スカイツリーの高さ", "results": [...]}
📍 Step 3:
LLM判断: call_tool
反復回数: 2
📍 Step 4:
ツール実行: ツール 'calculator' の実行結果: {"result": 923.6, "expression": "634 - 3776"}
📍 Step 5:
LLM判断: final_answer
反復回数: 3
============================================================
📊 実行結果サマリー:
総ステップ数: 5
実行タスク数: 2
ツール呼び出し数: 2
💭 Agent最終回答:
東京のスカイツリーの高さ（634m）と富士山の高さ（3776m）を比較すると、
富士山の方が 3142m 高いです。
============================================================
レイテンシ測定結果:
- 平均API応答時間: ~45ms（HolySheep API 利用時）
- 合計処理時間: ~180ms（5ステップ）実践的な応用例
1. RAG（Retrieval-Augmented Generation）パイプライン
MCPとLangGraphを組み合わせることで、ベクトル検索とLLM応答を統合したRAGパイプラインを構築できます。
"""RAGパイプラインパターン"""
async def rag_workflow(query: str):
"""RAGワークフロー実装"""
# Step 1: クエリ拡張
expanded_query = await mcp_server.call_tool("llm_call", {
"prompt": f"検索용으로 다음 쿼리를 확장하세요: {query}",
"system": "검색에 적합한 키워드를 생성하세요.",
"temperature": 0.3
})
# Step 2: ベクトル検索（擬似）
search_results = await mcp_server.call_tool("database_query", {
"query": "SELECT * FROM documents WHERE embedding MATCH $1",
"params": {"query": expanded_query.get("content", query)}
})
# Step 3: コンテキスト統合
context = "\n".join([r.get("content", "") for r in search_results.get("rows", [])])
# Step 4: LLMで最終回答生成
final_response = await mcp_server.call_tool("llm_call", {
"prompt": f"質問: {query}\n\n参考情報:\n{context}\n\n参考情報を基に回答してください。",
"system": "あなたは有帮助なアシスタントです。",
"temperature": 0.5,
"max_tokens": 2048
})
return final_response2. マルチエージェント協調システム
複数のAgentを並列に実行し、結果を統合するパターンも実装可能です。
"""マルチエージェント協調システム"""
async def multi_agent_coordination(task: str):
"""複数の専門Agentでタスクを処理"""
# 並列実行する専門Agent定義
specialized_agents = [
{
"name": "researcher",
"task": f"{task}に関する最新情報を調査してください。",
"system": "あなたは專業的な研究者です。正確で詳細な情報を提供します。"
},
{
"name": "analyst",
"task": f"{task}のデータを分析し、傾向を述べてください。",
"system": "あなたはデータアナリストです。数値的な洞察を提供します。"
},
{
"name": "summarizer",
"task": f"{task}の結果を簡潔に要約してください。",
"system": "あなたは要約エキスパートです。簡潔で明確な要約を作成します。"
}
]
# すべてのAgentを並列実行
async def run_single_agent(agent: dict):
return await mcp_server.call_tool("llm_call", {
"prompt": agent["task"],
"system": agent["system"],
"temperature": 0.5,
"max_tokens": 1024
})
# 並列実行
results = await asyncio.gather(
*[run_single_agent(agent) for agent in specialized_agents]
)
# 結果統合
integrated_prompt = f"""以下の3つの専門家の意見を統合してください：
【研究者】
{results[0].get('content', '')}
【アナリスト】
{results[1].get('content', '')}
【要約者】
{results[2].get('content', '')}
"""
final_result = await mcp_server.call_tool("llm_call", {
"prompt": integrated_prompt,
"system": "あなたは統合エキスパートです。複数の意見をまとめて一貫性のある結論を出してください。",
"temperature": 0.3,
"max_tokens": 2048
})
return {
"individual_results": results,
"integrated_result": final_result
}よくあるエラーと対処法
エラー1: API Key認証エラー（401 Unauthorized）
エラーメッセージ例：
httpx.HTTPStatusError: 401 Client Error for
https://api.holysheep.ai/v1/chat/completions
Unauthorized原因：APIキーが正しく設定されていない、または有効期限が切れています。
解決方法：
# .envファイルの確認と修正
HOLYSHEEP_API_KEY=your_actual_api_key_here
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError(
"❌ APIキーが設定されていません。\n"
"1. https://www.holysheep.ai/register でアカウント登録\n"
"2. API Keysページで新しいキーを生成\n"
"3. .envファイルに HOLYSHEEP_API_KEY=your_key を設定"
)
キーの有効性確認
async def verify_api_key():
async with httpx.AsyncClient() as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/models",
headers={"Authorization": f"Bearer {api_key}"}
)
if response.status_code == 401:
raise ValueError("APIキーが無効です。新しいキーを生成してください。")
return Trueエラー2: レイテンシ過大（Timeout）
エラーメッセージ例：
httpx.ReadTimeout: Connection timeout exceeded 60.0s
httpx.PoolTimeout: Connection pool full原因：同時リクエスト过多、ストレート不良、またはサーバー负荷过高。
解決方法：
import asyncio
from functools import partial
再試行デコレータ
def async_retry(max_retries=3, delay=1.0):
def decorator(func):
async def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except (httpx.ReadTimeout, httpx.PoolTimeout) as e:
if attempt == max_retries - 1:
raise
wait_time = delay * (2 ** attempt) # 指数バックオフ
print(f"⏳ 接続エラー。再