高階関数によるAI Agentのコンテキストエンジニアリング

スパイスコード CTO の櫻木です. スパイスコードは,「ロカルメ・オーダー」

order.localmet.com

という AI Agent を内包した ERP サービスを開発・提供しているスタートアップです. 本記事では我々の AI Agent 開発におけるコンテキストエンジニアリングに関する取り組みを紹介します.

1 コンテキストエンジニアリングの重要性

AI Agentの作成において「コンテキストエンジニアリング」が重要であることは以前より周知されています. blog.langchain.com

以前は人間がLLMへ直接プロンプトを与えていたため「プロンプトエンジニアリング」に意識が向いていました. しかし,AI Agentは人の手がほとんど介在しないまま自律的に推論と実行を繰り返します. そのため単なるプロンプトエンジニアリングだけでなく,「エージェントが参照するコンテキストをどう設計するか」が決定的に重要になります. コンテキストエンジニアリングでは必要十分な情報だけを安全に供給することにより

  1. LLM の推論力を最大限引き出す
  2. LLM が抱える入力長制約を回避する
  3. “Lost in the Middle”問題を回避する

ことを目的とします. この視点がなければ,エージェントは複雑な業務フローの途中で意図せぬ結果に陥りがちです. 本記事ではこのコンテキストエンジニアリングに対するスパイスコードでの取り組みをロカルメオーダーに内包している電子メール注文書受領Agentとデータ抽出 Agentの例を使って紹介します.

2 tool利用によるコンテキスト圧迫問題

我々が提供しているAI Agentは以下のようなツールを使うことができます

  • 自前のmcpサーバ経由でユーザに代わりロカルメオーダー上のデータを操作・抽出
  • お客様の代わりに電子メールの確認・操作(注文書のダウンロード・アップロード)

これらのツールで課題になるのが目的を達成するために使うツールの入出力によるコンテキストの圧迫です. 以下の例のように

  1. ツールの入出力のデータ量が大きい場合
  2. 繰り返し処理が生じる場合

それぞれでコンテキストの圧迫が発生することは想像に難くないと思います.

例1

特定の店舗にある在庫の平均原価額取得

TOOL CALL: localmet_mcp.get_shop(query: "shopA")
    returns shopA

TOOL CALL: localmet_mcp.get_stocks(shopId: shopA.id)
    returns 10000 rows stock info -> ❌

TOOL CALL: average_cost_by_sku(stocks: 10000 rows stock info) -> ❌
    returns 1000 rows averaged stock info -> ❌

例2

特定期間における電子メールから注文書を抽出, 一部のメールではいわゆるPPAPが利用されている

TOOL CALL: get_emails(from: 'YYYYMMDD', to: 'YYYYMMDD')
    returns 1000 emails -> ❌

以下対象全てに対して実行 -> ❌
TOOL CALL: evaluate_order_email(id: 'email1')
    returns {
        is_order_related,
        urls,
        password_candidates,
        attachments
    }

TOOL CALL: handle_file(attachmentId: 'attachment1', passwordCandidates: tooManyCandidates) -> ❌

TOOL CALL: handle_url(url 'example.com', passwordCandidates: tooManyCandidates) -> ❌

3 高階関数, コード実行によるアプローチ

上記のような問題に対して我々は大きく分けて2つのアプローチを用いて対策しています. 現在それぞれ独立した機構として設計しユースケースに応じて使い分けていますが全てを統合したアプローチも検討中です. 重要な点はどちらも高階関数ツールを用意し,中間処理をコンテキストに載せないことです. 更に最終結果に関してもartifactとして保存し, idのみを返すことで次に実行するchainに再利用したり, サマリのみをユーザに提供することができます.

3.1 tool chain tool

mcpツール含め, ツール群を全てchain toolを経由して使用させることで無駄な中間ツールの出力を削減した上で最終的な結果はartifactとして保存しidのみ返す

例1

Tool Input

