Mainframe-to-LLM Bridge: LangGraphを活用した大型機キャラクタインターフェースの自律操作ロジックの構築
皆様、本日は「Mainframe-to-LLM Bridge」というテーマでお話しします。現代のテクノロジーが日々進化する中で、依然として多くの企業がビジネスの中核に大型機（Mainframe）システムを据えています。これらのシステムは、その堅牢性、信頼性、処理能力から長年にわたり利用されてきましたが、多くの場合、操作インターフェースは時代遅れのキャラクタベース（CUI）であり、現代の自動化技術との連携には課題が伴います。
しかし、大規模言語モデル（LLM）の登場は、この状況を一変させる可能性を秘めています。LLMの強力な自然言語理解と生成能力、そして推論能力を駆使すれば、人間が画面を読み取り、判断し、キーボード入力を行う一連の大型機操作を、LLMが自律的に実行する「ブリッジ」を構築できるかもしれません。本講義では、この壮大なビジョンを実現するための具体的なアプローチとして、LangGraphを用いた自律エージェントの構築方法に焦点を当てて解説します。
1. 大型機キャラクタインターフェースの特性と課題
まず、大型機キャラクタインターフェースがどのようなもので、なぜ自動化が難しいのかを理解しましょう。
1.1. 3270端末エミュレーションと画面構造
大型機のキャラクタインターフェースは、主にIBM 3270端末プロトコルに基づいています。これは、現代のSSHのようなストリームベースの通信とは異なり、画面全体をブロックとして送受信する「ブロックモード」通信が特徴です。ユーザーは画面上の特定のフィールドにデータを入力し、Enterキーやファンクションキー（PFキー）を押すことで、そのブロックデータが大型機に送信され、処理結果として新しい画面が返されます。
- 画面の固定性: 画面は通常、行と列で構成され、入力可能なフィールド（Protected/Unprotected fields）が定義されています。
- カーソル位置: 入力はカーソル位置に依存し、特定のフィールドへの移動が必要です。
- ファンクションキー: Enter、PF1-PF24、PA1-PA3などの特殊なキーが特定の操作（次画面へ、メニュー選択、保存など）に割り当てられています。
- 状態管理: 大型機はステートフルなシステムです。ある画面での操作が次の画面の状態に影響を与えます。
1.2. 自動化における課題
従来の自動化ツール（RPAなど）でも大型機操作の自動化は行われてきましたが、以下の課題がありました。
- 画面認識の脆弱性: 画面上の要素を座標や固定テキストで識別するため、画面レイアウトの僅かな変更でもスクリプトが破綻しやすい。
- 複雑なロジック: 複数の画面にまたがる複雑なビジネスプロセスを自動化するには、大量の条件分岐と状態管理が必要になり、スクリプトが肥大化・複雑化しやすい。
- 非定型操作への対応: 想定外のエラーメッセージやイレギュラーな画面遷移が発生した場合、人間の介入なしでは対応が困難。
- メンテナンスコスト: システム変更のたびにスクリプトの修正が必要となり、高いメンテナンスコストが発生する。
LLMは、これらの課題、特に非定型操作への対応と複雑なロジックの管理において、革新的な解決策を提供します。
2. LangGraphの基礎：自律エージェント構築のフレームワーク
LangGraphは、LangChainフレームワークの一部であり、エージェントの複雑なシーケンスや状態遷移をグラフ構造で定義・実行するためのライブラリです。これは、LLMを中核とする自律的なワークフローを構築する上で非常に強力なツールとなります。
2.1. グラフ構造とエージェントの状態
LangGraphは、有限状態機械（Finite State Machine）の概念に基づいています。ワークフローはノード（Node）とエッジ（Edge）で構成され、エージェントは特定の状態（State）を保持しながらノード間を遷移します。
- ノード (Node): グラフにおける処理の単位。LLMの呼び出し、外部ツールの実行、データ処理など、具体的なアクションを実行します。
- エッジ (Edge): ノード間の接続。処理の流れを示します。条件付きエッジ（Conditional Edge）を使用することで、現在の状態やノードの出力に基づいて次に実行するノードを動的に決定できます。
- 状態 (State): グラフ全体で共有されるデータ。エージェントの現在の状況、これまでの操作履歴、LLMの思考過程、ツールの実行結果などを保持します。LangGraphは、この状態をミュータブル（変更可能）なオブジェクトとして扱います。
2.2. Agentic Behaviorの実現
LangGraphを使用することで、LLMは「思考 (Think) → 計画 (Plan) → 行動 (Act) → 観察 (Observe)」というサイクルを繰り返すエージェントとして振る舞うことができます。
- Observe (観察): 現在の大型機画面の状態や、これまでの操作結果を観察し、状態に反映します。
- Think (思考): LLMが現在の状態を分析し、次に何をすべきかを推論します。
- Plan (計画): 推論に基づき、利用可能なツールの中から最適なものを選択し、その引数を決定します。
- Act (行動): 選択されたツールを実行し、大型機に対して操作を行います。
このサイクルをグラフのノードとエッジにマッピングすることで、大型機操作の自律的な実行を実現します。
3. 「Mainframe-to-LLM Bridge」のアーキテクチャ設計
大型機キャラクタインターフェースの自律操作を実現するためのアーキテクチャを設計しましょう。主要なコンポーネントは以下の通りです。
3.1. 主要コンポーネント
| コンポーネント | 役割 |
|---|---|
| MainframeEnv | 大型機のエミュレータ（py3270など）を抽象化し、画面操作APIを提供する。 |
| AgentState | LangGraphのグラフ全体で共有される状態。 |
| Tools | LLMが利用できる特定の操作関数（画面読み取り、キー送信、テキスト入力など）。 |
| LLM Agent | 現在の状態に基づいて、次に行うべき操作（ツールと引数）を推論・決定する。 |
| LangGraph Workflow | ノードとエッジで構成される、エージェントの自律的な処理フロー。 |
3.2. 全体フローの概念図
LangGraphのノードとエッジを使い、以下のような基本的なワークフローを構築します。
+----------------+ +-------------------+ +-------------------+ +-------------------+ +-------------------+
| Start Node |------>| Observe State |------>| LLM Decide Action |------>| Execute Tool |------>| Check Outcome |
| (Initial Setup)| | (Read Mainframe) | | (Tool Selection) | | (Mainframe Op) | | (Success/Error) |
+----------------+ +-------------------+ +-------------------+ +-------------------+ +-------------------+
| |
| V
| +-------------------+
+--------------------------------------------------------------------------->| Finish Node |
| (Task Complete) |
+-------------------+ここで重要なのは、LLM Decide Action ノードがツールを選択し、Execute Tool ノードがそれを実行した後、Check Outcome ノードで結果を確認し、成功すれば Finish Node へ、そうでなければ再度 Observe State やエラーハンドリングのフローに戻る、という循環的な構造です。
4. 実装詳細：LangGraphとPythonによる構築
具体的なコード例を交えながら、各コンポーネントの実装を見ていきましょう。
4.1. Mainframe環境の抽象化 (MainframeEnv)
実際の大型機エミュレータ（例: py3270）との連携は、この MainframeEnv クラスにカプセル化します。これにより、LLMエージェントは具体的なエミュレータの実装詳細を知る必要がなくなります。ここでは概念的な実装を示します。
import time
import re
from typing import Dict, Any, Optional
# py3270などの実際のライブラリをラップするクラス
class MainframeEnv:
def __init__(self, host: str = "localhost", port: int = 23):
# 実際のpy3270クライアントの初期化などを行う
# self.tn = py3270.Emulator(host, port) # 例
print(f"Connecting to mainframe at {host}:{port}...")
self.current_screen_text = ""
self.history = []
self._mock_screen_state = {} # 模擬画面の状態管理用
def connect(self):
# 接続処理
# self.tn.connect()
print("Connected to mainframe (mock).")
self._load_mock_screen("initial") # 初期画面をロード
self._update_current_screen()
def disconnect(self):
# 切断処理
# self.tn.terminate()
print("Disconnected from mainframe (mock).")
def _load_mock_screen(self, screen_name: str):
"""模擬画面の状態をロードするヘルパー."""
mock_screens = {
"initial": [
"********************************************************************************",
"* *",
"* WELCOME TO THE MAINFRAME SYSTEM *",
"* *",
"* User ID: *",
"* Password: *",
"* *",
"* Press ENTER to login *",
"* *",
"********************************************************************************",
],
"menu": [
"********************************************************************************",
"* *",
"* MAIN MENU *",
"* *",
"* 1. Account Inquiry *",
"* 2. Fund Transfer *",
"* 3. System Administration *",
"* *",
"* Enter choice and press ENTER: *",
"* *",
"********************************************************************************",
],
"account_inquiry": [
"********************************************************************************",
"* *",
"* ACCOUNT INQUIRY *",
"* *",
"* Account Number: *",
"* *",
"* Press ENTER to view details, PF3 to return *",
"* *",
"********************************************************************************",
],
"account_details": [
"********************************************************************************",
"* *",
"* ACCOUNT DETAILS *",
"* *",
"* Account No: 123456789 *",
"* Balance: $1,234,567.89 *",
"* Status: Active *",
"* *",
"* Press PF3 to return *",
"* *",
"********************************************************************************",
],
"error_login": [
"********************************************************************************",
"* *",
"* WELCOME TO THE MAINFRAME SYSTEM *",
"* *",
"* User ID: invalid_user *",
"* Password: invalid_password *",
"* *",
"* ERROR: Invalid User ID or Password. Press ENTER to retry. *",
"* *",
"********************************************************************************",
],
"error_generic": [
"********************************************************************************",
"* *",
"* SYSTEM ERROR *",
"* *",
"* An unexpected error occurred. Please try again. *",
"* *",
"* *",
"* Press PF3 to return to Main Menu. *",
"* *",
"********************************************************************************",
]
}
self._mock_screen_state = {
'text': "n".join(mock_screens.get(screen_name, mock_screens["initial"])),
'name': screen_name,
'input_fields': self._parse_input_fields(mock_screens.get(screen_name, mock_screens["initial"]))
}
def _parse_input_fields(self, screen_lines):
"""画面から入力フィールドを解析する簡易的な関数（模擬）"""
fields = []
for r, line in enumerate(screen_lines):
# 例: "User ID: " のように、ラベルの後にスペースが続く箇所を探す
matches = re.finditer(r'([A-Za-zs]+?):(s+)', line)
for m in matches:
label = m.group(1).strip()
start_col = m.end(1) + len(m.group(2)) # ラベルとコロンの後の空白の次
# 簡略化のため、ここでは固定長で入力フィールドを仮定
# 実際の3270ではフィールドの属性（長さ、保護/非保護）を取得する
if label in ["User ID", "Password", "Account Number", "Enter choice"]:
fields.append({
'label': label,
'row': r,
'col': start_col,
'length': 20 # 模擬的な長さ
})
return fields
def _update_current_screen(self):
"""現在の模擬画面の内容を更新し、履歴に追加."""
self.current_screen_text = self._mock_screen_state['text']
self.history.append(self.current_screen_text)
print("n--- Current Mainframe Screen ---")
print(self.current_screen_text)
print("--------------------------------")
def read_screen(self) -> str:
"""現在の画面テキスト全体を取得する."""
# 実際のpy3270では self.tn.display.get_screen_as_text() など
return self.current_screen_text
def find_input_field(self, label: str) -> Optional[Dict[str, Any]]:
"""指定されたラベルの入力フィールドを見つける（模擬）."""
for field in self._mock_screen_state['input_fields']:
if field['label'].lower() == label.lower():
return field
return None
def type_text(self, row: int, col: int, text: str):
"""指定された座標に入力し、画面を更新する（模擬）."""
# 実際のpy3270では self.tn.send_string(text, row, col) など
print(f"Typing '{text}' at ({row}, {col})")
lines = self._mock_screen_state['text'].split('n')
if 0 <= row < len(lines):
line = list(lines[row])
for i, char in enumerate(text):
if col + i < len(line):
line[col + i] = char
lines[row] = "".join(line)
self._mock_screen_state['text'] = "n".join(lines)
else:
print(f"Warning: Row {row} is out of bounds for typing.")
def send_key(self, key: str):
"""特定のキーを送信し、画面を更新する（模擬）."""
# 実際のpy3270では self.tn.send_key(py3270.Keys.ENTER) など
print(f"Sending key: {key}")
# 模擬的な画面遷移ロジック
current_screen_name = self._mock_screen_state['name']
if key.upper() == "ENTER":
if current_screen_name == "initial":
# ログインロジックの模擬
user_id_field = self.find_input_field("User ID")
password_field = self.find_input_field("Password")
user_id_val = self._get_text_at(user_id_field['row'], user_id_field['col'], user_id_field['length']) if user_id_field else ""
password_val = self._get_text_at(password_field['row'], password_field['col'], password_field['length']) if password_field else ""
if user_id_val.strip() == "user" and password_val.strip() == "pass":
self._load_mock_screen("menu")
else:
self._load_mock_screen("error_login")
elif current_screen_name == "menu":
choice_field = self.find_input_field("Enter choice")
choice_val = self._get_text_at(choice_field['row'], choice_field['col'], choice_field['length']) if choice_field else ""
if choice_val.strip() == "1":
self._load_mock_screen("account_inquiry")
else:
# 他の選択肢はエラーとして扱うか、別の画面へ
self._load_mock_screen("error_generic") # 模擬的にエラー
elif current_screen_name == "account_inquiry":
account_num_field = self.find_input_field("Account Number")
account_num_val = self._get_text_at(account_num_field['row'], account_num_field['col'], account_num_field['length']) if account_num_field else ""
if account_num_val.strip() == "123456789":
self._load_mock_screen("account_details")
else:
self._load_mock_screen("error_generic") # 模擬的にエラー
elif current_screen_name == "error_login":
# エラー画面でENTERを押すと、再度ログイン画面
self._load_mock_screen("initial")
else:
# デフォルトの動作
pass
elif key.upper() == "PF3":
if current_screen_name == "account_inquiry" or current_screen_name == "account_details" or current_screen_name == "error_generic":
self._load_mock_screen("menu")
elif current_screen_name == "menu":
# メインメニューからPF3でログアウトや終了を模擬
self._load_mock_screen("initial")
else:
pass
self._update_current_screen()
time.sleep(0.5) # 画面更新の待機を模擬
def _get_text_at(self, row, col, length) -> str:
"""模擬画面の特定座標からテキストを取得."""
lines = self._mock_screen_state['text'].split('n')
if 0 <= row < len(lines):
line = lines[row]
return line[col : col + length].strip()
return ""この MainframeEnv クラスは、py3270 のような実際のライブラリをラップすることを想定していますが、ここでは内部的に模擬的な画面状態 _mock_screen_state を持ち、type_text や send_key が呼ばれると、その状態を更新し、read_screen で最新の画面テキストを返すようにしています。
4.2. AgentStateの定義
LangGraphのワークフロー全体で共有される状態を定義します。TypedDict を使用すると、型ヒントによって状態の構造を明確にできます。
from typing import TypedDict, List, Dict, Any, Optional
class AgentState(TypedDict):
"""LangGraphエージェントの状態を定義します。"""
# 現在の大型機画面のテキストコンテンツ
current_screen: str
# これまでの操作履歴（アクションと結果）
operation_history: List[Dict[str, Any]]
# LLMが次に実行すべきツールとその引数（LLMの出力）
next_action: Optional[Dict[str, Any]]
# 現在のタスクの目標や指示
task_goal: str
# エラーメッセージ、または成功メッセージ
status_message: str
# LLMの思考過程（デバッグ用）
llm_thought: Optional[str]
# 試行回数（ループ検出やリトライ制御用）
attempts: int4.3. Toolの定義
LLMが大型機を操作するために利用できるツールを定義します。LangChainの tool デコレータを使用すると、Python関数を簡単にツールとして公開できます。
from langchain_core.tools import tool
# MainframeEnvのインスタンスはグローバルまたはDIで管理
mainframe_env = MainframeEnv()
@tool
def read_mainframe_screen() -> str:
"""
現在の大型機画面のテキスト内容全体を読み取ります。
このツールは、現在の画面状態を把握するために常に最初に呼び出すべきです。
"""
screen_text = mainframe_env.read_screen()
return screen_text
@tool
def type_text_into_field(field_label: str, text_to_type: str) -> str:
"""
指定されたラベルの入力フィールドにテキストを入力します。
例: type_text_into_field(field_label="User ID", text_to_type="myuser")
Args:
field_label (str): 入力したいフィールドのラベル（例: "User ID", "Password", "Account Number"）。
画面上の表示と完全に一致させる必要があります。
text_to_type (str): フィールドに入力するテキスト。
Returns:
str: 成功メッセージ、またはエラーメッセージ。
"""
field = mainframe_env.find_input_field(field_label)
if field:
mainframe_env.type_text(field['row'], field['col'], text_to_type)
return f"Successfully typed '{text_to_type}' into field '{field_label}'."
else:
return f"Error: Field '{field_label}' not found on screen. Current screen: {mainframe_env.read_screen()}"
@tool
def send_key_to_mainframe(key_name: str) -> str:
"""
大型機に特定のキーを送信します（例: 'ENTER', 'PF3', 'PF7'）。
操作を確定したり、画面を遷移させたりするために使用します。
Args:
key_name (str): 送信するキーの名前（大文字で指定）。例: "ENTER", "PF3", "PF7", "PF12"。
Returns:
str: 成功メッセージ、またはエラーメッセージ。
"""
try:
mainframe_env.send_key(key_name.upper())
return f"Successfully sent key '{key_name}' to mainframe."
except Exception as e:
return f"Error sending key '{key_name}': {str(e)}"
# 利用可能なツールのリスト
tools = [read_mainframe_screen, type_text_into_field, send_key_to_mainframe]4.4. LLM Agentの定義とプロンプトエンジニアリング
LLMは、現在の画面状態と利用可能なツールを考慮して、次に実行すべきアクションを推論します。Function CallingをサポートするLLMを使用すると、LLMは直接ツールとその引数を出力できます。
プロンプトは、LLMが大型機操作の文脈を理解し、適切な判断を下すために非常に重要です。
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents import AgentExecutor, create_tool_calling_agent
# LLMの初期化
# OpenAIのAPIキーは環境変数で設定されていると仮定
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# プロンプトテンプレートの定義
prompt = ChatPromptTemplate.from_messages(
[
("system",
"あなたは大型機システムを操作する熟練したエキスパートです。ユーザーの指示に従い、"
"与えられたツールを使って大型機キャラクタインターフェースを操作してください。n"
"現在の画面状態を常に把握し、適切な入力とキー送信を行ってください。n"
"特に重要な指示:n"
"1. まずは必ず `read_mainframe_screen` ツールを呼び出し、現在の画面状態を把握してください。n"
"2. `type_text_into_field` ツールを使用する際は、フィールドのラベルを正確に指定してください。n"
"3. `send_key_to_mainframe` ツールを使用する際は、EnterやPFキーなど、適切なキー名を指定してください。n"
"4. 最終的にタスクが完了したら、`status_message` に完了メッセージを設定し、処理を終了してください。n"
"5. エラーが発生した場合は、`status_message` にエラー内容を設定し、必要に応じてリトライ戦略を検討してください。n"
"6. 常に現在の画面に表示されている情報に基づいて判断してください。画面にない情報を推測しないでください。n"
"7. 必要に応じて、`operation_history` を参照してこれまでの操作を確認してください。"
),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
]
)
# LLMとツール、プロンプトを使ってエージェントを作成
# create_tool_calling_agentはLLMがツール呼び出しを生成できるようにします。
agent_runnable = create_tool_calling_agent(llm, tools, prompt)4.5. LangGraphワークフローの構築
いよいよLangGraphを使って、これらのコンポーネントを繋ぎ合わせ、自律的なワークフローを定義します。
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
# メインフレーム環境の初期化
# mainframe_env = MainframeEnv() # 既にグローバルに定義済み
# グラフの状態を初期化するヘルパー関数
def initialize_state(task_goal: str) -> AgentState:
mainframe_env.connect() # メインフレームに接続
initial_screen = mainframe_env.read_screen()
return AgentState(
current_screen=initial_screen,
operation_history=[],
next_action=None,
task_goal=task_goal,
status_message="Initializing...",
llm_thought=None,
attempts=0
)
# ノード関数の定義
# 各ノードはAgentStateを受け取り、更新されたAgentStateを返すか、
# LangGraphの特殊な値（ENDなど）を返します。
def observe_mainframe(state: AgentState) -> AgentState:
"""現在の大型機画面を読み取り、状態を更新するノード."""
print("--- Node: observe_mainframe ---")
current_screen_text = mainframe_env.read_screen()
# 画面が変更されていない場合は、無駄なLLM呼び出しを避けるためにスキップするロジックも追加可能
# if state['current_screen'] == current_screen_text:
# print("Screen not changed, skipping observation.")
# return state
print("Observed new screen.")
return {**state, "current_screen": current_screen_text, "status_message": "Screen observed."}
def call_llm_agent(state: AgentState) -> AgentState:
"""LLMエージェントを呼び出し、次のアクションを決定するノード."""
print("--- Node: call_llm_agent ---")
current_screen = state["current_screen"]
task_goal = state["task_goal"]
history = state["operation_history"]
attempts = state["attempts"] + 1
# LLMへの入力メッセージを構築
chat_history_messages = []
for op in history:
chat_history_messages.append(AIMessage(content=op['tool_name'], tool_calls=[op['tool_call']]))
chat_history_messages.append(ToolMessage(content=op['tool_output'], tool_call_id=op['tool_call']['id']))
# LLMに渡すinput
llm_input = f"現在の大型機画面:n{current_screen}nnタスク目標: {task_goal}nn"
f"これまでの操作履歴:n{history}nn"
f"現在の画面と目標を考慮して、次に行うべきアクションを決定してください。"
f"もしタスクが完了したと判断できる場合は、`status_message` に完了メッセージを設定し、ツール呼び出しは行わないでください。"
f"エラー状態が継続している場合は、`status_message` にエラー内容を設定し、ツール呼び出しは行わないでください。"
# LLMを実行し、次のアクション（ツール呼び出し）を取得
agent_output = agent_runnable.invoke(
{"input": llm_input, "chat_history": chat_history_messages}
)
# LLMの出力からツール呼び出し情報を抽出
next_action = None
llm_thought = str(agent_output.content) # LLMの思考過程を記録
# LLMがツール呼び出しを決定した場合
if agent_output.tool_calls:
tool_call = agent_output.tool_calls[0] # 最初のツール呼び出しを使用
next_action = {
"tool_name": tool_call.get("name"),
"tool_args": tool_call.get("args"),
"tool_call": tool_call # LangChainのToolCallオブジェクトをそのまま格納
}
print(f"LLM decided to call tool: {next_action['tool_name']} with args {next_action['tool_args']}")
else:
# LLMがツールを呼び出さなかった場合、タスク完了またはエラーと判断
print("LLM did not propose a tool call. Assuming task completion or error.")
# LLMの出力内容をstatus_messageに反映させる
return {**state, "status_message": llm_output_content, "attempts": attempts, "llm_thought": llm_thought}
return {**state, "next_action": next_action, "attempts": attempts, "llm_thought": llm_thought}
def execute_tool_action(state: AgentState) -> AgentState:
"""LLMが決定したツールアクションを実行するノード."""
print("--- Node: execute_tool_action ---")
next_action = state["next_action"]
if not next_action:
return {**state, "status_message": "Error: No action to execute.", "next_action": None}
tool_name = next_action["tool_name"]
tool_args = next_action["tool_args"]
tool_call = next_action["tool_call"] # LangChainのToolCallオブジェクト
# ツールを実行
tool_output = "Tool execution failed."
try:
# 利用可能なツールの中から名前で検索して実行
for t in tools:
if t.name == tool_name:
tool_output = t.invoke(tool_args)
break
else:
tool_output = f"Error: Tool '{tool_name}' not found."
except Exception as e:
tool_output = f"Error executing tool '{tool_name}': {str(e)}"
print(f"Tool '{tool_name}' executed. Output: {tool_output}")
# 実行結果を履歴に追加
updated_history = state["operation_history"] + [{
"tool_name": tool_name,
"tool_args": tool_args,
"tool_output": tool_output,
"tool_call": tool_call # ToolCallオブジェクトも履歴に保存
}]
return {**state, "operation_history": updated_history, "next_action": None, "status_message": "Tool executed."}
def check_task_status(state: AgentState) -> str:
"""
現在の状態を確認し、次のノードを決定する条件付きエッジ関数。
タスク完了、エラー、または継続のいずれかを返す。
"""
print("--- Node: check_task_status ---")
current_screen = state["current_screen"]
status_message = state["status_message"]
attempts = state["attempts"]
# LLMがツールを呼び出さなかった場合、それが完了またはエラーを示すかチェック
if state["next_action"] is None:
if "完了" in status_message or "Success" in status_message:
print("Task deemed complete by LLM.")
return "finish"
elif "Error" in status_message or "エラー" in status_message or "Invalid" in status_message:
print(f"Error detected in status message: {status_message}")
if attempts >= 3: # リトライ回数制限
print("Max attempts reached for error. Terminating.")
return "error_terminate"
else:
print("Retrying after error.")
return "retry_after_error"
else:
# LLMがツールを呼び出さず、かつ明確な完了/エラーメッセージもない場合、何らかの問題
print(f"LLM did not propose action and no clear status: {status_message}. Terminating.")
return "error_terminate"
# 通常の処理継続
print("Task still ongoing, continuing to LLM decision.")
return "continue"
# LangGraphの構築
workflow = StateGraph(AgentState)
# ノードの追加
workflow.add_node("observe_mainframe", observe_mainframe)
workflow.add_node("call_llm_agent", call_llm_agent)
workflow.add_node("execute_tool_action", execute_tool_action)
# エントリーポイントの設定
workflow.set_entry_point("observe_mainframe")
# エッジの設定
# 1. 画面観察 -> LLM呼び出し
workflow.add_edge("observe_mainframe", "call_llm_agent")
# 2. LLM呼び出し -> ツール実行 (条件付き)
# LLMがツール呼び出しを提案した場合のみ実行
workflow.add_conditional_edges(
"call_llm_agent",
check_task_status, # LLMの出力に基づいて次のノードを決定
{
"continue": "execute_tool_action", # LLMがツールを提案し、続行する場合
"finish": END, # LLMがタスク完了と判断した場合
"retry_after_error": "observe_mainframe", # LLMがエラーを検知し、リトライする場合
"error_terminate": END # エラーで終了する場合
}
)
# 3. ツール実行 -> 画面観察 (ループバック)
workflow.add_edge("execute_tool_action", "observe_mainframe")
# グラフのコンパイル
app = workflow.compile()5. 具体的なシナリオ例：アカウント照会
構築したLangGraphエージェントを使って、特定のシナリオを実行してみましょう。
「ユーザー ‘user’ とパスワード ‘pass’ でログインし、メニューで ‘1’ を選択してアカウント照会画面に進み、アカウント番号 ‘123456789’ の詳細を表示する」というタスクを想定します。
# ワークフローの実行
task = "ユーザー 'user' とパスワード 'pass' で大型機にログインし、"
"メニューから 'Account Inquiry' を選択し、アカウント番号 '123456789' の詳細を照会してください。"
# 初期状態を設定し、LangGraphを実行
initial_state = initialize_state(task_goal=task)
print("n--- Starting LangGraph Workflow ---")
final_state = None
for s in app.stream(initial_state):
print(f"n--- Current State ---")
for key, value in s.items():
print(f"Node: {key}")
# 状態の変更点のみ表示
for state_key, state_value in value.items():
if state_key != 'operation_history': # 履歴は長くなるので省略
print(f" {state_key}: {state_value}")
else:
print(f" {state_key}: {len(state_value)} entries")
final_state = s
print("n--- Workflow Completed ---")
print(f"Final Status: {final_state['__end__']['status_message']}")実行の流れ (コンソールの出力例):
- Connecting to mainframe… Connected to mainframe (mock).
- — Current Mainframe Screen — (初期画面表示)
- — Node: observe_mainframe —
- — Node: call_llm_agent — (LLMが画面を分析し、
type_text_into_field("User ID", "user")を提案) - — Node: execute_tool_action — (User ID入力)
- — Node: observe_mainframe — (画面が更新されないまま、LLMはPassword入力を提案)
- — Node: call_llm_agent — (LLMが
type_text_into_field("Password", "pass")を提案) - — Node: execute_tool_action — (Password入力)
- — Node: observe_mainframe —
- — Node: call_llm_agent — (LLMが
send_key_to_mainframe("ENTER")を提案) - — Node: execute_tool_action — (Enterキー送信 -> メインメニュー画面へ遷移)
- — Current Mainframe Screen — (メインメニュー画面表示)
- — Node: observe_mainframe —
- — Node: call_llm_agent — (LLMが
type_text_into_field("Enter choice", "1")を提案) - — Node: execute_tool_action — (選択肢1入力)
- — Node: observe_mainframe —
- — Node: call_llm_agent — (LLMが
send_key_to_mainframe("ENTER")を提案 -> アカウント照会画面へ遷移) - — Current Mainframe Screen — (アカウント照会画面表示)
- — Node: observe_mainframe —
- — Node: call_llm_agent — (LLMが
type_text_into_field("Account Number", "123456789")を提案) - — Node: execute_tool_action — (アカウント番号入力)
- — Node: observe_mainframe —
- — Node: call_llm_agent — (LLMが
send_key_to_mainframe("ENTER")を提案 -> アカウント詳細画面へ遷移) - — Current Mainframe Screen — (アカウント詳細画面表示)
- — Node: observe_mainframe —
- — Node: call_llm_agent — (LLMが「タスク完了」と判断し、ツールを提案しない)
- — Node: check_task_status — (LLMの出力から
status_messageを評価し、finishを返す) - — Workflow Completed —
この一連の操作は、LLMが画面を「見て」、目標を達成するためにどのツールをどの順序で使うべきかを「推論」し、それらのツールを「実行」することで、人間が手動で行うのと同様の自律的な操作を実現しています。
6. 課題と今後の展望
この「Mainframe-to-LLM Bridge」は大きな可能性を秘めていますが、実運用にはいくつかの課題があります。
6.1. 画面認識の堅牢性
現在の実装では、画面テキストから正規表現などで入力フィールドを解析していますが、これは画面レイアウトの変更に脆弱です。より堅牢な方法として、以下の点が考えられます。
- 構造化データ抽出: LLM自体に画面テキストから構造化された情報（フィールド名、値、入力可能か否かなど）を抽出させるプロンプトエンジニアリング。
- セマンティックな画面理解: 画面上の要素を単なるテキストとしてではなく、セマンティックな意味を持つコンポーネントとして認識させる技術（例: 「これはログインフォームだ」「これはメニューリストだ」）。
- OCRの活用: 画像ベースの3270エミュレータの場合、OCR技術と組み合わせることで、より柔軟な画面要素認識が可能になるかもしれません。
6.2. エラーハンドリングの強化
LLMは予測不能な状況に遭遇すると、時に誤った判断を下すことがあります。
- リトライ戦略: エラー発生時の自動リトライ、異なるアプローチの試行、バックオフ戦略。
- 人間へのエスカレーション: 自動処理が困難な重大なエラーが発生した場合、人間のオペレーターに通知し、介入を求めるメカニズム。
- LLMの自己修正: エラーメッセージをLLMにフィードバックし、その情報に基づいて次のアクションを修正させるプロンプト設計。
6.3. 性能とコスト
LLMのAPI呼び出しにはレイテンシーとコストがかかります。
- 効率的なプロンプト設計: 不要な情報の削減、few-shot学習の最適化により、トークン使用量を削減。
- 小規模モデルの活用: 特定のタスクには、より小さく高速なモデルを使用することを検討。
- キャッシュと状態管理の最適化: 頻繁に変わらない画面情報や、繰り返し実行されるロジックの結果をキャッシュする。
6.4. セキュリティと監査
大型機は機密性の高いデータを扱うことが多いため、セキュリティは最優先事項です。
- 認証と認可: LLMエージェントが使用する資格情報の安全な管理。
- 操作のログ記録: LLMが行ったすべての操作、その理由、結果を詳細にログに記録し、監査可能性を確保。
- アクセス制御: LLMエージェントがアクセスできる大型機機能の範囲を最小限に制限。
6.5. 学習と適応
初期のプロンプトとツールだけではカバーできない複雑なシナリオや、システムの変更に対応するには、エージェントが学習し、適応する能力が必要です。
- 人間からのフィードバック: エージェントが失敗した際に人間が修正し、そのフィードバックを元にLLMのプロンプトやツールを改善する仕組み。
- 強化学習: 長期的な目標達成に向けて、試行錯誤を通じて最適な行動シーケンスを学習させる可能性。
7. 自律エージェントが拓く未来
本講義では、LangGraphを用いてLLMが大型機キャラクタインターフェースを自律的に操作する「Mainframe-to-LLM Bridge」のアーキテクチャと実装について解説しました。LLMの強力な推論能力とLangGraphの柔軟なワークフロー管理を組み合わせることで、これまで自動化が困難だった大型機操作を、より知的で適応性の高い方法で実現する道筋が見えてきました。この技術は、レガシーシステムと現代AIの間のギャップを埋め、企業のデジタル変革を加速させる可能性を秘めています。今後、この分野の研究と開発が進むことで、より堅牢でインテリジェントな自動化ソリューションが実用化されることを期待しています。