はじめに: エージェントは本当に必要か?

LLMアプリケーションを開発する際、多くの開発者が最初に考えるのは「強力なエージェント」です。しかし、どのフレームワークを選ぶかで迷っているうちに時間が過ぎてしまいます。CrewAI、LangGraph、Microsoft Agent Framework... 選択肢が多すぎて、コードを書き始める前に挫折してしまうことも少なくありません。

でも待ってください。本当にエージェントフレームワークが必要ですか? 私は様々なドメインでLLMアプリケーションを開発してきましたが、実際に安定して動作するのは自律的なエージェントではなくワークフローであることが多いと気づきました。そして、ワークフローを構築する際には、フレームワークすら不要な場合があります。

この記事では、純Pythonとローカル関数、構造化出力、OpenAI Responses APIを使用してLLMワークフローをプロトタイピングする方法を紹介します。異常検知問題を実践例として取り上げ、なぜワークフローが最初に検討すべき抽象化なのかを説明します。

核心となる主張: エージェントは柔軟ですが、多くの実務問題は固定された手順で解決可能です。ワークフローは透明性と安定性を提供し、フレームワークなしでも十分に実装できます。

LLM workflow control flow diagram with decision nodes and structured outputs Dev Environment Setup

LLMワークフロー設計の4つの核心要素

1. 制御フロー (Control Flow)

ワークフローの骨格です。入力から出力までの移動経路をグラフで表現します。各ノードはコードベースの処理またはLLM呼び出しであり、エッジは情報伝達の方法(静的/条件付き)を決定します。重要なのは、コードがグラフを所有し、LLMは特定のノードに限定されることです。

2. 役割指示 (Role Instructions)

各LLMノードにどのような役割を与えるかを定義します。システムプロンプトにペルソナ、実行タスク、注意点、ドメインルールを明記します。

3. プロンプトビルダー (Prompt Builders)

LLMが現在のワークフロー状態で必要なコンテキストを組み立てる関数です。動的な値を受け取り、プロンプトテンプレートに注入し、最終的なプロンプトを生成します。コンテキストウィンドウを制御する核心的なポイントです。

4. 構造化出力 (Structured Output)

LLM出力を事前定義されたスキーマ(JSONまたはPydanticモデル)に強制します。これはLLMステップ間の契約(Contract)となり、下流のパースを安定させます。

実践: 純Pythonでデータ品質調査ワークフローを構築

Irisデータセットを使用して異常値を検出し、LLMで原因を診断するワークフローを構築します。全3ステップで構成されます。

ステップ1: LLM呼び出しヘルパーを作成

import os
import time
from pydantic import BaseModel
from openai import AzureOpenAI

client = AzureOpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    azure_endpoint=os.environ["OPENAI_API_BASE"],
    api_version=os.environ["OPENAI_API_VERSION"],
)

llm_telemetry = []  # パフォーマンス計測用

def call_llm(
    step_name: str,
    instructions: str,
    prompt: str,
    output_schema: type[BaseModel],
) -> BaseModel:
    """LLMを呼び出し、構造化出力を返す"""
    model = "gpt-5.4-mini"
    started = time.perf_counter()
    response = client.responses.parse(
        model=model,
        instructions=instructions,
        input=prompt,
        reasoning={"effort": "medium"},
        text_format=output_schema,
    )
    usage = response.usage.model_dump() if response.usage else {}
    llm_telemetry.append(
        {
            "step": step_name,
            "schema": output_schema.__name__,
            "model": model,
            "prompt_chars": len(prompt),
            "elapsed_s": round(time.perf_counter() - started, 2),
            "input_tokens": usage.get("input_tokens"),
            "output_tokens": usage.get("output_tokens"),
            "total_tokens": usage.get("total_tokens"),
        }
    )
    return response.output_parsed

ステップ2: 異常値スクリーニング (コードベース)

import pandas as pd
from scipy import stats