{
    context: "shopAある在庫の平均原価額取得"
    purpose: "shopAの存在確認をした上で全在庫を取得しskuごとに平均コストを算出する"
    tool_names: ["mcp_get_shop", "mcp_get_stocks", "average_cost_by_sku"]
    each_args: {
        "mcp_get_shop": { "query": "shopA"},
        "mcp_get_stocks": { "shopId": "get_shop.id"},
        "average_cost_by_sku": { "stocks": "get_stocks.result" },
    },
    expected_result: "skuごとの平均在庫金額リスト"
}

Tool Output

{
    "artifactId": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
}

例2

こちらはより内部状態の管理を明確にしたもの,artifactとして保存されるのは最終的結果に加えてプロセス全体の最終状態なども含める

Tool Input

{
    "initial_state": {
        "purpose": "YYYYMMDDからYYYYMMDDまでのメールから注文書googleDriveへアップロード",
        "password_candidates_all": []
    },
    {
        "steps": [
            {
                "tool_name": "mailbox_message_search_tool",
                "arguments": {
                    "after": "2025/11/01",
                    "before": "2025/12/31"
                },
                "result_key": "search"
            },
            {
                "items_key": ".search.messages",
                "item_alias": "message",
                "body": [
                    {
                        "tool_name": "mailbox_body_extractor_tool",
                        "arguments": {
                            "subject": ".message.subject",
                            "body": ".message.snippet"
                        },
                        "result_key": "body_collect"
                    },
                    {
                        "predicate": ".body_collect.password_candidates",
                        "body": [
                            {
                                "operation": "extend",
                                "target_key": "password_candidates_all",
                                "value_from": ".body_collect.password_candidates"
                            }
                        ]
                    }
                ]
            },
            {
                "items_key": ".search.messages",
                "item_alias": "message",
                "body": [
                    // 1件づつに対する処理
                ]
            }
        ]
    }
}

Tool Output

{
    "artifactId": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
}

3.2 コード実行

tech-blog.localmet.com で紹介しているような安全な実行環境においてコード実行し, 最終的な結果はartifactとして保存しidのみ返す. こちらはまだ実験段階ではありますがDSLよりAI Agentにとって明瞭で強力なツールになると考えています.

Tool Input

def exec():
    shop = mcp_get_shop("shopA")
    stocks = mcp_get_stocks(shop.id)
    return average_cost_by_sku(stocks)

Tool Output

{
    "artifactId": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
}

4 自己学習サイクルによる継続的改善

我々は以下の3段階のフィードバックループを実装することにより複雑なDSL, コード生成の難しさを解決しています.

実行時の自己修復

Pydantic modelによるバリデーションに加え,実行可能性も評価します.エラー時は詳細なエラー情報をエージェントに返却し,自己修復を促します.

# 例: shopIdが存在しないケース
{
    "error": "ValidationError",
    "detail": "shopId 'invalid-id' not found",
    "suggestion": "Use mcp_search_shop first"
}

成功パターンのKnowledge化

成功した実行ログに対して:

  1. LLM as a Judge または人間が品質評価
  2. 高評価のパターンをElasticsearchに索引化
  3. タスクコンテキストでベクトル検索
  4. 類似パターンをSystem Promptに注入

学習曲線

  • 初回実行: 平均2-3回のvalidation失敗を経て成功
  • Knowledge化後: 同様のタスクはほぼ一発で成功
  • 運用数ヶ月: 頻出タスクの初回成功率が向上

この仕組みにより,エージェントは使うほど賢くなり, 懸念される可能性のあるJSON生成の不安定性も実運用では問題になりません.

┌─────────────────────────────────────────┐
│         初回実行 (Cold Start)            │
├─────────────────────────────────────────┤
│ LLM生成 → ❌ → 修正 → ❌ → 修正 → ✅     │
│              (平均2-3回)                 │
└─────────────────────────────────────────┘
                    ↓
         ┌──────────────────┐
         │  評価 + 索引化    │
         └──────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│      2回目以降 (Warm Start)              │
├─────────────────────────────────────────┤
│ Knowledge検索 → LLM生成 → ✅            │
│              (ほぼ一発)                  │
└─────────────────────────────────────────┘

5 高階関数で表現していることのメリット

