"Your previous response was not a valid tool call. You must call one of the
available tools: [search, lookup, answer]. Respond only with a valid tool call."

"You cannot call 'answer' yet. You must first complete: [search, lookup]."

pip install forge-guardrails

# With Anthropic support:
pip install "forge-guardrails[anthropic]"

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

# ── Tool Implementations ───────────────────────────────────────────────────────

def search_web(query: str) -> str:
    """Simulate a web search — replace with real search API."""
    return f"Top results for '{query}': [Result 1], [Result 2], [Result 3]"

def fetch_page(url: str) -> str:
    """Simulate fetching a page — replace with real HTTP client."""
    return f"Content of {url}: <article>Detailed content about the topic</article>"

def write_summary(content: str, format: str = "markdown") -> str:
    """Write a structured summary of gathered content."""
    return f"Summary ({format}):\n\n{content[:200]}..."

# ── Pydantic Parameter Schemas ─────────────────────────────────────────────────

class SearchParams(BaseModel):
    query: str = Field(description="The search query string")

class FetchParams(BaseModel):
    url: str = Field(description="The URL to fetch content from")

class SummaryParams(BaseModel):
    content: str = Field(description="The content to summarize")
    format: str = Field(default="markdown", description="Output format: markdown or plain")

# ── Workflow Definition ────────────────────────────────────────────────────────

research_workflow = Workflow(
    name="research_and_summarize",
    description="Research a topic online and produce a structured summary.",
    tools={
        "search_web": ToolDef(
            spec=ToolSpec(
                name="search_web",
                description="Search the web for information on a topic",
                parameters=SearchParams,
            ),
            callable=search_web,
        ),
        "fetch_page": ToolDef(
            spec=ToolSpec(
                name="fetch_page",
                description="Fetch and read the content of a web page",
                parameters=FetchParams,
            ),
            callable=fetch_page,
        ),
        "write_summary": ToolDef(
            spec=ToolSpec(
                name="write_summary",
                description="Write a structured summary of gathered content",
                parameters=SummaryParams,
            ),
            callable=write_summary,
        ),
    },
    # Guardrail: search and fetch must complete before write_summary is allowed
    required_steps=["search_web", "fetch_page"],
    terminal_tool="write_summary",
    system_prompt_template=(
        "You are a precise research assistant. Use the available tools in order: "
        "first search for relevant sources, then fetch the most promising page, "
        "then write a structured summary. Do not skip steps."
    ),
)

# ── Runner Setup ───────────────────────────────────────────────────────────────

async def main():
    # Backend: Ollama with Ministral-3 8B (recommended entry-point model)
    client = OllamaClient(
        model="ministral-3:8b-instruct-2512-q4_K_M",
        recommended_sampling=True,  # Forge's optimized sampling params for this model
    )

    # Context manager: tiered compaction, 8K token budget
    ctx = ContextManager(
        strategy=TieredCompact(keep_recent=3),   # Keep last 3 full turn pairs verbatim
        budget_tokens=8192,
        warn_threshold=0.85,                      # Log warning at 85% of budget
    )

    runner = WorkflowRunner(
        client=client,
        context_manager=ctx,
        max_iterations=15,           # Hard cap — prevents runaway loops
        on_message=lambda m: print(f"[{m.role}] {str(m.content)[:80]}..."),
        on_compact=lambda e: print(f"📦 Compacted: {e.tokens_before}→{e.tokens_after} tokens"),
    )

    result = await runner.run(
        research_workflow,
        "Research the latest developments in LLM agent guardrails"
    )

    print(f"\n✅ Workflow complete: {result.terminal_output}")

asyncio.run(main())

import asyncio
from forge.guardrails import Guardrails