def screen_suspicious_samples(df: pd.DataFrame, species_col: str, features: list[str]) -> pd.DataFrame:
    """種別の特徴量のz-scoreを計算し、異常値候補を特定"""
    anomaly_scores = []
    for _, row in df.iterrows():
        species = row[species_col]
        subset = df[df[species_col] == species]
        z_scores = {}
        for feature in features:
            mean = subset[feature].mean()
            std = subset[feature].std()
            z_scores[feature] = abs((row[feature] - mean) / std) if std > 0 else 0
        anomaly_scores.append(max(z_scores.values()))
    df = df.copy()
    df["anomaly_score"] = anomaly_scores
    return df.sort_values("anomaly_score", ascending=False)

ステップ3: LLM調査者と説明者

from typing import Literal
from pydantic import BaseModel, Field

class InvestigationDecision(BaseModel):
    status: Literal["need_more_evidence", "enough_evidence"] = Field(..., description="さらに証拠が必要かどうか")
    reasoning: str = Field(..., description="証拠に基づく簡単な根拠")
    tool_name: Literal["compare_row_to_species_profile", "find_nearest_neighbors"] | None = Field(None, description="呼び出すローカル関数名")
    species: Literal["setosa", "versicolor", "virginica"] | None = Field(None, description="比較する種のプロフィール")
    k: int | None = Field(None, description="最近傍の数")

# 調査ループ
for round_id in range(1, MAX_ROUNDS + 1):
    decision = call_llm(
        step_name=f"investigator_row_{row_id}_round_{round_id}",
        output_schema=InvestigationDecision,
        instructions=INVESTIGATOR_INSTRUCTIONS,
        prompt=build_investigation_prompt(flagged_row, collected_evidence),
    )
    if decision.status == "enough_evidence":
        break
    tool_result = execute_tool_call(row_id, decision)
    collected_evidence.append(tool_result)

Python code editor with OpenAI Responses API and Pydantic model for structured output IT Technology Image

注意点および批判的視点

ワークフローの限界

このアプローチは、問題が明確で手順がある程度決まっている場合に強力です。しかし、以下のような限界があります:

  • 柔軟性の欠如: 問題がオープンエンドまたは予期しない状況が多い場合、ワークフローはむしろ制約になります。
  • 保守コスト: 分岐や条件が多くなると、複雑さが増す可能性があります。この場合、エージェントの方が良い選択かもしれません。
  • ドメイン知識への依存: ワークフロー設計にはドメインへの深い理解が必要です。初心者には参入障壁になり得ます。

フレームワークの価値

フレームワークは、本番環境での障害処理、可観測性、人間の介入、永続化などを提供します。しかし、アイデア検証段階ではオーバーエンジニアリングになる可能性があります。ワークフローが実際に機能することを確認した後、必要に応じてフレームワークに移行するのが効率的です。

日本開発コミュニティでの適用文脈

日本の開発現場では、特にレガシーシステムとの統合が重要です。ワークフローベースのアプローチは、既存コードベースにLLMを自然に統合できるため、有効です。また、金融業界では説明可能性が重視されるため、ワークフローは各ステップの入出力が明確で監査に有利です。ただし、日本ではLLM APIのコスト問題があるため、社内LLMやオープンソースモデルと組み合わせることを検討すべきです。

Data quality investigation dashboard showing anomaly detection and evidence gathering Technical Structure Concept

結論: シンプルに始め、必要に応じて複雑さを加える

多くの有用なLLMアプリケーションで必要なのは、自律エージェントや重いフレームワークではなく、明確なワークフローと制御フロー、役割指示、プロンプトビルダー、構造化出力、そしてそれらを結びつける純Pythonコードです。

手書きのワークフローは、後でエージェントやフレームワークに移行する際にもそのまま活用できます。制御フロー、プロンプト、スキーマ、ツール定義はすべて再利用可能です。

次のステップの学習方向:

  • OpenAI Responses APIの機能(ストリーミング、関数呼び出し)をさらに深く学びましょう。
  • LangGraphなどのフレームワークの核心概念を理解し、いつ移行するかの判断基準を立てましょう。
  • 実際のプロジェクトにワークフローを適用し、可観測性のためのロギングを追加してみましょう。

原文参照: この記事は Towards Data Scienceの原文 に基づき、日本の開発者視点で再構成しました。

合わせて読みたい記事

本コンテンツは、信頼性の高い情報源をもとにAIツールを活用して作成され、編集者によるレビューを経て公開されています。専門家によるアドバイスの代替となるものではありません。