どのアプローチも,「処理定義(関数)を引数に取り,新しい複合処理(関数)を返す」高階関数として設計されています. これによってエージェントが逐次実行した場合と比較してコンテキストエンジニアリングの文脈で大きな強みがあります. それ以外にも以下のようなメリットがありました.

  • 宣言と実行の分離:エージェントは JSON 宣言を返すだけで,実動作に費やすトークンを削減.実行側は常に同じ検証ルールで評価するため,一貫性を確保.
  • 合成の安全性:chain tool は バリデーションや入出力の取り扱いを内部に閉じ込めているので,関数合成(ツール連鎖)時の型崩れや引数抜けをチェーン内で吸収可能
  • 秘匿情報の安全性:秘匿情報の受け渡しをchain内に閉じ込めることでLLMに対する入出力にそれらの情報が顕在化しない.
  • 知識の再利用:高階関数ツールへのInputとその結果をロギング -> LLM as a Judgeなどで評価 -> Knowledge として保存することで,同じ宣言を別エージェントが再評価したり,成果の出た構成をテンプレ化したりできる.
  • プロセスの階層化:テンプレ化したchain toolをchain toolで利用することで複雑なパターンを容易に表現
  • スケーリング: 並列実行可能な箇所が明確かつ処理機構は閉じているため,スケール時にコントロールが容易

6 今後の展望

  • Python サンドボックス応用:既存の sandbox を使って,chain tool が動的な Python タスクを compile/decompile できるモジュールを挟み,宣言の一部をコード化 → 検証 → 再宣言するハイブリッド実行を検討中.
  • コード実行との連携:Anthropic が提唱するような MCP ベースのコード実行環境とも親和性が高く,彼らが提唱するsearch_toolsと組み合わせることによってより効率的で安全な状態に拡張できる可能性が高い.
  • healerの構築, playwright-healerのように出力されたtool inputに対してより適切な構造に自律的に修正を促すエージェントとの協調

tech-blog.localmet.com www.anthropic.com

終わりに

スパイスコードでは現在積極的に採用を行なっています. この記事を読んでAIを使ったチャレンジングな機能を開発してみたいと思った方,興味を持った方はぜひお話ししましょう!

corp.spicescode.co.jp

引用

arxiv.org blog.langchain.com www.anthropic.com

コンパイラ技術を使用したAI Agentの安定性を向上させる仕組み

スパイスコード代表の中河です。 スパイスコードは、「ロカルメ・オーダー」

order.localmet.com

という AI Agent を内包した ERP サービスを開発・提供しているスタートアップです。
本ブログでは、私たちの AI Agent がどのような仕組みで動いているのか、そして他の Agent とは何が異なるのかについて、ご紹介していく予定です。
第 2 回目となる今回は、「コンパイラ技術を使用した AI Agent の安定性を向上させる仕組み」として、第 1 回目に続いて古の技術を用いた弊社の AI Agent の仕組みについて、その概要をご紹介します。
※ Sandbox 技術の続きはそのうち別記事として書く予定です

AIエージェントアーキテクチャ

TL;DR: ロカルメ・オーダーでは Automation RPA と呼ばれる RPA を内蔵しているのですが、同じく内蔵している AI Agent がそれらを構築・実行する機能を持っており、特に構築部分では少し特殊なアーキテクチャを採用しています。


1. なぜいま “コンパイラ技術” なのか

  • AI Agentをプロダクトに投入するためにはパフォーマンスの安定性が大きな壁となる。

  • AI Agentが確率的に出力した結果をコンパイラ技術を通じて、確定的な物へ変換する事が出来る。

  • コンパイラ技術を元にAI Agentにフォードバックを行い自己改善するループを構築できる。

Automation RPA

例えば上記のようなワークフローを弊社のAI Agentが構築する場合、いきなりワークフローのデータ構造を出力するのではなく

  1. Python コードの出力
  2. Python コードを中間コードへ変換
  3. 中間コードをAutomationの処理構造に変換

という処理を行なっており、コンパイラのフロントエンド・バックエンドのようなアーキテクチャになっています。


2. アーキテクチャと実装

コンパイラアーキテクチャ概要