async def run_agent_with_guardrails(user_message: str, call_llm, execute_tools):
    guardrails = Guardrails(
        tool_names=["search_web", "fetch_page", "write_summary"],
        required_steps=["search_web", "fetch_page"],
        terminal_tool="write_summary",
        max_retries=3,
    )

    messages = [
        {"role": "system", "content": "You are a research assistant. Use tools to answer."},
        {"role": "user",   "content": user_message},
    ]

    while True:
        response = await call_llm(messages)      # Your existing LLM call — unchanged
        result = guardrails.check(response)       # Forge guardrail check

        if result.action == "retry":
            # Malformed response — append targeted nudge and retry
            print(f"⚠️  Retry nudge: {result.nudge.content[:80]}...")
            messages.append({"role": result.nudge.role, "content": result.nudge.content})
            continue

        if result.action == "step_blocked":
            # Model tried to skip a required step — correct it
            print(f"🚫 Step blocked: {result.reason}")
            messages.append({"role": result.nudge.role, "content": result.nudge.content})
            continue

        if result.action == "fatal":
            # Max retries exceeded or unrecoverable error
            raise RuntimeError(f"Agent failed: {result.reason}")

        # result.action == "execute" — tool calls are valid, execute them
        tool_outputs = execute_tools(result.tool_calls)

        # Tell Forge which steps completed (step enforcement state tracking)
        is_done = guardrails.record([tc.tool for tc in result.tool_calls])

        for tc, output in zip(result.tool_calls, tool_outputs):
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": str(output)})

        if is_done:
            print("✅ Workflow complete!")
            break

from forge.guardrails import ResponseValidator, StepEnforcer, ErrorTracker

# Instantiate individual guardrail components for full control
validator = ResponseValidator(
    tool_names=["search_web", "fetch_page", "write_summary"]
)
enforcer = StepEnforcer(
    required_steps=["search_web", "fetch_page"],
    terminal_tools=frozenset(["write_summary"])
)
errors = ErrorTracker(
    max_retries=3,
    max_tool_errors=2    # Abort after 2 consecutive tool execution failures
)

async def custom_agent_loop(messages, call_llm, execute_tool):
    while True:
        response = await call_llm(messages)

        # Step 1: Validate response structure + rescue parse if needed
        val_result = validator.validate(response)

        if val_result.needs_retry:
            if errors.retry_budget_exhausted():
                raise RuntimeError("Max retries reached — aborting agent loop.")
            errors.record_retry()
            messages.append({
                "role": val_result.nudge.role,
                "content": val_result.nudge.content
            })
            continue

        # Step 2: Enforce step ordering constraints
        step_check = enforcer.check(val_result.tool_calls)

        if step_check.needs_nudge:
            messages.append({
                "role": step_check.nudge.role,
                "content": step_check.nudge.content
            })
            continue

        # Step 3: Execute tools and track outcomes for error budget
        for tc in val_result.tool_calls:
            success = execute_tool(tc)
            enforcer.record(tc.tool)
            errors.record_result(success=success)

            if enforcer.is_terminal(tc.tool):
                return    # Reached terminal tool — workflow complete

# Option A: External mode — you manage llama-server, Forge proxies it
llama-server -m ./Ministral-3-8B-Instruct-2512-Q8_0.gguf \
  --jinja -ngl 999 --port 8080

python -m forge.proxy --backend-url http://localhost:8080 --port 8081

# Option B: Managed mode — Forge starts llama-server and the proxy together
python -m forge.proxy \
  --backend llamaserver \
  --gguf ./Ministral-3-8B-Instruct-2512-Q8_0.gguf \
  --port 8081

from openai import OpenAI

# Point at Forge proxy instead of the model server directly
client = OpenAI(
    base_url="http://localhost:8081/v1",
    api_key="not-needed-for-local"
)

# This identical code works whether the backend is a raw 8B local model
# (with Forge guardrails applied transparently) or a frontier API
response = client.chat.completions.create(
    model="ministral-3:8b-instruct-2512-q4_K_M",
    messages=[
        {"role": "system", "content": "You are a precise research assistant."},
        {"role": "user",   "content": "Search for recent papers on LLM agent guardrails."}
    ],
    tools=[
        {
            "type": "function",
            "function": {
                "name": "search_papers",
                "description": "Search for academic papers on a topic",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "The search query"
                        },
                        "max_results": {
                            "type": "integer",
                            "default": 5,
                            "description": "Maximum number of results to return"
                        }
                    },
                    "required": ["query"]
                }
            }
        }
    ],
    tool_choice="auto"
)

