以前の記事では、Deep Agents CLI を使用して Deep Agent が Skills を使用するパターンをシミュレートする方法について検討しました。現在、LangChain はこの機能をネイティブにサポートしており、開発プロセスを大幅に簡素化しています。この記事では、この機能を深く体験し、よりインテリジェントな SQL アシスタントを構築します。

複雑な AI Agent を構築する際、開発者はジレンマに陥ることがよくあります。すべてのコンテキスト（データベースのテーブル構造、API ドキュメント、ビジネスルール）を System Prompt に一度に注入し、コンテキストウィンドウ（Context Window）がオーバーフローし、モデルの注意が散漫になるのか？それとも、コストのかかる頻繁な微調整（Fine-tuning）を選択するのか？

**Skills パターン（Skills Pattern）**は、エレガントな中間ルートを提供します。必要な知識を動的にロードすることで、コンテキストの効率的な利用を実現します。LangChain によるこのパターンのネイティブサポートは、「オンデマンド学習」能力を備えた Agent をより簡単に構築できることを意味します。

この記事では、公式ドキュメント Build a SQL assistant with on-demand skills と組み合わせて、読者がゼロから「オンデマンドで知識をロードする」SQL Assistant を構築できるようにガイドします。

1. コアコンセプト：なぜ Skills パターンを選択するのか？

従来の SQL Agent の限界

従来の SQL Agent アーキテクチャでは、通常、System Prompt で完全な Database Schema を提供する必要があります。ビジネスの発展に伴い、テーブルの数が数百に拡張されると、この方法では重大な問題が発生します。

Token の消費量が膨大：毎回、大量の無関係なテーブル構造を携帯するため、リソースが無駄になります。
幻覚のリスクが増加：過剰な無関係な干渉情報により、モデルの推論精度が低下します。
メンテナンスが困難：すべてのビジネスラインの知識が密接に結合されており、個別に反復処理することが困難です。

Skills パターン：漸進的な開示に基づくソリューション

Skills パターンは、**漸進的な開示（Progressive Disclosure）**原則に基づいて、知識の取得プロセスを階層化して処理します。

Agent の初期状態：どのような「スキル」（Skills）とその簡単な説明（Description）のみを把握し、軽量に保ちます。
実行時のロード：具体的な問題（「在庫のクエリ」など）に直面した場合、Agent はツール（load_skill）を積極的に呼び出して、そのスキルの詳細なコンテキスト（Schema + Prompt）をロードします。
タスクの実行：ロードされた正確なコンテキストに基づいて、具体的なタスク（SQL の作成と実行など）を実行します。

このパターンは、無限の拡張とチームの分離を効果的にサポートし、Agent がますます複雑になるビジネスシナリオに適応できるようにします。

2. システムアーキテクチャの設計

この実践プロジェクトでは、このパターンの実際の応用を示すために、2 つのコア Skills を含む SQL Assistant を構築します。

Sales Analytics（販売分析）：sales_dataテーブルを担当し、収入統計、注文トレンド分析などを処理します。
Inventory Management（在庫管理）：inventory_itemsテーブルを担当し、在庫レベルの監視、場所のクエリなどを処理します。

3. 開発環境の構築

このプロジェクトでは、Pythonuvを使用して効率的な依存関係管理を行います。

コア依存関係のインストール

uv add langchain langchain-openai langgraph psycopg2-binary python-dotenv langchain-community

PostgreSQL 環境の構成

ローカルで Postgres インスタンスを起動し、agent_platformデータベースを作成します。テーブル構造とテストデータを自動的に初期化するためのsetup_db.pyスクリプトを提供しています（詳細は文末のソースコードを参照）。

4. コア実装手順の詳細### ステップ 1: 領域スキル (The Knowledge) の定義

スキルを辞書構造として定義し、ファイルシステムまたはデータベースからのロードをシミュレートします。description (Agent が意思決定に使用) と content (実際にロードされる詳細なコンテキスト) を区別してください。

SKILLS = {"sales_analytics": {"description":"Useful for analyzing sales revenue, trends...","content":"""... Table Schema: sales_data ..."" },"inventory_management": {"description":"Useful for checking stock levels...","content":"""... Table Schema: inventory_items ..."" }}

ステップ 2: コアツール (The Capabilities) の実装

Agent はタスクを完了するために 2 つの重要なツールに依存します。