2.1 Python コードの出力

AI Agentはドキュメントや人間からの自然言語での指示を入力として以下のテンプレートを元にprocess関数を実装したPythonコードを出力します

import copy
from typing import Any
from {evaluate_module} import ( # you must keep this line
    {evaluate_functions}
)
from {process_module} import ( # you must keep this line
    {process_functions}
)

def process(arg: Any) -> Any:
    target = copy.deepcopy(arg)
    # your code, should convert target object
    return target

# you should keep following code
target = {target}
print(process(target))
2.2 Python コードを中間コードへ変換

コンパイラ・フロントエンドは Python の ast モジュールを使用し、AI Agent が出力した Python コードを構文解析して S 式で記述された中間コードを生成します。 この段階で文法エラーや事前に定めた制約に出力コードが従っていない場合にはエラーを返し、AI Agent にフィードバックします。 また、S 式で設計された中間コードは様々な用途に応用可能な仕様になっています。

2.3 中間コードをAutomationの処理構造に変換

コンパイラ・バックエンドは S 式フォーマットの中間コードを Automation の処理構造に変換します。 Automation の処理構造は JSON で記述され、ロカルメ・オーダー本体で実行されます。

構築されたAutomation


3. プロダクトでの構成例

実際のプロダクトでは、今回紹介した AI Agent 機構は独自の Agent OS(仮称)の上で独立したマイクロサービスとして動作し、Automation の実行系などは Go で実装されたロカルメ・オーダー本体で実行されるようになっています。


4. まとめと展望

  • 本手法のメリット: コンパイラ技術を使用して、ある程度 AI Agent の安定性を担保できる
  • 課題: 相変わらず実装が複雑で運用が大変
  • 次の一手: 今回はコンパイラ部分のみにフォーカスしてご紹介しました。そのうちアーキテクチャ図には記載したもののご紹介していなかった逆コンパイラ部分の仕組みもご紹介したいと思います。こちらはいわゆる human‑in‑the‑loop を実現する部分で、少し? 特殊な実装になっています ※ 他のコンテンツとの兼ね合いで、順番が前後する可能性があります ;-)

参考文献

AI Agent向けSandbox実装

はじめましての方も、お久しぶりの方もこんにちは。スパイスコード代表の中河です。 スパイスコードは、「ロカルメ・オーダー」

order.localmet.com

というAI Agentを内包したERPサービスを開発・提供しているスタートアップです。 近年、AI Agentという言葉を耳にする機会が増えましたが、実は私たちはこのブームが来る前から、AI Agentの実用化に向けた開発に取り組んできました(例えば、本日ご紹介する機能の実装を行っていたのは2023年12月〜2024年1月頃です)。そして現在では、ERPの中核機能として、AI Agentを実際にお客様にご利用いただいています。 本ブログでは、私たちのAI Agentがどのような仕組みで動いているのか、そして他のAgentとは何が異なるのかについて、ご紹介していく予定です。 第1回目となる今回は、「AI Agentが生成したコードを安全に実行する独自のSandbox機構」について、その概要をご紹介します。

Agent Architecture

TL;DR: ロカルメ・オーダーでは ptrace を使った独自のSandbox機構を運用しています。本稿は短いサンプルコードでどのようにサンドボックスを構築しているのか?をご紹介いたします。

1. なぜいま “軽量サンドボックス” なのか

  • Agent 時代の要請: LLM エージェントは外部プラグインや自己生成コードを即座に実行する─安全な実行環境が必須。

  • 実行情報のフィードバック: ptrace で取得したシステムコールやリソース統計をリアルタイムに Agent へ返送し、自己改善するループを構築できる。

  • 実行環境の柔軟性: ptrace ベースはイメージ再ビルド不要でポリシーをコードから動的変更可能。オンプレや制約の厳しい閉域ネットワーク環境でも、Python 実行権さえあればそのまま導入できる。CI・サーバレス・ローカルデバッグまで同一コードで再利用可。

1.1 類似実装・活用例

