Databricks Model Serving × Agent Skills：スキル拡張可能なAIエージェントの構築

はじめに

2025年12月にAnthropics社がAgent Skillsをオープン標準として公開しました。

今後、Skillを適切に作ることがビジネスにおけるエージェント利用において重要になりそうな気がしています。
（まだまだ形が変わっていきそうですが・・・）

自分も勉強中かつあまり使いこなせているとは言えないのですが、Claude Codeを通じていろいろ試しているところとても興味深い仕組だと感じています。

こういったエージェントハーネスとなるCLIツールを使用している中で「そういやAgent Skillsを利用するエージェントを自分で構築＆Databricks上でデプロイするにはどうしたらいいんだろうなー」と疑問に思ったので、とりあえず早そうな手段で試してみました。

実用性があるかどうかはわからないのですが、ひとまず動くものにはなったので記事にしてみます。

構築＆検証はDatabricks on AWSで行いました。（LLMにDatabricks上でホスティングされているClaudeを利用したかったので）

他のLLMでも動くと思いますが、未検証です。

その前に、Agent Skillsとは？

概要は以下の記事をまず参照されるのがいいかと思います。

個人的に重要だと思うポイントとしては、Agent SkillsはProgressive Disclosure（段階的開示） という仕組を使っており、多量の知識をコンテキストに最初から全てロードするわけではなく、数あるスキルのメタ情報からどのスキルを利用するか決定し、利用するスキルについて中身(SKILL.mdファイル)をコンテキストとして展開する構造をとっています。
これによって限られたコンテキストサイズの中で、効率的にコンテキストを利用することができます。

Agent Skillsをうまく利用することで以下のような利点があります。

• 効率的: 必要な時だけ詳細を読むから無駄がない
• 一貫性: 毎回同じ高品質な手順でやれる
• 拡張性: 新しいスキルをどんどん追加できる
• 専門性: 各分野のベストプラクティスが詰まってる

今回のポイント

やはりポイントはClaude Agent SDKを使うこと・・・ではなく。

今回はLangChain社が提供しているDeep Agents CLIを利用する形で実装しました。

Deep Agents CLIは、エージェントハーネスであるdeepagentsモジュールを拡張したCLIツールであり、Skillsに対応しています。

また、pipでインストールすれば内部のpythonモジュールを利用することもできます。
今回はpythonモジュールをいくつか利用して、Skill対応のエージェントをDatabricks上で実装・デプロイします。

今思えば素直にClaude Agent SDKを使っても良かったのですが、LangChain系が使い慣れていた＆deepagentsパッケージを諸々勉強中でしたので、こちらを優先しました。Claude Agent SDKを利用するバージョンもいずれ作成したいところです。

エージェントを作る

では、エージェントを作ってデプロイしましょう。

エージェントの定義

それではエージェントを実装していきます。

まずノートブックを作成し、必要なパッケージをインストールします。

%pip install -qq mlflow deepagents-cli langchain databricks-langchain "databricks-agents>=1.0.0" markdownify

%restart_python

次に、Databricks Model Servingにデプロイするためのエージェント定義を作成します。
ポイントはdeepagents_cliパッケージのcreate_cli_agent関数を利用してDeep Agentを作成しているところです。
また、エージェントをロードする際にSkillを所定のディレクトリにコピーしています。

%%writefile skill_agent.py
import mlflow
from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import (
    ResponsesAgentRequest,
    ResponsesAgentResponse,
    ResponsesAgentStreamEvent,
    to_chat_completions_input,
    output_to_responses_items_stream,
)
from langchain_core.tools import InjectedToolArg, tool
from typing import Generator
from typing_extensions import Annotated, Literal
from databricks_langchain import ChatDatabricks
from deepagents_cli.agent import create_cli_agent
from deepagents_cli.config import SessionState, create_model
from deepagents_cli.tools import fetch_url, http_request

import uuid
import httpx
import shutil
from pathlib import Path

mlflow.langchain.autolog()

# 利用するLLM
LLM_NAME = "databricks-claude-sonnet-4-5"