print(response.choices[0].message)

{
  "name": "respond",
  "description": "Use this to send a text response to the user.",
  "parameters": {
    "type": "object",
    "properties": {
      "message": {
        "type": "string",
        "description": "Your text response to the user"
      }
    },
    "required": ["message"]
  }
}

from forge import ContextManager, TieredCompact, SlidingWindowCompact
from forge.context import NoCompact
from forge.context.hardware import detect_hardware

# ── VRAM-Aware Auto-Detection ─────────────────────────────────────────────────
hw = detect_hardware()
print(f"Detected VRAM: {hw.vram_gb:.1f} GB")
print(f"Recommended token budget: {hw.recommended_budget_tokens:,}")

# ── Strategy 1: TieredCompact (recommended for most agentic workflows) ─────────
# Keeps the last `keep_recent` full turn pairs verbatim.
# Summarizes or drops older turns to stay within budget.
# Best for: multi-step task workflows where recent context matters most.
ctx_tiered = ContextManager(
    strategy=TieredCompact(
        keep_recent=3,          # Always preserve last 3 complete turn pairs
        summary_tokens=256,     # Token budget for summarizing dropped turns
    ),
    budget_tokens=hw.recommended_budget_tokens,
    warn_threshold=0.85,        # Log warning when 85% of budget is used
)

# ── Strategy 2: SlidingWindowCompact (for long-running conversational agents) ──
# Maintains a fixed-size rolling window; oldest messages are dropped first.
# Best for: persistent chat sessions where old context is genuinely stale.
ctx_sliding = ContextManager(
    strategy=SlidingWindowCompact(window_size=10),  # Keep last 10 messages
    budget_tokens=4096,
)

# ── Strategy 3: NoCompact (for debugging or short workflows) ──────────────────
ctx_none = ContextManager(
    strategy=NoCompact(),
    budget_tokens=16384,     # Warn only — never compact
)

# ── Compaction Event Callback ─────────────────────────────────────────────────
def on_compact(event):
    """Monitor compaction events for observability."""
    print(
        f"📦 Context compacted: {event.tokens_before:,} → {event.tokens_after:,} tokens | "
        f"Dropped {event.messages_dropped} messages, kept {event.messages_kept} verbatim"
    )

runner = WorkflowRunner(
    client=client,
    context_manager=ctx_tiered,
    on_compact=on_compact,
)

from forge.context import filter_transient_messages

# After a workflow completes, clean the session history before the next task:
clean_history = filter_transient_messages(
    messages=session.history,
    keep_terminal_outputs=True,           # Preserve final summaries and answers
    drop_intermediate_tool_calls=True,    # Drop search/fetch intermediate steps
)

# Feed clean_history into the next workflow as the starting context
next_result = await runner.run(next_workflow, next_task, history=clean_history)

# Start llama-server first (separate terminal)
llama-server -m ./Ministral-3-8B-Instruct-2512-Q8_0.gguf \
  --jinja -ngl 999 --port 8080

# Run eval suite — 10 runs per scenario for statistical confidence
python -m tests.eval.eval_runner \
  --backend llamaserver \
  --backend-url http://localhost:8080 \
  --runs 10 \
  --verbose

# Generate a human-readable report
python -m tests.eval.report eval_results.jsonl

import os
from forge import OllamaClient, AnthropicClient, WorkflowRunner

# Swap backends with a single environment variable
USE_LOCAL = os.getenv("FORGE_BACKEND", "local") == "local"

client = (
    OllamaClient(
        model="ministral-3:8b-instruct-2512-q4_K_M",
        recommended_sampling=True,
    )
    if USE_LOCAL else
    AnthropicClient(
        model="claude-opus-4-5",
        api_key=os.environ["ANTHROPIC_API_KEY"],
    )
)

# All Forge guardrail logic applies identically to both backends
runner = WorkflowRunner(client=client, context_manager=ctx)