ツール/プロジェクト 主な用途 ptrace の役割
gVisor (runsc, 旧 ptrace モード) コンテナ向けユーザ空間カーネル すべての syscall を ptrace でフックし、Go 実装の仮想カーネルで処理
LangChain Sandbox Agent向けsandbox Python codeをWebAssemblyにコンパイルして実行

2. ptraceとは?

ptrace() は トレーサ (親) がトレース対象プロセス (子) を観察・制御できる LinuxUNIX 系 OS のシステムコールです。レジスタやメモリの読み書き、シグナル挿入、システムコール前後での割り込み停止などを行えるため、デバッガ (gdb)、システムコールトレーサ (strace)、軽量サンドボックス、故障解析ツールなどの基盤として利用されています

2.1 主要リクエストと用途(代表例)

区分 リクエスト (req) 典型的な用途・ポイント
トレース開始 PTRACE_TRACEME 子→親へ「自分をトレースしてほしい」と通知。execve() 直後に gdb などが使う入口。
アタッチ/デタッチ PTRACE_ATTACH / PTRACE_DETACH 既存プロセスの動的アタッチと解除。解除時はシグナルや PTRACE_O_EXITKILL で安全に終了させることも可能。
停止制御 PTRACE_CONT / PTRACE_SYSCALL / PTRACE_SINGLESTEP 連続実行・システムコール毎停止・命令単位ステップ実行。監査・改竄ポイントとして SYSCALL 前後 で介入できる。
レジスタ I/O PTRACE_GETREGS / PTRACE_SETREGSx86 以外は PTRACE_GETREGSET 戻り値を -EPERM に書き換える、引数を編集するなど動的パッチを実施。
メモリ I/O PTRACE_PEEKDATA / PTRACE_POKEDATA, PTRACE_PEEKUSER ダンプ取得、コードインジェクション、データ改竄。
イベント通知 PTRACE_O_TRACECLONE TRACEFORK TRACEEXEC TRACESECCOMP など clone(2) / fork(2) / execve(2) / seccomp 発火時点で自動停止し、親がハンドリングできる。
安全終了 PTRACE_O_EXITKILL トレーサ異常終了時に tracee を自動 SIGKILLstrace --kill-on-exit が内部で利用 (man7.org)。

2.2 典型的なイベントループ (高水準フロー)

  1. waitpid(-1, &status, __WALL) で子の停止を待つ。

  2. WIFSTOPPED(status) のときかつ WSTOPSIG(status) が SIGTRAP+0x80 (PTRACE_O_TRACESYSGOOD) → システムコール入口 or 退出、同じくWIFSTOPPED(status) のときかつ status>>16 に PTRACE_EVENT_* が入っていれば PTRACE_GETEVENTMSG で追加情報を取得。

  3. システムコール入口なら PTRACE_GETREGS で orig_rax (x86-64)、第一引数(rdi など) を読み取る。

  4. ホワイトリスト外なら 戻り値を改竄 (PTRACE_SETREGS) したり PTRACE_KILL / SIGKILL を送る。

  5. 次の停止条件に合わせて PTRACE_SYSCALL または PTRACE_CONT で再開。

性能コスト: system call ごとに 2 回 ユーザ空間⇆カーネルを往復。I/O 多用処理は体感で~3× 遅れる。

なぜseccompを使用しないのか?: 単純なsystem callの許可/不許可だけではなく、Agent向けの他の処理も行っているからです(ここは別の機会にご紹介..)


3. Sandboxアーキテクチャ・実装

┌ sandbox.py (parent tracer)
│  ├─ fork → child (untrusted code)
│  ├─ waitSignals ⇆ syscall enter/exit
│  └─ timeout / rlimit / decision engine
└─→ 親が子の終了コードを返して終了

構成としては、コードを実行する worker プロセスと、その挙動を監視する親プロセス(Tracerプロセス)の二層に分かれています。Tracerプロセスでは、syscall(システムコール)の呼び出しを検知・制御するほか、必要に応じた各種セキュリティ処理を担っています。

3.1 実装(抜粋)