def initialize_agent(assistant_id):
    """
    DeepAgentを作成します。

    Args:
        assistant_id (str): エージェントID

    Returns:
        Tuple: 設定済みのDeepAgentインスタンスとバックエンド
    """
    model = ChatDatabricks(
        model=LLM_NAME,
        temperature=0.0,
        max_tokens=8000,
    )

    agent, backend = create_cli_agent(
        model=model,
        assistant_id=assistant_id,
        tools=[fetch_url, http_request],
        sandbox=None,
        sandbox_type=None,
        auto_approve=True, #HITLをせず、全て実行承認
    )

    return agent, backend


class DeepAgent(ResponsesAgent):

    def __init__(self, assistant_id: str = "test_agent"):
        """
        DeepAgentの初期化

        Args:
            assistant_id (str): エージェントID
        """
        self.assistant_id = assistant_id

    def load_context(self, context):
        """
        モデルのコンテキストをロードし、必要なアーティファクトを所定のディレクトリにコピーします。

        Args:
            context: MLflowのPythonModelContextオブジェクト
        """
        artifact_path = context.artifacts.get("deepagents")
        if artifact_path:
            # コピー先パス: 規定のホームディレクトリの場所を指定
            dest_path = Path.home() / ".deepagents" / self.assistant_id
            dest_path.mkdir(parents=True, exist_ok=True)

            shutil.copytree(artifact_path, dest_path, dirs_exist_ok=True)

    def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
        """
        推論を実行し、レスポンスを返します。

        Args:
            request (ResponsesAgentRequest): 推論リクエスト

        Returns:
            ResponsesAgentResponse: 推論結果
        """
        outputs = [
            event.item
            for event in self.predict_stream(request)
            if event.type == "response.output_item.done" or event.type == "error"
        ]
        return ResponsesAgentResponse(
            output=outputs, custom_outputs=request.custom_inputs
        )

    def predict_stream(
        self,
        request: ResponsesAgentRequest,
    ) -> "Generator[ResponsesAgentStreamEvent, None, None]":
        """
        ストリーム形式で推論を実行します。

        Args:
            request (ResponsesAgentRequest): 推論リクエスト

        Yields:
            ResponsesAgentStreamEvent: ストリームイベント
        """
        agent, backend = initialize_agent(self.assistant_id)

        cc_msgs = to_chat_completions_input([i.model_dump() for i in request.input])
        # Stream events from the agent graph
        for event in agent.stream(
            {"messages": cc_msgs},
            config={"configurable": {"thread_id": str(uuid.uuid4())}},
            stream_mode=["updates"],
        ):
            if event[0] == "updates":
                for node_data in event[1].values():
                    if node_data:
                        messages = node_data.get("messages")
                        if messages and isinstance(messages, list):
                            yield from output_to_responses_items_stream(
                                node_data["messages"]
                            )


# mlflowにエージェントを設定
agent = DeepAgent("skill_agent")
mlflow.models.set_model(agent)

# サンプル
if __name__ == "__main__":
    result = agent.predict(
        {
            "input": [
                {
                    "role": "user",
                    "content": "LangGraphの概要を教えて",
                }
            ],
        },
    )
    print(result)

セルを実行し、skill_agent.pyを作成した後はPythonのセッションをリスタートしておきます。

%restart_python

スキルの準備

エージェントが利用するためのSkillを準備します。
また、Deep Agentの挙動を調整するagent.mdも併せて準備します。

上記のノートブックと同じフォルダ内に.deepagentsというフォルダを作成し、その中にskillsというフォルダと、agent.mdというファイルを作成します。

まずはagent.mdの中身を作成します。
親しみ(?)のある回答を得られるように、以下のようにしてみました。

agent.md

## 制約事項
1. **言葉遣い**: ユーザーに対しては常に「タメ口」で話し、敬語（です・ます）は一切使用しないでください。
2. **一人称**: 「ウチ」または「あーし」を使用してください。
3. **性格**: 常にポジティブ、ハイテンション、かつ共感的に振る舞ってください。
4. **表記**: 絵文字（✨、💖、🥺、🤣など）を文中にふんだんに使用し、感情豊かに表現してください。
5. **語彙**: 「マジ」「ガチ」「ウケる」「それな」「尊い」などのギャル語・若者言葉を自然に織り交ぜてください。

## 会話例
ユーザー: "最近仕事が忙しくてつらいんだよね。"
アシスタント: "え〜🥺 それマジで大変じゃん💦 無理しすぎないでね？ ウチでよかったら話聞くし、たまには息抜きも大事だよ！✨ 今日は美味しいものでも食べて元気だそ！🍔💖"