load_skill(skill_name): 指定されたスキルの詳細をランタイムに動的にロードします。
run_sql_query(query): 具体的な SQL ステートメントを実行します。

ステップ 3: Agent ロジック (The Brain) の編成

LangGraph を利用して ReAct Agent を構築します。System Prompt はここで重要な役割を果たし、Agent が Identify -> Load -> Query の標準作業手順 (SOP) に厳密に従うように指示します。

system_prompt ="""1. Identify the relevant skill.2. Use 'load_skill' to get schema.3. Write and execute SQL using 'run_sql_query'....Do not guess table names. Always load the skill first.""

5. 実行効果の検証

test_agent.py を実行して、Sales と Inventory の 2 つの異なる領域のクエリをテストしました。以下はコンソールの実際の出力ログであり、Agent が問題に応じてスキルを動的にロードする方法を示しています。

Testing Sales Query...Agent calling tools: [{'name': 'load_skill', 'args': {'skill_name': 'sales_analytics'}, 'id': 'call_f270d76b7ce4404cb5f61bf2', 'type': 'tool_call'}]Tool output:You are a Sales Analytics Expert.You have access to the 'sales_data' table.Table Schema:- id: integer...Agent calling tools: [{'name': 'run_sql_query', 'args': {'query': 'SELECT SUM(amount) as total_revenue FROM sales_data;'}, 'id': 'call_b4f3e686cc7f4f22b3bb9ea7', 'type': 'tool_call'}]Tool output: [(Decimal('730.50'),)]...Agent response: The total revenue is $730.50.Testing Inventory Query...Agent calling tools: [{'name': 'load_skill', 'args': {'skill_name': 'inventory_management'}, 'id': 'call_18c823b2d5064e95a0cfe2e3', 'type': 'tool_call'}]Tool output:You are an Inventory Management Expert.You have access to the 'inventory_items' table.Table Schema...Agent calling tools: [{'name': 'run_sql_query', 'args': {'query': "SELECT warehouse_location FROM inventory_items WHERE product_name = 'Laptop';"}, 'id': 'call_647ee3a444804bd98a045f00', 'type': 'tool_call'}]Tool output: [('Warehouse A',)]...Agent response: The Laptop is located in **Warehouse A**.## 6. \u5b8c\u5168\u30bd\u30fc\u30b9\u30b3\u30fc\u30c9\u53c2\u8003\n\n\u4ee5\u4e0b\u306f\u3001\u30d7\u30ed\u30b8\u30a7\u30af\u30c8\u306e\u5b8c\u5168\u30bd\u30fc\u30b9\u30b3\u30fc\u30c9\u3067\u3001\u30c7\u30fc\u30bf\u30d9\u30fc\u30b9\u306e\u521d\u671f\u5316\u30b9\u30af\u30ea\u30d7\u30c8\u3068Agent\u30e1\u30a4\u30f3\u30d7\u30ed\u30b0\u30e9\u30e0\u3092\u542b\u307f\u307e\u3059\u002e\n\n### 1. \u30c7\u30fc\u30bf\u30d9\u30fc\u30b9\u521d\u671f\u5316 (setup_db.py)\n\n`importpsycopg2frompsycopg2.extensionsimportISOLATION_LEVEL_AUTOCOMMITimportosfromdotenvimportload_dotenvload_dotenv()# \u30c7\u30fc\u30bf\u30d9\u30fc\u30b9\u63a5\u7d9a\u60c5\u5831\u304c .env \u306b\u8a2d\u5b9a\u3055\u308c\u3066\u3044\u308b\u3053\u3068\u3092\u78ba\u8a8d\u3057\u3066\u304f\u3060\u3055\u3044DB_HOST = os.getenv(### 2. Agent メインプログラム (main.py)

`import os from typing import Annotated, Literal, TypedDict, Union, Dict from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage from langchain_community.utilities import SQLDatabase from langchain_community.agent_toolkits import SQLDatabaseToolkit from langgraph.graph import StateGraph, START, END, MessagesState from langgraph.prebuilt import ToolNode, tools_condition

load_dotenv()

--- Configuration ---

BASE_URL = os.getenv("BASIC_MODEL_BASE_URL") API_KEY = os.getenv("BASIC_MODEL_API_KEY") MODEL_NAME = os.getenv("BASIC_MODEL_MODEL") DB_URI = f"postgresql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"

--- Database Setup ---

db = SQLDatabase.from_uri(DB_URI)

--- Skills Definition ---

SKILLS: Dict[str, Dict[str, str]] = { "sales_analytics": { "description": "売上収益、トレンド、地域パフォーマンスの分析に役立ちます。", "content": """ あなたはセールス分析のエキスパートです。 'sales_data'テーブルにアクセスできます。テーブルスキーマ：