実装の主要な部分を抜粋してご紹介します。 本Sandboxはpython-ptrace https://pypi.org/project/python-ptrace/ を使用したPure Pythonで実装されており、対象のPythonコードを exec(3) を使わずに実行できる構造になっています。この仕組みにより、対象コードが利用するライブラリや機械学習モデルなどを、事前にロードしておくことが可能となり、より効率的かつ柔軟な実行環境を実現しています。

  def _spawn_sandbox_worker(self, server: socket.socket) -> None:
      pid, r, w = None, None, None
      try:
          r, w = os.pipe()
          pid = os.fork()   # MLモデル等を使用する場合はCoWを効かせる為、forkより前にロードする

          if pid == 0:
              try:
                  self._sync_child(r, w)
                  r, w = None, None

                  self._start_sandbox_worker(server)
              finally:
                  os._exit(0)
      finally:
          self._sync_parent(pid, r, w)

      self._logger.debug(f"Spawned sandbox worker {pid}")

  def _watch_sandbox_worker(self, pid: int, status: int) -> None:
      if os.WIFSTOPPED(status) and os.WSTOPSIG(status) == signal.SIGTRAP:
          reg = self._get_registers(pid)
          nr = get_syscall_number(reg)

          if not is_approved_syscall(pid, nr, reg):   # syscallが許可されたものか判断
              self._logger.warning(f"Unapproved syscall PID:{pid},SYSCALL_NR:{nr}")
              ptrace_kill(pid)
              return

      ptrace_syscall(pid)

  def _get_registers(self, pid: int) -> linux_reg | freebsd_reg | openbsd_reg:
      if HAS_PTRACE_GETREGS or HAS_PTRACE_GETREGSET:
          return ptrace_getregs(pid)

      words: List[Any] = []
      num_words = sizeof(ptrace_registers_t) // CPU_WORD_SIZE
      for offset in range(num_words):
          word: Any = ptrace_peekuser(pid, offset * CPU_WORD_SIZE)
          bytes = word2bytes(word)
          words.append(bytes)

      bytes = "".join(words)
      return bytes2type(bytes, ptrace_registers_t)

...

def get_syscall_number(
    reg: linux_reg | freebsd_reg | openbsd_reg,
) -> Any:
    if platform.machine() == "x86_64":
        return reg.orig_rax
    elif platform.machine() == "aarch64":   # docker on mac 環境用
        return reg.r8
    else:
        raise NotImplementedError


def is_approved_syscall(pid: int, nr: int, reg: linux_reg | freebsd_reg | openbsd_reg) -> bool:
    if platform.machine() == "x86_64":
        name = X86_64_SYSCALL_WHITE_LIST.get(nr)
    elif platform.machine() == "aarch64":
        name = AARCH64_SYSCALL_WHITE_LIST.get(nr)
    else:
        raise RuntimeError("Unsupported platform")

    if name is None:
        return False

    ...  # プロダクトではここでsystem callのgateway処理などを行っている

    return confirm_func(pid, reg)

実装上の工夫として、共通で使用され、かつメモリ消費の大きい機械学習モデルや大規模データは、fork前にロードする設計を採用しています。 これにより、Copy-on-Write(CoW)機構が有効に働き、子プロセス側で不要なメモリ確保が発生せず、全体として効率的なメモリ利用が可能になります。


4. プロダクトでの構成例

実際のプロダクトでは、Sandbox本体は「Server」と、Agentがツールとして利用する「Client」に分かれた構成になっています。 両者の通信は Unix Domain Socket を用いており、Sandbox自体はKubernetes環境上でSidecarとしてデプロイされています。


5. まとめと展望

  • 本手法のメリット: ユーザランドPython実装だけでAgent用の隔離環境が実現できる
  • 課題: ptrace オーバーヘッド、カーネル exploit への脆弱性、安全性を適切担保する為にはsystem call実装の理解が必要
  • 次の一手: 今回は、Sandbox機能にフォーカスしてご紹介しました。次回は、Agentの再現性を高めるために、処理系からどのような実行時情報をAgent側にフィードバックしているのかについて、ご説明できればと考えています。 ※他のコンテンツとの兼ね合いで、順番が前後する可能性があります;-)

参考文献