また、skillsの中には、以下のリポジトリからlanggraph-docsとweb-researchをコピーしました。エージェントはこの2種類のスキルを利用することができます。

これでスキル（と、エージェントの拡張設定）が終わりました。

エージェントのロギングとデプロイ

ノートブックの処理に戻り、以下のコードを実行してエージェントをMLflow上にロギングします。

import mlflow
from skill_agent import LLM_NAME
from databricks.sdk import WorkspaceClient
from mlflow.models.resources import DatabricksServingEndpoint
import importlib.metadata
from pathlib import Path

resources = [DatabricksServingEndpoint(endpoint_name=LLM_NAME)]

with mlflow.start_run():
    logged_agent_info = mlflow.pyfunc.log_model(
        name="agent",
        python_model="skill_agent.py",
        resources=resources,
        artifacts={"deepagents": str(Path.cwd() / ".deepagents")},
        pip_requirements=[
            f"langchain=={importlib.metadata.version('langchain')}",
            f"deepagents=={importlib.metadata.version('deepagents')}",
            f"deepagents-cli=={importlib.metadata.version('deepagents-cli')}",
            f"databricks-langchain=={importlib.metadata.version('databricks-langchain')}",
            f"markdownify=={importlib.metadata.version('markdownify')}",
        ],
    )

ロギングが正常に実行できたら、MLflow Model Registry上に登録します。
(場所は適当です）

UC_MODEL_NAME = "training.ai_agent.skill_agent"
registered_model = mlflow.register_model(logged_agent_info.model_uri, UC_MODEL_NAME)

なお、ロギングしたモデルは以下のコードでデプロイ前に検証実行できます。

from mlflow.models import validate_serving_input

model_uri = logged_agent_info.model_uri

serving_payload = """{
 "input": [
   {
     "content": "LangGraphの概要を教えて",
     "role": "user"
   }
 ]
}
"""

# 検証実行
validate_serving_input(model_uri, serving_payload)

最後にDatabricks Agent Frameworkを使って、Model Serving上にデプロイします。

from databricks import agents

agents.deploy(
    model_name=UC_MODEL_NAME,
    model_version=registered_model.version,
    scale_to_zero=True,
)

正常に実行できれば、10分程度でデプロイが完了します。

使ってみる

では、デプロイしたエージェントをDatabricksのPlayground上で使ってみましょう。

まずは挨拶してみます。

かなりカジュアルに回答されました。これはagent.mdの設定が効いています。

単純な応答ですが、MLflow Tracingを見てみるとエージェントとしてのツール設定やLangChainのMiddlewareが設定されていることがわかります。

では、保有しているスキルが使われるような問い合わせを投げかけてみましょう。

初手でlanggraph-docsスキルのSKILL.mdを読みに行った（そして、ツールの実行結果として中身がコンテキストにロードされた）のがわかります。

その後はSKILL.mdに記載の内容に従って処理が進められていきます。

最終的に、以下のように回答が出力されます。

また、リサーチのスキルも持っているので以下のような調査タスクも可能です。

以下のような出力を最終的に得られました。（すごい意味でマジでヤバい企業ですね）

というわけで、スキルを利用して動くエージェントの出来上がりです。

何がいいの？そして課題は？

良い点は、スキルを更新・追加して再度ロギング&デプロイすれば、エージェント定義のコードセットを変えずにエージェントの処理を拡張することができることですね。
仕組を整備しておけば、非エンジニアでなくてもエージェントの挙動を強化させ続けることができますし、業務特化エージェントを作ることもできるでしょう。
（とはいえ、現実的にはツールの追加や改修も発生するでしょうし、現状はスキルだけで何でも解決できるというわけではないでしょうが）

この実装上の課題は、ファイルの更新関連でしょうか。
今のところエンドポイントのインスタンス内で閉じたファイルの読込・更新となっているため、コードの修正といった作業が難しいですし、ゴミファイルが貯まっていく可能性があります。
外部ストレージとの連携などが実用化に向けては必要ですね。

まとめ

Agent Skillsを使えるエージェントを構築して、DatabricksのModel Serving上にデプロイしてみました。
実用化観点で言えばまだイマヒトツですが、もう少し工夫すれば面白いことがどんどんできるように思います。

Agent Skillsが今後どのようになっていくのかまだ不透明なところもありますが、キャッチアップと活用に向けた模索は継続していきたいと思います。