第09回学習メモ: Research Agent（タスク分解型エージェント）¶

全体像¶

第08回で「LangGraph による状態遷移グラフ」を学んだ流れを汲み、今回は「Plan-and-Execute」型の調査エージェントを組み立てる回。ユーザーが投げた1つのクエリ（例:「生成AIスタートアップの最新動向について調査してください」）に対して、

plan: LLM が調査をサブタスクに分解する（複数の検索 + 最後のレポート執筆）
route: 依存関係を解決しながら、次に実行できるタスクを1つ選ぶ
search / write: タスクの種別に応じて Tavily 検索 or LLM 執筆を実行
route に戻る: 完了済みタスクを記録し、次のタスクを選び直す
全タスクが終わったら END

という流れを LangGraph の StateGraph として書く。要点は以下の3つ。

タスクの DAG（有向非巡回グラフ）を LLM に生成させる — related_ids で依存関係を表現
route ノードがディスパッチャとして機能 — Python の while ループの代わりに、グラフ自身がループする
Annotated[Sequence[X], add] で状態を累積 — 各ノードの戻り値が自動でリストに連結される

動的処理フロー¶

flowchart TD
    Start([START]) --> Plan[plan<br/>クエリをタスク群に分解]
    Plan --> Route[route<br/>次に実行できるタスクを探す]
    Route -->|next_node = search| Search[search<br/>Tavily で検索]
    Route -->|next_node = write| Write[write<br/>LLM でレポート執筆]
    Route -->|next_node = end| End([END])
    Search --> Route
    Write --> Route

第08回の「2ノードを往復してメッセージ数で終了」より一段抽象度が高く、「次に何をするか」を毎ステップ動的に決めているのがポイント。実質的には Plan-and-Execute エージェントを LangGraph で書き直したもの。

1クエリあたりの実行シーケンス（例）¶

