OpenAI の Structured Outputs 構造化出力のプロセスを outlines から学ぶ

8月6日、OpenAI 社は新機能 [Structured Outputs](https://openai.com/index/introducing-structured-outputs-in-the-api/) を公開しました。これにより Function calling 等でも確実に出力を構造化できるようになり、プログラムとの連携をより確実に行えます。

`response_format` に Pydantic モデルをそのまま投入できますし、`response_format` に `json_schema` を使用して JSON スキーマを指定できるようになりました。

https://qiita.com/nohanaga/items/8f66bb759639aca88a6f

`gpt-4o-2024-08-06` モデルの Structured Outputs の出力結果が送信した入力スキーマと完全に一致、つまり 100% の精度を実現したとのことです。この精度を実現した理由として、2 つのアプローチを採用したと OpenAI 社は言っています。気になるのは後半、決定論的なエンジニアリング ベースのアプローチというところ。

>JSON スキーマに一致するモデル出力の信頼性を向上させるために、2 つのアプローチを採用しました。まず、最新のモデルをトレーニングして、gpt-4o-2024-08-06 複雑なスキーマを理解し、それらに一致する出力を最も効果的に生成する方法を学びました。ただし、モデルの動作は本質的に非決定論的です。このモデルのパフォーマンスは向上しましたが (ベンチマークでは 93%)、それでも、開発者が堅牢なアプリケーションを構築するために必要な信頼性を満たしていませんでした。そのため、モデルの出力を制限して 100% の信頼性を実現する、決定論的なエンジニアリング ベースのアプローチも採用しました。

https://openai.com/index/introducing-structured-outputs-in-the-api/

## Azure OpenAI Service の `gpt-4o(2024-08-06)` も対応
米国の全地域およびスウェーデン中央で、標準およびグローバルデプロイ両方で利用が可能です。このモデルは、`2024-08-01-preview` 推論 API と併用することで、Structured Outputs のサポートが追加されます。

# 決定論的なエンジニアリング ベースのアプローチとは
## 制約された出力
Structured Outputs は提供されたスキーマに従って有効なトークンのみにモデルを制約する機能を実装しています。「制約された出力」とは、モデルが正しい形式（この場合は JSON スキーマ）を保ちながらテキストを生成するように、選べるトークンを制限することです。つまり、生成している途中で「次に来るべきトークンはこれ！」という選択肢を絞り込むということですね。例えば、`{` を生成した後には、次に来るべきトークンは `"key"` のような文字列で、再び `{` は選べないようにします。

このために、最初に与えられたスキーマを使って、どのタイミングでどのトークンが有効かを定義するルール＝**文法**を事前に計算します。この文法を使って、モデルが常に正しいトークンを選ぶようにします。

OpenAI によるとこの機能は、構造化出力の OSS である outlines や Lark にインスピレーションを得たとのことです。インスピレーションの元となっている outlines などを利用して、どのように構造化出力を実現しているのか実際に動かして確かめてみたいと思います。

:::note warn
注意
以下は実際に OpenAI がやっていることと関係ありません😛
:::


## 処理の流れ
1. スキーマの解析
1. 文法生成
1. トークン生成
1. 有効なトークンのリストを更新
1. トークンマスキングの適用
1. 次のトークンを選択
1. プロセスを繰り返す

[outlines](https://github.com/outlines-dev/outlines): LLM の出力結果に高レベルの制約を指定するメカニズムを提供
[Lark](https://github.com/lark-parser/lark): 文法規則に基づいてテキストを解析し解析ツリーを構築
[guidance](https://github.com/guidance-ai/guidance): 正規表現や CFG などを使用して生成を制限できる
[instructor](https://github.com/jxnl/instructor): 構造化された出力を簡単に操作できるライブラリ

## 文法規則に基づいてテキストを解析
文法規則には、文脈自由文法（CFG: Context-Free Grammar）が用いられます。CFG の記述には、Azure AI Search でもおなじみ EBNF (拡張バッカス・ナウア記法) を用いて記述します。CFG は、文法規則を用いて文字列を生成するための形式的なルールセットであり、これらの規則は「生成規則」として定義され、非終端記号と終端記号から構成されます。CFG はパーサーによって**入力文字列が文法に従っているかどうかを判断するために使われます**。

outlines は Lark ライブラリを使用して、LLM が文法の言語でテキストを生成するようにします。そのため、EBNF 構文に基づいて、Lark が理解できる形式で定義された文法を使用する必要があります。今回、outlines が LLM の生成に制約を課している部分を分析し、手元でシミュレートします。

`pip install outlines`

### 制約ルール(CFG)
例1：括弧内（）に数字が入るような文字列になるような制約をかける

```yaml
start: expr

expr: "(" NUMBER ")"
NUMBER: /[0-9]+/
```

expr ルールでは、数値を括弧で囲んだ形式 (NUMBER) を期待しています。NUMBER が 1 つ以上の数字から構成されていることを表現するため、正規表現 `/[0-9]+/` を使用しています。この文法を使うことで、`(7)` のような文字列が正しい形式として解析されるようになります。



# 有限状態マシン（FSM） の構築
有限状態マシン（FSM）は、各状態で次にどのトークンを生成すべきか、またはどの状態に遷移すべきかを決定します。生成されたテキストや現在の状態に基づいて、この FSM を使用して次の指示を取得しています。

ひとまず CFG とモックのトークナイザーを使って FSM を作成します。シミュレーションで使用する文字列だけトークンマッピング表にして `vocabulary` に格納します。`{"文字":トークンID}` となっています。

```python
from outlines.fsm.guide import CFGGuide, Generate, Write

class MockTokenizer:
    # モックのトークナイザーを定義します。簡略化された辞書とトークン変換メソッドを持っています。
    vocabulary = {"(": 1, ")": 2, "7": 3, "]": 4, "EOS": 5}
    special_tokens = {"EOS"}
    eos_token = "EOS"
    eos_token_id = 5
    
    def convert_token_to_string(self, token):
        return token  # トークンを文字列に変換するメソッド（ここではそのまま返します）

    @property
    def inverse_vocabulary(self):
        return {v: k for k, v in self.vocabulary.items()}  # 辞書のキーと値を逆転させたものを返します

    def decode(self, token_ids):
        return [self.inverse_vocabulary[t] for t in token_ids]  # トークンIDを対応する文字列に変換します


tokenizer = MockTokenizer()
fsm = CFGGuide(cfg_str, tokenizer)
```

`CFGGuide` は文法文字列 `cfg_str` を解析して FSM を構築し、次に生成するべきトークンやアクションを決定するためのロジックを含んでいるクラスです。内部では正規表現を使用して文法規則の一部を処理しています。

https://github.com/outlines-dev/outlines/blob/5e8f7709e3cecd02943120ed01420f00159cedbc/tests/test_grammars.py

# トークンの逐次生成をシミュレート
## 1 文字目の提案
`get_next_instruction` メソッドは、現在の状態と生成済みのトークン列に基づいて、FSM から次の遷移先や生成すべきトークンを提案します。まだモデルが何も生成していない初期の状態からスタートします。

`fsm.start_state` は `0` です。

```python
state = fsm.start_state  # FSM の初期状態を取得
instruction = fsm.get_next_instruction(state)  # 次の指示を取得します
print(instruction)  # 指示を表示します
```

>Compiling FSM index for all state transitions: 100%|██████████| 2/2 [00:00<00:00, 85.06it/s]
Generate(tokens=[1])

トークン ID:`1`が帰ってきました。ID: `1` はモックトークナイザーで `(` が対応します。制約ルールによって最初の文字は `(` しか許されないということです。

実際は LogitsProcesser が生成された logits に対して動的にマスキングを適用しています。

## 状態遷移①
`get_next_state` は生成されたトークンが現在の文法状態に合致するかどうかを確認し、次の状態に遷移します。文法に合致しない場合、別の状態遷移が選択されるか、トークンの再生成が必要になります。まずは LLM が `(` を生成したと仮定して、`token_id` に `(` を表す `1` を指定します。
```python
state = fsm.get_next_state(state=fsm.start_state, token_id=1)
print(fsm.generation)
print(state)
```

> (
> state: 1

## 2 文字目の提案
```python
instruction = fsm.get_next_instruction(state)
print(instruction)
assert isinstance(instruction, Generate)
```

>Compiling FSM index for all state transitions: 100%|██████████| 2/2 [00:00<00:00, 4566.47it/s]
Generate(tokens=[3])

トークン ID: `3` が返ってきました。ID `3` は NUMBER の `7` が対応します。ちゃんと `(` の次に来るべき文字が NUMBER であるという制約ルールに従った提案ができていますね。

## 状態遷移②
`token_id` に 2 番目の生成結果 `7` を表す ID `3` を指定します。
```python
state = fsm.get_next_state(state=fsm.start_state, token_id=3)
print(fsm.generation)
print(state)
```

> (7
> state: 1

## 3 文字目の提案
`(7` の次に来るトークンを提案します。
```python
instruction = fsm.get_next_instruction(state)
print(instruction)
assert isinstance(instruction, Generate)
```

>Compiling FSM index for all state transitions: 100%|██████████| 2/2 [00:00<00:00, 4718.00it/s]
Generate(tokens=[3, 2])

面白いことになってきました。トークンID: `3`と`2`が返ってきました。つまり、 `)` か `7` が許されるということです！

## 状態遷移③

ここで、さらに `get_next_state` の `token_id` に `3` を指定すれば、`fsm.generation`は `(77` になるはずですね。

```python
## 実際の値を投入して、次の状態へ遷移させる
state = fsm.get_next_state(state=state, token_id=3)
print(fsm.generation)
print(state)
```

> (77
> state: 1

## 状態遷移*n*

あとの結果は同上となります。何回か繰り返し、最後に `token_id` に `2` つまり `)` を指定して括弧を閉じます。

```python
state = fsm.get_next_state(state=state, token_id=2)
print(fsm.generation)
print(state)
```

> (777)
> state: 1

## 終端文字の提案
前回の状態は `)` でしたので、制約ルール上 `)` に続く文字はありません。そのため終端文字 `EOS` である ID `5` が返ります。

```python
instruction = fsm.get_next_instruction(state)
print(instruction)
```

>Write(tokens=[5])

```python
state = fsm.get_next_state(state=state, token_id=5)
print(fsm.generation)
print(state)
```

>(777)
>-1

`status` が `-1` となり、`fsm.is_final_state(state)` が `True` となります。最後の状態まで遷移したのでこれで終了です。


# 正規表現アプローチ
正規表現と FSM の同等性を利用した `RegexGuide` クラスがあります。outlines は Pydantic オブジェクトや JSON スキーマによる制約に対応しており、内部では、`build_regex_from_schema` によって正規表現に変換されて、その後 `RegexGuide` で FSM へ変換されます。

とりあえず OpenAI のページに書いてある JSON スキーマ例を入力してみる。

```py
schema = """{
  "type": "object",
  "properties": {
    "value": { "type": "number" }
  },
  "required": ["value"],
  "additionalProperties": false
}
"""

regex_str = build_regex_from_schema(schema, whitespace_pattern)
print(regex_str)
```

>\{[ ]?"value"[ ]?:[ ]?((-)?(0|[1-9][0-9]*))(\.[0-9]+)?([eE][+-][0-9]+)?[ ]?\}

JSONスキーマが正規表現になったので、`RegexGuide` クラスに入れる。

```python
class MockTokenizer:
    # モックのトークナイザーを定義します。簡略化された辞書とトークン変換メソッドを持っています。
    vocabulary = { "{": 1, "}": 2, "\"value\"": 3, ":": 4, "0": 5, "1": 6, "2": 7, "3": 8, "4": 9, "5": 10, "6": 11, "7": 12, "8": 13, "9": 14, "EOS": 15,  }
    special_tokens = {"EOS"}
    eos_token = "EOS"
    eos_token_id = 15
    ...

tokenizer = MockTokenizer()  # 上記で定義したMockTokenizerをインスタンス化します
regex_string = '\{[ ]?"value"[ ]?:[ ]?((-)?(0|[1-9][0-9]*))(\.[0-9]+)?([eE][+-][0-9]+)?[ ]?\}'
fsm = RegexGuide(regex_string, tokenizer)
```

`RegexGuide` だと `states_to_token_maps` が参照できます。これは FSM の各状態から次に遷移可能なトークンとそれに対応する遷移先の状態を保持する辞書です。

構造は `{origin_state : {token_id : next_state}}` となっています。最初の時点で遷移先のマッピングができているんですね。


```python
fsm.states_to_token_maps

{0: {1: 1},
 1: {3: 9},
 9: {4: 11},
 11: {10: 15,
    6: 15,...
 14: {2: 19},
 15: {10: 15,
    5: 15,}}
```
ちゃんと `{"value": { "type": "number" }}` に従うようにできていますね！　あれ、こっちは JSON スキーマ -> CFG ではないのか。

>これを実現するために、提供された JSON スキーマを文脈自由文法 (CFG) に変換します。

🤔

>この問題に対する代替アプローチでは、制約付きデコードに有限状態マシン (FSM) または正規表現 (通常は FSM を使用して実装) がよく使用されます。

# 参考：Lark による 逐次パース
outlines は Lark ライブラリを内部で使用しています。

```python
from lark import Lark, Transformer, v_args
import json

# Lark で使用する文法を定義します
cfg_str = """
start: expr

expr: "(" NUMBER ")"
NUMBER: /[0-9]+/
"""

# パーサーを作成します
parser = Lark(cfg_str, parser='lalr', lexer="contextual")
```

https://lark-parser.readthedocs.io/en/latest/classes.html#lark.Lark.parse_interactive

## 逐次チェックと構文解析
入力文字列から生成されたトークンのリストを確認。
```python
interactive = parser.parse_interactive('(')
interactive.exhaust_lexer()
```

>[Token('LPAR', '(')]

次の状態に遷移する可能性のあるトークンのセットを確認。
```python
interactive.accepts()
```

>{'NUMBER'}

これが次のトークン生成時に使用され、生成プロセスが文法に従って進行することを保証します。


# Github
https://github.com/nohanaga/outlines_samples/blob/main/sample_cfg.ipynb