id: integer (主キー)
transaction_date: date
product_id: varchar(50)
amount: decimal(10, 2)
region: varchar(50) 一般的なクエリ：
総収益：SUM(amount)
地域別の収益：GROUP BY region
売上トレンド：GROUP BY transaction_date""" }, "inventory_management": { "description": "在庫レベル、製品の場所、倉庫管理の確認に役立ちます。", "content": """ あなたは在庫管理のエキスパートです。 'inventory_item""" } }s' table.Table Schema:
id: integer (primary key)
product_id: varchar(50)
product_name: varchar(100)
stock_count: integer
warehouse_location: varchar(50) Common queries:
Check stock: WHERE product_name = '...'
Low stock: WHERE stock_count < threshold""" }}# --- Tools ---@tooldefload_skill(skill_name: str)-> str:""" 特定のスキルの詳細なプロンプトとスキーマをロードします。利用可能なスキル：
sales_analytics: 売上、収益、トランザクション分析用。
inventory_management: 在庫、製品、倉庫のクエリ用。 """ skill = SKILLS.get(skill_name) ifnotskill: returnf"Error: Skill '{skill_name}' not found. Available skills:{list(SKILLS.keys())}" returnskill["content"] @tool defrun_sql_query(query: str)-> str: """ データベースに対してSQLクエリを実行します。スキーマを理解するために、適切なスキルをロードした後でのみこのツールを使用してください。 """ try: returndb.run(query) exceptExceptionase: returnf"Error executing SQL:{e}" @tool deflist_tables()-> str: """データベースで使用可能なすべてのテーブルをリストします。""" returnstr(db.get_usable_table_names()) tools = [load_skill, run_sql_query, list_tables]

--- Agent Setup ---

llm = ChatOpenAI( base_url=BASE_URL, api_key=API_KEY, model=MODEL_NAME, temperature=0 ) llm_with_tools = llm.bind_tools(tools)

--- Graph Definition ---

classAgentState(MessagesState):

必要に応じてカスタム状態を追加できますが、単純なチャットにはMessagesStateで十分です

pass

defagent_node(state: AgentState): messages = state["messages"] response = llm_with_tools.invoke(messages) return{"messages": [response]}

workflow = StateGraph(AgentState) workflow.add_node("agent", agent_node)workflow.add_node("tools", ToolNode(tools))workflow.add_edge(START,"agent")workflow.add_conditional_edges("agent", tools_condition)workflow.add_edge("tools","agent")app = workflow.compile()# --- Main Execution ---if__name__ =="main": system_prompt ="""You are a helpful SQL Assistant.You have access to specialized skills that contain database schemas and domain knowledge.To answer a user's question:1. Identify the relevant skill (sales_analytics or inventory_management).2. Use the 'load_skill' tool to get the schema and instructions.3. Based on the loaded skill, write and execute a SQL query using 'run_sql_query'.4. Answer the user's question based on the query results.Do not guess table names. Always load the skill first.""" print("SQL Assistant initialized. Type 'quit' to exit.") print("-"*50) messages = [SystemMessage(content=system_prompt)]# Pre-warm connection checktry: print(f"Connected to database:{DB_URI.split('@')[-1]}")exceptExceptionase: print(f"Database connection warning:{e}")whileTrue:try: user_input = input("User: ")ifuser_input.lower()in["quit","exit"]:break messages.append(HumanMessage(content=user_input))# Stream the execution print("Agent: ", end="", flush=True) final_response =Noneforeventinapp.stream({"messages": messages}, stream_mode="values"):# In 'values' mode, we get the full state. We just want to see the last message if it's new. last_message = event["messages"][-1]# Update our message history with the latest statepass# After stream finishes, the last state has the final answer final_state = app.invoke({"messages": messages})

last_msg = final_state["messages"][-1]

if isinstance(last_msg, AIMessage):

print(last_msg.content)

messages = final_state["messages"]

Update history

print("-"*50)

except Exception as e:

print(f"\nError:{e}")

break`

LangChain Skills パターン実践：オンデマンドで知識をロードする SQL アシスタントの構築