sequenceDiagram
    participant U as User
    participant G as Graph
    participant L as LLM (GPT-4 Turbo)
    participant T as Tavily

    U->>G: "生成AIスタートアップの動向"
    G->>L: plan (system prompt + query)
    L-->>G: tasks = [search#1, search#2, search#3, write#4(related=[1,2,3])]
    loop タスクが残っている間
        G->>G: route (依存解決済みの未完了タスクを1つ選ぶ)
        alt action == search
            G->>T: search(description)
            T-->>G: documents
        else action == write
            G->>L: write(task, 関連 artifacts を結合)
            L-->>G: レポート本文
        end
    end
    G-->>U: final_output（最後の write artifact）

使用ライブラリ・原理¶

LangGraph — `StateGraph` と「状態の累積」¶

第08回の復習＋今回の新要素。

要素	役割
`StateGraph(AgentState)`	TypedDict で定義した状態を引き回すグラフ。各ノードは `state -> dict` の純粋関数
`add_node(name, fn)`	ノード登録。fn の戻り dict が state の該当キーにマージされる
`add_edge(src, dst)`	無条件遷移
`add_conditional_edges(src, router_fn, mapping)`	router_fn が返す文字列を見て分岐先を選ぶ
`compile()`	実行可能なエージェントに変換

今回の肝は AgentState の型定義:

class AgentState(TypedDict):
    query: str
    tasks: list[Task]
    artifacts: Annotated[Sequence[Artifact], add]          # ← 累積
    next_task: Task | None
    next_node: str
    completed_task_ids: Annotated[Sequence[int], add]      # ← 累積

Annotated[Sequence[X], add] の 第2引数 add は operator.add。LangGraph はノードの戻り値をマージするとき、このキーについては「上書き」ではなく「+ 演算子で連結」する
そのため _run_search が {"artifacts": [new_artifact], "completed_task_ids": [current_task.id]} を返すたびに、リストに追記される
第08回の messages フィールドと同じ仕組み

これを知らずに Annotated を外すと、毎ステップ前回の artifacts が消える挙動になる。LangGraph 特有のハマりどころ。

Pydantic v1（`langchain_core.pydantic_v1`）¶

from langchain_core.pydantic_v1 import BaseModel

LangChain 0.1 系は内部で Pydantic v1 を使っていた頃の名残。LangChain 0.3 以降は素の pydantic (v2) でよい。

役割は 「LLM 出力の構造化バリデーション」+「JsonOutputParser のスキーマ定義」。Task / Tasks を BaseModel で書くことで、

LLM が JSON を返したあと、Tasks(...) に詰めて型チェック
IDE 補完が効く（task.description のように属性アクセス可能）

ができる。本コードでは結局 dict_to_task で手動変換しているので Pydantic の恩恵を活かしきれていない部分もある（後述）。

Tavily — LLM 用に最適化された検索 API¶

通常の Google / Bing 検索 API は「タイトル + URL + 短いスニペット」を返すだけで、RAG に使うには別途スクレイピングが必要だった。Tavily は

1リクエストで複数サイトを並列クロールし、
LLM で関連性をフィルタリング・ランキングし、
本文（raw_content）まで丸ごと返す

ことで「検索 → スクレイピング → クレンジング」を1コールにまとめてくれる。include_raw_content=True で本文も取得している。

response = tavily_client().search(query, max_results=5, include_raw_content=True)
# response["results"] = [{"title": ..., "url": ..., "content": ..., "raw_content": ...}, ...]

`retry` デコレータ + JSON モード¶

llm = ChatOpenAI(model=...).bind(response_format={"type": "json_object"})

@retry(tries=3)
def invoke_chain(query: str) -> list[Task]:
    ...

response_format={"type": "json_object"} は OpenAI の JSON モード。出力が必ず valid JSON になることを保証する（ただしスキーマまでは保証しない）
@retry(tries=3) で JsonOutputParser のパース失敗時に3回までリトライ。LLM が稀に壊れた JSON を返したり、properties キーを忘れたりするケースの保険
現代版なら OpenAI Structured Outputs（response_format={"type": "json_schema", ...}）または with_structured_output(Tasks) を使えばリトライ不要に近づく

`@lru_cache` で TavilyClient をシングルトン化¶

@lru_cache
def tavily_client() -> TavilyClient:
    return TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

引数なし関数に @lru_cache を付けると 初回呼び出し時に1回だけ生成して以降はキャッシュを返す = グローバル変数の代わりになるイディオム。テスト時にモックを差し替えやすい。

ファイル別の役割¶

ファイル	役割
`research_agent.py`	エージェント本体。データクラス定義 / プラン関数 / 検索関数 / 書き込み関数 / グラフ定義 / CLI エントリポイントが1ファイルに集約
`prompts/plan_system.prompt`	タスク分解 LLM のシステムプロンプト。tool_definitions / output_format / output_example を含む few-shot 風指示書
`prompts/write_system.prompt`	レポート執筆 LLM のシステムプロンプト。「ジャーナリスティックなトーンで、出典 URL を必ず付けて、日本語で書け」と指示
`prompts/write_user.prompt`	ユーザーメッセージのテンプレート。`{task}` と `{documents}` を埋め込む
`requirements.txt`	langchain 0.1.15 / langgraph 0.0.32 / tavily-python 0.3.3 など2024年4月時点のバージョン固定

プロンプトをファイル化する設計¶

load_prompt(name) で prompts/*.prompt を読み込んでいる。

Python コードとプロンプト本文を分離できる → プロンプトエンジニアリングの試行錯誤がコードの diff を汚さない
多言語対応や A/B テストもファイル差し替えで済む
第05回までは f-string で埋め込んでいたが、ここから「プロンプトはアセット」という意識に切り替わっている

行レベルの工夫¶

`find_next_task` — 依存解決の本体（L89-95）¶

def find_next_task(tasks: list[Task], completed_task_ids: list[int]) -> Task | None:
    for task in tasks:
        if task.id not in completed_task_ids and all(
                related_id in completed_task_ids for related_id in task.related_ids
        ):
            return task
    return None

完了していない かつ 依存タスクが全て完了している タスクを順に探す
DAG のトポロジカルソートを毎回 O(N × M) で愚直にやっている版。タスク数が少ない（〜10件）想定なので問題なし
related_ids が空のタスク（最初の検索群）は all([]) が True を返すので即実行可能

`SearchContent.str` — LLM に渡すフォーマットを意識した整形（L40-44）¶

def __str__(self):
    return "\n\n".join(
        f'"""\ntitle: {item["title"]}\nurl: {item["url"]}\ncontent: {item["raw_content"]}\n"""'
        for item in self.documents
    )

各検索結果を """...""" で囲む = LLM に「ここがブロック境界」と認識させやすい慣習
write プロンプトで「出典 URL を書け」と指示しているので、URL を構造化して渡すのが効く

`_run_write` — 関連 artifacts を結合してから渡す（L188-202）¶

related_artifacts = fetch_artifact(list(state["artifacts"]), current_task.related_ids)
documents = "\n\n".join(str(artifact.content) for artifact in related_artifacts)
report = write(query, documents)

related_ids で参照されている artifact だけを抽出 → 不要な情報で context window を埋めない
__str__ が search/write で分岐していたので、検索結果も中間レポートも同じ書式で結合できる

`_router` — 状態を直接読むだけの薄いラッパー（L215-216）¶

def _router(self, state: AgentState) -> str:
    return state["next_node"]

分岐ロジックは _run_route で完結させ、_router は「state を文字列に射影する」だけ
LangGraph の add_conditional_edges は「router 関数が返した文字列で分岐先を選ぶ」設計なので、状態に next_node を書いておけば router は1行で済むというイディオム
第08回の should_continue (return "continue" if len(messages) < 9 else "end") より柔軟（条件分岐ノード自体を1つの計算ノードとして独立させた形）

`graph.agent.stream(...)` で逐次表示（L246-262）¶

for s in graph.agent.stream(input=initial_state, config={"recursion_limit": 1000}):
    if "plan" in s: ...
    elif "route" in s: ...
    elif "write" in s: ...

stream() は ノードが1つ実行されるたびに {node_name: state_diff} を yield する
invoke() だと最終結果しか取れないが、stream() だと進捗を CLI に流せる
recursion_limit=1000 は LangGraph のデフォルト（25）だとタスク数が多いとき早期に止まるので拡張している

学んだこと（要点）¶

Plan-and-Execute パターンの最小実装: LLM に DAG を返させて、Python 側はトポロジカル順に実行するだけ。「LLM の出力を Python の制御フローに変換する」発想
LangGraph の Annotated[Sequence, add] は state を「累積バッファ」化するマーカー。これがないと毎ステップ上書きされて履歴が消える
route ノードが state machine のディスパッチャ。条件分岐ロジックを node に閉じ込めれば、_router 関数は薄く保てる
JSON 出力の保険を3層で組む: ①response_format={"type": "json_object"} でフォーマット保証 ②JsonOutputParser(pydantic_model=...) でスキーマ寄せ ③@retry(tries=3) でパース失敗時にやり直し
Tavily は LLM 用検索 API。スニペットだけでなく本文（raw_content）を返すので RAG にそのまま投入できる
プロンプトはコードと分離する: prompts/*.prompt に切り出すと試行錯誤の diff が見やすい

ハマりどころ¶

AgentState の Annotated[..., add] を外すと artifact が累積しない（最後の write が空ドキュメントを受け取る）
find_next_task は DAG にサイクルがあると無限ループの素になる。LLM 生成タスクなので「id=3 が related_ids=[3]」のような自己参照を返してくる事故が稀にある（リトライで誤魔化す前提）
final_output = s["write"]["artifacts"][0] は「最後に実行された write ノードの新規 artifact」を捕まえる前提。write が複数ある計画だと最後の write artifact が「最終レポート」とは限らない

実際にやった移植差分（2026-05 時点）¶

連載原典は LangChain 0.1.15 + LangGraph 0.0.32 + Pydantic v1 を使っていたが、本リポジトリでは以下の差分で動く状態に書き換えた。詳細は README.md と research_agent.py 冒頭の docstring を参照。

旧 (2024年4月時点)	移植後 (現リポジトリ)
`langchain==0.1.15`, `langgraph==0.0.32`	`langchain>=0.3,<1.0`, `langgraph>=0.2,<1.0`（`uv run --with` で都度インストール）
`from langchain_core.pydantic_v1 import BaseModel`	素の `from pydantic import BaseModel`（Pydantic v2）
`JsonOutputParser(pydantic_model=Tasks)` + `dict_to_task` 手動変換 + `@retry(tries=3)`	`llm.with_structured_output(Tasks)` 1行。リトライも `dict_to_task` も不要に
`response_format={"type": "json_object"}` を `.bind()`	`with_structured_output` がスキーマ準拠まで保証
`gpt-4-turbo-2024-04-09`	`gpt-4o-mini`（コスト・速度のバランス）
`OPENAI_API_KEY` / `TAVILY_API_KEY` を `.env` から `load_dotenv()` で読む	両方とも 1Password CLI から読む（`op read op://Personal/openAI_API/credential` と `op://Personal/Tavily_API_key/credential`）。`.env` 不要
`Annotated[Sequence[Artifact], add]`	`Annotated[list[Artifact], add]`（`Sequence` 不要、素直に `list`）
`from langgraph.graph import StateGraph, END` + `set_entry_point("plan")`	`START` も明示 import して `add_edge(START, "plan")` で一貫させた
`action: str`	`action: Literal["search", "write"]` で型を絞る（バリデーションが効く）
`initial_state` に未使用キー `"documents": []` が混入	削除して `AgentState` 型に厳密一致
`class Task(BaseModel): id: int; ...` フィールド説明なし	`Field(description=...)` を付与。`with_structured_output` のスキーマ生成時に LLM に渡る情報になるので、説明を書くだけで出力品質が上がる
`TAVILY_MAX_RESULTS=5` × 各 raw_content フル投入	`TAVILY_MAX_RESULTS=3` + `MAX_CONTENT_CHARS=3000` で本文を truncate。gpt-4o-mini の 128k context 制限に収まる（実測: 5検索×5件×フル本文 = 162k トークンで BadRequestError）

`plan` 関数の Before / After¶

Before（旧版・37行）:

def plan(query: str) -> list[Task]:
    def dict_to_task(data: dict) -> list[Task]:
        return [Task(id=item["id"], action=item["action"], ...) for item in data["properties"]]

    @retry(tries=3)
    def invoke_chain(query: str) -> list[Task]:
        llm = ChatOpenAI(model=...).bind(response_format={"type": "json_object"})
        prompt = ChatPromptTemplate.from_messages(...).partial(...)
        chain = prompt | llm | JsonOutputParser(pydantic_model=Tasks) | dict_to_task
        return chain.invoke({"message": query})

    return invoke_chain(query)

After（現版・9行）:

def plan(query: str) -> list[Task]:
    llm = ChatOpenAI(model=LLM_MODEL_NAME, temperature=0)
    structured_llm = llm.with_structured_output(Tasks)
    prompt = ChatPromptTemplate.from_messages(
        [("system", "{system_message}"), ("user", "{message}")]
    ).partial(system_message=load_prompt("plan_system"))
    chain = prompt | structured_llm
    result: Tasks = chain.invoke({"message": query})
    return result.tasks

JsonOutputParser + dict_to_task + @retry の3つが消えた。with_structured_output は内部で OpenAI Structured Outputs（response_format={"type": "json_schema"}）を使うので、LLM 側でスキーマが強制される → 不正な JSON はそもそも返ってこない。

解消済みの不具合¶

~~main() の initial_state に未使用キー "documents": [] が混入~~ → 削除済み
~~langchain==0.1.15 系は現行依存と衝突しがち~~ → uv run --with で隔離環境に最新版を入れる方式に統一
~~write 段階で context length exceeded（162k > 128k）になる~~ → TAVILY_MAX_RESULTS=3 + 本文 3000 文字 truncate で対応。実測で 5検索 × 5件 × フル本文だと余裕で 128k を超える事故が出るので、デフォルトを抑えた

注意点（残るもの）¶

find_next_task は DAG にサイクルがあると無限ループする。with_structured_output で related_ids が「自己参照」になる確率は下がったが、LLM 出力なのでゼロではない。recursion_limit=1000 は安全装置として大きすぎる気もするが、検索タスク 20件 + 書き込み 1件のような計画を許す余白
Tavily の無料枠は月 1,000 リクエスト。1クエリで 3〜5 回検索するので 200〜300 回の実行が上限
final_output = s["write"]["artifacts"][0] は「最後に実行された write ノードの新規 artifact」を捕まえる前提なので、計画に write が複数ある場合は最終 write が「最終レポート」とは限らない

拡張アイデア¶

進捗を Markdown / HTML レポートとして保存 — 現状は標準出力だけ。final_output を report_{timestamp}.md に書き出す
タスクごとの artifact 単位でストリーミング — Chainlit や Streamlit と組み合わせて、検索完了ごとに UI に追記
route ノードに「コスト見積もり」を追加 — タスク数が多すぎると LLM 課金が爆発するので、len(tasks) > N で警告 or 計画を再生成
Tavily を別の検索 API（Brave / Exa / SerpAPI）に差し替え — search() 関数を Protocol で抽象化して、SearchBackend を注入できるようにする
タスク分解の結果をユーザーに確認させる Human-in-the-Loop — plan の後に承認ステップを挟むノードを追加（第22回でやる Human-in-the-Loop の予習）
CRAG（第10回）への接続 — 検索結果の品質判定ノードを search の後に挟み、低品質ならクエリを書き直してリトライ

記事参照¶

Software Design 2024年6月号連載第09回「Research Agent（タスク分解型エージェント）」
関連回:
第07回・第08回（LangGraph の基礎）
第10回（CRAG）— 検索結果の品質判定を加える発展
第11回（ReAct Agent）— create_react_agent で似たことをもっと簡潔に書く
第22回・第24回（Human-in-the-Loop）— 計画承認ステップの追加

作成: 2026-05-19 ／最終更新: 2026-06-10

第09回 学習メモ: Research Agent（タスク分解型エージェント）¶