Forge guardrails入門：ローカル8Bモデルのエージェント精度を53%→99%に引き上げるオープンソースガードレールフレームワーク

import asyncio
from pydantic import BaseModel, Field
from forge import (
    Workflow, ToolDef, ToolSpec,
    WorkflowRunner, OllamaClient,
    ContextManager, TieredCompact,
)

def get_weather(city: str) -> str:
    return f"72°F and sunny in {city}"

class GetWeatherParams(BaseModel):
    city: str = Field(description="City name")

workflow = Workflow(
    name="weather",
    description="Look up weather for a city.",
    tools={
        "get_weather": ToolDef(
            spec=ToolSpec(
                name="get_weather",
                description="Get current weather",
                parameters=GetWeatherParams,
            ),
            callable=get_weather,
        ),
    },
    required_steps=[],
    terminal_tool="get_weather",
    system_prompt_template="You are a helpful assistant.",
)

async def main():
    client = OllamaClient(
        model="ministral-3:8b-instruct-2512-q4_K_M",
        recommended_sampling=True,
    )
    ctx = ContextManager(
        strategy=TieredCompact(keep_recent=2),
        budget_tokens=8192,
    )
    runner = WorkflowRunner(client=client, context_manager=ctx)
    await runner.run(workflow, "What's the weather in Paris?")

asyncio.run(main())

# 自分でllama-serverを起動して連携
python -m forge.proxy --backend-url http://localhost:8080 --port 8081

# クライアント側（OpenAI互換API）
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8081/v1")
# → 以降、通常のOpenAI API呼び出しでForgeガードレールが透過適用

# Claude Code連携
ANTHROPIC_BASE_URL=http://localhost:8081 ANTHROPIC_AUTH_TOKEN=anything claude

from forge.guardrails import Guardrails

guardrails = Guardrails(
    tool_names=["search", "lookup", "answer"],
    required_steps=["search", "lookup"],
    terminal_tool="answer",
)

while True:
    response = my_custom_llm_call(messages)
    result = guardrails.check(response)
    if result.action == "retry":
        messages.append({"role": result.nudge.role, "content": result.nudge.content})
        continue
    if result.action == "step_blocked":
        messages.append({"role": "user", "content": result.nudge.content})
        continue
    if result.action == "fatal":
        raise RuntimeError(result.reason)
    execute(result.tool_calls)
    if guardrails.record([tc.tool for tc in result.tool_calls]):
        break

# インストール
pip install forge-guardrails
# Anthropic連携時
pip install "forge-guardrails[anthropic]"

# バックエンド準備（Ollama）
ollama pull ministral-3:8b-instruct-2512-q4_K_M

# Proxyモード起動
python -m forge.proxy --backend ollama --port 8081

# 評価ハーネス実行（カスタムバックエンドの評価を追加可能）
python -m forge.eval