概述

Forge 是一个基于 LLM（大型语言模型）的工具调用框架，专注于构建可靠、可预测的多步骤 AI 工作流程。该项目由 Antoine Zambelli 开发，旨在解决 LLM 在工具调用场景中的常见问题，如响应格式不稳定、上下文长度失控、错误恢复困难等。

Forge 的核心设计理念是将 LLM 响应校验、上下文压缩、错误重试和步骤强制等机制封装为可配置的保护栏（Guardrails），使开发者能够专注于业务逻辑而非基础设施细节。资料来源：README.md

核心概念

Workflow（工作流）

Workflow 是 Forge 中的核心抽象，定义了一个完整的多步骤任务。它包含以下关键组件：

资料来源：src/forge/core/workflow.py:1-50

ToolDef 和 ToolSpec

ToolDef 将工具的模式定义与实际实现绑定在一起：

@dataclass
class ToolDef:
    spec: ToolSpec
    callable: Callable[..., Any]
    prerequisites: list[str | dict[str, str]] = field(default_factory=list)

前置条件（prerequisites）表达条件依赖关系：

• 字符串形式："read_file" — 任何对 read_file 的先前调用都满足要求
• 字典形式：{"tool": "read_file", "match_arg": "path"} — 需要使用相同 path 参数的先前调用

资料来源：src/forge/core/workflow.py:90-105

LLM 客户端

Forge 通过统一的客户端接口支持多个后端：

资料来源：README.md 资料来源：CONTRIBUTING.md 资料来源：src/forge/proxy/__main__.py:1-50

架构设计

系统组件图

graph TD
    subgraph 客户端层
        User[用户代码]
        Workflow[Workflow 定义]
    end
    
    subgraph 核心层
        Runner[WorkflowRunner]
        Guardrails[Guardrails 检查]
        Context[ContextManager]
    end
    
    subgraph 后端层
        Client[LLMClient]
        Proxy[ProxyServer]
        Server[ServerManager]
    end
    
    subgraph 支持服务
        Sampling[采样默认参数]
        Errors[错误追踪]
    end
    
    User --> Workflow
    Workflow --> Runner
    Runner --> Guardrails
    Runner --> Context
    Guardrails --> Client
    Context --> Client
    Client --> Proxy
    Proxy --> Server
    Server --> Ollama[(Ollama)]
    Server --> Llamafile[(Llamafile)]
    Runner --> Sampling
    Guardrails --> Errors

工作流程

sequenceDiagram
    participant U as 用户
    participant R as WorkflowRunner
    participant G as Guardrails
    participant C as ContextManager
    participant L as LLMClient
    
    U->>R: run(workflow, prompt)
    R->>L: send(messages)
    L-->>R: LLMResponse
    R->>G: check(response)
    G-->>R: CheckResult
    alt action == retry
        R->>G: 获取 nudge
        R->>L: send(messages + nudge)
    end
    alt action == step_blocked
        R->>G: 获取 nudge
        R->>L: send(messages + nudge)
    end
    alt action == fatal
        R-->>U: 抛出异常
    end
    R->>C: update(token_count)
    R->>L: 继续或结束

采样参数系统

推荐采样参数

Forge 维护了一个经过验证的模型采样参数映射表，包含以下模型家族：

• Qwen3 / Qwen3.5 / Qwen3.6
• Qwen3-Coder
• Gemma 4
• Mistral Small 3.2
• Devstral Small 2
• Ministral 3 Instruct + Reasoning
• Mistral Nemo
• Granite 4.0

每个参数条目都包含内联的 HuggingFace 模型卡片链接，确保数值经过逐一验证。

资料来源：src/forge/clients/sampling_defaults.py:1-40

采样策略四象限

资料来源：src/forge/clients/sampling_defaults.py:40-60

recommended_sampling 参数

client = OllamaClient(
    model="ministral-3:8b-instruct-2512-q4_K_M",
    recommended_sampling=True  # 启用推荐采样参数
)

启用后，客户端会自动应用 MODEL_SAMPLING_DEFAULTS 中的参数，包括：

• temperature
• top_p
• top_k
• min_p
• repeat_penalty
• presence_penalty

资料来源：README.md

上下文管理

ContextManager

ContextManager 负责管理对话历史的长度，防止超出模型的上下文窗口。

from forge import ContextManager, TieredCompact

ctx = ContextManager(
    strategy=TieredCompact(keep_recent=2),
    budget_tokens=8192
)

BudgetMode 预算模式

资料来源：src/forge/server.py:80-100

KV Cache 量化

Forge 支持 KV 缓存量化以减少显存占用：

资料来源：src/forge/server.py:30-35

KV Unified 模式

当 kv_unified=True 时，所有槽位共享单一 KV 缓存池，每个槽位可使用完整上下文长度。

Guardrails 系统

Guardrails 是 Forge 的核心保护机制，包含三个子组件：

组件架构

graph LR
    subgraph Guardrails
        V[ResponseValidator]
        E[StepEnforcer]
        T[ErrorTracker]
    end

ResponseValidator

验证 LLM 响应的格式和内容，支持 XML 救援解析（针对 Qwen Coder 等模型）。

StepEnforcer

强制执行必需的步骤顺序：

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

ErrorTracker

追踪错误并控制重试次数：

资料来源：src/forge/guardrails/guardrails.py:1-60

快速开始

安装

git clone https://github.com/antoinezambelli/forge.git
cd forge
python -m venv .venv
pip install -e ".[dev]"

基本使用示例

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. Use the available tools to answer the user.",
)

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())

资料来源：README.md

后端自动管理

Forge 可以自动启动和管理 LLM 后端：

from forge import setup_backend, BudgetMode

async def main():
    client = OllamaClient(model="ministral-3:14b-instruct-2512-q4_K_M")
    server, ctx = await setup_backend(
        backend="ollama",
        model="ministral-3:14b-instruct-2512-q4_K_M",
        budget_mode=BudgetMode.FORGE_FAST,
        client=client,
    )
    runner = WorkflowRunner(client=client, context_manager=ctx)
    # ... 运行工作流 ...
    await server.stop()

支持的后端：

• ollama — 使用 Ollama 服务
• llamaserver — 使用 llama.cpp 服务器
• llamafile — 使用单文件可执行模型

资料来源：src/forge/server.py:100-150

项目结构

forge/
├── src/forge/           # 库源码
│   ├── clients/         # LLM 后端适配器
│   ├── core/            # 工作流、运行器、消息、步骤
│   ├── context/         # 上下文管理和压缩
│   ├── prompts/         # 提示词模板和引导
│   ├── guardrails/      # 保护栏实现
│   ├── proxy/           # 代理服务器
│   └── tools/           # 内置工具
├── tests/
│   ├── unit/            # 单元测试（确定性）
│   └── eval/            # 评估测试（需要后端）
│       ├── scenarios/   # 评估场景定义
│       └── dashboard/   # React 评估仪表板
├── docs/
│   ├── decisions/       # 架构决策记录 (ADR)
│   └── results/         # 评估结果
├── examples/            # 使用示例
└── scripts/             # 辅助脚本

资料来源：CONTRIBUTING.md

版本历史

v0.6.0 (2026-04-29)

• 推荐采样参数 — 每个模型都有经过验证的采样配置
• 采样覆盖 — 支持 per-call 采样参数覆盖
• 代理采样透传 — 代理透传 OpenAI 兼容的采样字段
• 高级推理评估套件 — 支持 Gemma 4、Qwen 3.5 等模型的推理任务

v0.5.0 (2026-04-19)

• 消融研究运行器 — 支持自动化消融实验
• 移除硬编码温度 — OllamaClient 和 LlamafileClient 不再发送硬编码采样参数
• Granite 4.0 支持 — 支持 OpenAI 风格的 tool call 格式

v0.4.x

• Qwen Coder XML 救援解析
• 28 模型评估数据集
• llama.cpp 推理预算修复

资料来源：CHANGELOG.md

测试

单元测试

# 完整套件（865 个测试）
python -m pytest tests/unit/ -v --tb=short

# 带覆盖率
python -m pytest tests/unit/ --cov=forge --cov-report=term-missing

# 单文件测试
python -m pytest tests/unit/test_runner.py -v

集成测试

集成测试需要运行中的后端，跳过方式：

python -m pytest tests/ -m "not integration"

资料来源：CONTRIBUTING.md

相关资源

项目概述

Forge 是一个基于 Python 的 LLM 工作流框架，支持多种后端（Ollama、llamafile、llamaserver），提供工作流管理、上下文管理和工具调用等功能。安装与配置涉及环境准备、后端选择、客户端初始化以及可选的代理服务器部署。

资料来源：README.md:1-45

环境要求

系统要求

Python 版本

Forge 使用现代 Python 语法，包括类型联合操作符（|）等特性，明确要求 Python 3.12 或更高版本。

资料来源：CONTRIBUTING.md:1-15

安装步骤

标准安装

git clone https://github.com/antoinezambelli/forge.git
cd forge
python -m venv .venv
pip install -e ".[dev]"

.[dev] 标志安装包含开发依赖的完整包，包括测试框架 pytest。

资料来源：CONTRIBUTING.md:1-12

可选依赖

后端配置

Forge 支持三种 LLM 后端，后端配置通过 ServerManager 类统一管理。

支持的后端类型

资料来源：src/forge/server.py:1-50

后端初始化参数

ServerManager 类的构造函数接受以下参数：

def __init__(
    self,
    backend: str,                    # 后端类型
    port: int = 8080,                # 服务端口
    models_dir: str | Path | None = None,  # GGUF 文件目录
) -> None:

资料来源：src/forge/server.py:120-140

服务启动配置

start() 方法支持丰富的配置选项：

# 启动示例
server = ServerManager(backend="ollama", port=8080)
await server.start(
    model="qwen3:8b",
    mode="reasoning",
    extra_flags=["--reasoning-format", "auto"]
)

资料来源：src/forge/server.py:140-180

上下文长度解析

资料来源：src/forge/server.py:200-220

客户端配置

OllamaClient

from forge.clients import OllamaClient

client = OllamaClient(
    model="ministral-3:8b-instruct-2512-q4_K_M",
    recommended_sampling=True  # 使用推荐采样参数
)

资料来源：README.md:20-35

LlamafileClient

from forge.clients import LlamafileClient

client = LlamafileClient(
    gguf_path="/path/to/model.gguf",
    recommended_sampling=True
)

推荐采样参数

forge.clients.sampling_defaults 模块提供经过验证的模型推荐采样参数：

每个配置行包含指向 HuggingFace 模型卡的 URL，参数值经过逐一验证。

资料来源：src/forge/clients/sampling_defaults.py:1-80

采样参数策略

# 严格模式示例
params = apply_sampling_defaults(model, strict=True)

# 非严格模式示例
params = apply_sampling_defaults(model, strict=False)

资料来源：src/forge/clients/sampling_defaults.py:80-120

单次调用采样覆盖

send() 和 send_stream() 方法支持 sampling 字典参数，字段逐一合并到客户端实例级采样：

response = await client.send(
    messages,
    sampling={"temperature": 0.8, "top_p": 0.9}
)

资料来源：CHANGELOG.md:1-50

代理服务器配置

Forge 提供 ProxyServer 用于转发 OpenAI 兼容请求到后端。

CLI 参数

python -m forge.proxy --backend ollama --model qwen3:8b

资料来源：src/forge/proxy/__main__.py:1-80

代理采样穿透

代理将 OpenAI 兼容的采样参数字段透传到后端：

资料来源：CHANGELOG.md:1-50

工作流配置

Workflow 组件

from forge import Workflow, ToolDef, ToolSpec, WorkflowRunner, ContextManager, TieredCompact

workflow = Workflow(
    name="weather",
    description="查询城市天气",
    tools={
        "get_weather": ToolDef(
            spec=ToolSpec(
                name="get_weather",
                description="获取当前天气",
                parameters=GetWeatherParams,
            ),
            callable=get_weather,
        ),
    },
    required_steps=[],
    terminal_tool="get_weather",
    system_prompt_template="你是一个有帮助的助手。",
)

上下文管理器配置

ctx = ContextManager(
    strategy=TieredCompact(keep_recent=2),
    budget_tokens=8192
)

资料来源：README.md:1-50

完整运行示例

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="城市名称")

workflow = Workflow(
    name="weather",
    description="查询城市天气",
    tools={
        "get_weather": ToolDef(
            spec=ToolSpec(
                name="get_weather",
                description="获取当前天气",
                parameters=GetWeatherParams,
            ),
            callable=get_weather,
        ),
    },
    required_steps=[],
    terminal_tool="get_weather",
    system_prompt_template="你是一个有帮助的助手。",
)

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())

资料来源：README.md:15-55

常见问题

后端选择

版本信息

通过 importlib.metadata 暴露版本：

from importlib.metadata import version
print(version("forge"))

资料来源：CHANGELOG.md:1-50

环境准备

系统要求

安装步骤

git clone https://github.com/antoinezambelli/forge.git
cd forge
python -m venv .venv
pip install -e ".[dev]"

资料来源：CONTRIBUTING.md:1-10

运行测试验证安装

# 完整单元测试（865 个测试，无需 LLM 后端）
python -m pytest tests/unit/ -v --tb=short

# 单文件测试
python -m pytest tests/unit/test_runner.py -v

资料来源：CONTRIBUTING.md:14-22

核心概念

组件架构概览

graph TD
    A[用户请求] --> B[WorkflowRunner]
    B --> C[Workflow]
    C --> D[ToolDef + ToolSpec]
    D --> E[LLMClient]
    E --> F[Backend Ollama/llama.cpp/llamafile]
    F --> G[LLMResponse]
    G --> H[Guardrails]
    H --> I{校验结果}
    I -->|通过| J[执行工具]
    I -->|失败| K[重试/终止]
    J --> L[ContextManager]
    L --> B

核心组件表

资料来源：README.md:1-35, src/forge/core/workflow.py:1-75

快速开始示例

基础天气查询工作流

以下是一个完整的基础示例，展示了如何使用 Forge 构建天气查询工具调用工作流：

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}"

# 第二步：定义工具参数 Schema
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. Use the available tools to answer the user.",
)

# 第四步：初始化并运行
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())

资料来源：README.md:7-42

工作流执行流程

内部执行步骤

sequenceDiagram
    participant User as 用户
    participant Runner as WorkflowRunner
    participant Workflow as Workflow
    participant LLM as LLM Client
    participant Backend as LLM 后端
    participant Guardrails as Guardrails
    participant Tools as 工具函数
    participant Context as ContextManager

    User->>Runner: run(workflow, prompt)
    Runner->>Workflow: 获取工具定义和提示
    Runner->>LLM: 发送请求
    LLM->>Backend: API 调用
    Backend-->>LLM: LLMResponse
    LLM-->>Runner: LLMResponse
    Runner->>Guardrails: check(response)
    Guardrails->>Guardrails: 验证工具调用
    Guardrails-->>Runner: CheckResult
    alt 校验通过
        Runner->>Tools: 执行工具
        Tools-->>Runner: 工具结果
        Runner->>Context: 更新上下文
        Runner->>Runner: 循环直到完成
    else 校验失败
        Runner->>LLM: 发送重试提示
    end
    Runner-->>User: 工作流完成

ToolDef 与 ToolSpec 的关系

ToolDef 是核心数据结构，将工具的 Schema 与实际实现绑定：

@dataclass
class ToolDef:
    """将工具 Schema 绑定到其实现。
    
    Prerequisites 表达条件依赖："如果你调用这个工具，
    必须先调用工具 X"。
    """
    spec: ToolSpec           # 工具规范（名称、描述、参数）
    callable: Callable       # 实际执行的函数
    prerequisites: list[str | dict[str, str]] = field(default_factory=list)

资料来源：src/forge/core/workflow.py:60-75

多步骤工作流

带步骤要求的工作流

对于复杂任务，可以定义必须按顺序执行的步骤：

# 定义工具
def search(query: str) -> str:
    return f"Search results for: {query}"

def lookup(url: str) -> str:
    return f"Content from: {url}"

def answer(question: str) -> str:
    return "Final answer"

# 定义参数 Schema
class SearchParams(BaseModel):
    query: str

class LookupParams(BaseModel):
    url: str

class AnswerParams(BaseModel):
    question: str

# 构建需要按顺序执行的工作流
workflow = Workflow(
    name="research",
    description="Multi-step research workflow",
    tools={
        "search": ToolDef(
            spec=ToolSpec(name="search", description="Search", parameters=SearchParams),
            callable=search,
        ),
        "lookup": ToolDef(
            spec=ToolSpec(name="lookup", description="Look up URL", parameters=LookupParams),
            callable=lookup,
        ),
        "answer": ToolDef(
            spec=ToolSpec(name="answer", description="Provide answer", parameters=AnswerParams),
            callable=answer,
        ),
    },
    required_steps=["search", "lookup"],  # 必须先执行 search 和 lookup
    terminal_tool="answer",
    system_prompt_template="You are a research assistant.",
)

护栏（Guardrails）系统

Forge 内置护栏系统用于验证 LLM 响应：

graph TD
    A[LLMResponse] --> B[ResponseValidator]
    A --> C[StepEnforcer]
    A --> D[ErrorTracker]
    B --> E{检查结果}
    C --> E
    D --> E
    E -->|fatal| F[终止]
    E -->|retry| G[重试]
    E -->|step_blocked| H[步骤阻塞]
    E -->|execute| I[执行工具]

资料来源：src/forge/guardrails/guardrails.py:1-80

上下文管理

TieredCompact 策略

TieredCompact 是推荐的上下文压缩策略：

from forge import ContextManager, TieredCompact

ctx = ContextManager(
    strategy=TieredCompact(keep_recent=2),  # 保留最近 2 轮完整
    budget_tokens=8192
)

预算模式

资料来源：src/forge/server.py:1-50

常用配置

OllamaClient 配置

client = OllamaClient(
    model="ministral-3:8b-instruct-2512-q4_K_M",
    recommended_sampling=True  # 使用推荐的采样参数
)

推荐采样参数

recommended_sampling=True 时，Forge 会自动应用模型卡推荐的采样参数：

资料来源：src/forge/clients/sampling_defaults.py:1-60

完整项目结构

forge/
├── src/forge/               # 库源码
│   ├── clients/             # LLM 后端适配器
│   │   ├── ollama.py
│   │   ├── llamafile.py
│   │   └── sampling_defaults.py
│   ├── core/                # 核心组件
│   │   ├── workflow.py      # Workflow 和 ToolDef
│   │   ├── runner.py        # WorkflowRunner
│   │   ├── context.py       # 上下文管理
│   │   └── steps.py         # 步骤追踪
│   ├── guardrails/          # 护栏系统
│   └── prompts/             # 提示模板
├── tests/
│   ├── unit/                # 单元测试
│   └── eval/                # 评估工具
└── examples/
    └── foreign_loop.py      # 外部循环集成示例

资料来源：CONTRIBUTING.md:25-40

外部循环集成

如果你已有 LLM 调用逻辑，可以使用 Forge 的护栏组件进行验证：

from forge.guardrails import Guardrails, ErrorTracker

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

def handle_response(response):
    result = guardrails.check(response)
    
    if result.action == "fatal":
        return f"FATAL: {result.reason}"
    
    if result.action in ("retry", "step_blocked"):
        return f"{result.action}: {result.nudge.content[:80]}..."
    
    # 执行工具
    tool_calls = result.tool_calls
    executed = [tc.tool for tc in tool_calls]
    done = guardrails.record(executed)
    return f"executed {executed}" + (" -- DONE" if done else "")

资料来源：examples/foreign_loop.py:1-100

常见问题

运行报连接错误

确保 Ollama 服务已启动：

ollama serve

上下文超出限制

调整 budget_tokens 或使用更激进的压缩策略：

ctx = ContextManager(
    strategy=TieredCompact(keep_recent=1),  # 减少保留轮数
    budget_tokens=4096                       # 减小预算
)

模型不支持

检查 MODEL_SAMPLING_DEFAULTS 是否包含该模型，或禁用推荐采样：

client = OllamaClient(
    model="your-model",
    recommended_sampling=False  # 不使用推荐参数
)

下一步

• 阅读 用户指南 了解高级特性
• 查看 MODEL_GUIDE.md 了解支持的模型
• 运行评估：python -m tests.eval.eval_runner --scenarios your_scenario --runs 5

核心设计原则

系统架构图

graph TD
    subgraph 客户端层
        OA[OllamaClient]
        LF[LlamafileClient]
        AC[AnthropicClient]
        PR[ProxyClient]
    end

    subgraph 核心层
        WR[WorkflowRunner]
        WF[Workflow]
        CT[ToolCall]
    end

    subgraph 上下文层
        CM[ContextManager]
        TC[TieredCompact]
    end

    subgraph 后端管理层
        SM[ServerManager]
        BM[BudgetMode]
    end

    subgraph 工具执行层
        GR[Guardrails]
        RV[ResponseValidator]
        SE[StepEnforcer]
    end

    OA --> WR
    LF --> WR
    AC --> WR
    PR --> WR
    
    WR --> WF
    WR --> CM
    CM --> TC
    
    WF --> CT
    CT --> GR
    GR --> RV
    GR --> SE

项目目录结构

src/forge/           # 库源码
  clients/           # LLM 后端适配器（每个后端一个）
  core/              # Workflow、Runner、消息、步骤
  context/           # 上下文管理和压缩
  prompts/           # 提示模板和 Nudge
  guardrails/        # 安全护栏机制
tests/               # 测试套件
  unit/              # 确定性单元测试
  eval/              # 评估测试（需真实后端）
docs/                # 用户文档

资料来源：CONTRIBUTING.md:44-53

核心组件详解

1. Workflow（工作流）

Workflow 是任务定义的中心模型，包含工具注册、步骤约束和终止条件：

资料来源：src/forge/core/workflow.py

#### ToolDef 结构

classDiagram
    class ToolDef {
        spec: ToolSpec
        callable: Callable
        prerequisites: list[str | dict]
        name: str
    }
    
    class ToolSpec {
        name: str
        description: str
        parameters: BaseModel
        get_json_schema() dict
    }
    
    ToolDef --> ToolSpec : contains

ToolDef 将工具模式绑定到其实现，支持先决条件表达：

• 字符串形式："read_file" — 任何先前对 read_file 的调用都满足
• 字典形式：{"tool": "read_file", "match_arg": "path"} — 要求先前调用具有相同 path 参数值

资料来源：src/forge/core/workflow.py:45-57

2. WorkflowRunner（工作流运行器）

WorkflowRunner 是执行引擎，协调 LLM 交互、工具调用和上下文管理：

sequenceDiagram
    participant U as 用户
    participant R as WorkflowRunner
    participant C as LLMClient
    participant G as Guardrails
    participant T as 工具执行
    participant CM as ContextManager

    U->>R: run(workflow, user_input)
    loop 执行循环
        R->>C: generate(messages)
        C-->>R: LLMResponse
        R->>G: check(response)
        G-->>R: CheckResult
        alt 需要重试
            R->>C: generate(retry_nudge)
        else 工具调用
            R->>T: execute(tool_calls)
            T-->>R: results
            R->>CM: compact()
            R->>C: continue
        else 终止
            R-->>U: final_result
        end
    end

资料来源：src/forge/core/runner.py

#### Runner 执行流程

1. 验证阶段：检查 LLM 响应是否包含有效工具调用
2. 护栏检查：通过 Guardrails.check() 验证响应
3. 执行阶段：运行工具并收集结果
4. 上下文更新：压缩上下文以保持在 token 预算内
5. 循环继续：直到调用终止工具或达到最大重试次数

3. ContextManager（上下文管理器）

上下文管理器负责在多轮对话中维护和压缩 token 预算：

资料来源：src/forge/server.py:140-150

graph LR
    A[用户输入] --> B[构建消息]
    B --> C{Token 超限?}
    C -->|否| D[发送给 LLM]
    C -->|是| E[执行压缩]
    E --> D
    D --> F[收集响应]
    F --> G[工具执行]
    G --> B

#### TieredCompact 策略

# 示例配置
ctx = ContextManager(
    strategy=TieredCompact(keep_recent=2),
    budget_tokens=8192
)

资料来源：README.md:18

4. Guardrails（安全护栏）

Guardrails 是确保工作流正确执行的核心机制，由三个子组件构成：

classDiagram
    class Guardrails {
        _validator: ResponseValidator
        _enforcer: StepEnforcer
        _errors: ErrorTracker
        check(response) CheckResult
    }
    
    class ResponseValidator {
        rescue_enabled: bool
        retry_nudge_fn: Callable
    }
    
    class StepEnforcer {
        required_steps: list
        terminal_tools: frozenset
        max_premature_attempts: int
    }
    
    class ErrorTracker {
        max_retries: int
        max_tool_errors: int
    }
    
    Guardrails --> ResponseValidator
    Guardrails --> StepEnforcer
    Guardrails --> ErrorTracker

资料来源：src/forge/guardrails/guardrails.py:45-75

#### Guardrails 参数

#### CheckResult 动作

5. LLM 后端适配器

Forge 支持多种 LLM 后端，通过统一接口抽象差异：

graph TD
    subgraph 后端适配器
        OC[OllamaClient]
        LFC[LlamafileClient]
        ANC[AnthropicClient]
        PC[ProxyClient]
    end

    subgraph 公共接口
        PI[LLMClient 接口]
        GC[generate]
        GX[get_context_length]
    end

    OC --> PI
    LFC --> PI
    ANC --> PI
    PC --> PI

#### 支持的后端

资料来源：src/forge/server.py:90-110

6. ServerManager（服务器管理器）

ServerManager 负责后端进程的生命周期管理：

stateDiagram-v2
    [*] --> Idle: 创建实例
    Idle --> Running: start()
    Running --> Idle: stop()
    Running --> Running: start() (不同模型)
    
    note right of Running: 自动复用相同配置
    note right of Idle: 进程已终止

#### 缓存优化机制

ServerManager 会缓存当前运行的配置，仅在配置变更时重启服务器：

# 配置相等性检查
if (
    self._current_model == model
    and self._current_mode == mode
    and self._current_ctx == ctx_override
    and self._current_flags == flags
):
    return  # 复用现有服务器

资料来源：src/forge/server.py:40-50

#### Budget 模式解析

flowchart LR
    A[BudgetMode] --> B{MANUAL?}
    A --> C{OLLAMA?}
    A --> D{FORGE_FAST?}
    A --> E{FULL?}
    
    B -->|是| F[使用 manual_tokens]
    C -->|是| G[获取 ollama 上下文]
    D --> H[计算 fast 预算]
    E --> I[获取服务器完整上下文]

数据流图

flowchart TD
    subgraph 输入层
        U[用户输入]
        W[Workflow 定义]
        T[工具实现]
    end

    subgraph 核心引擎
        R[WorkflowRunner]
        G[Guardrails]
        M[ContextManager]
    end

    subgraph LLM 层
        C[LLMClient]
        S[ServerManager]
    end

    U --> R
    W --> R
    T --> R
    R --> C
    C --> S
    R --> G
    R --> M
    
    M -->|压缩消息| C
    G -->|验证结果| R

快速启动示例

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. Use the available tools to answer the user.",
)

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())

资料来源：README.md:7-42

扩展机制

添加新 LLM 后端

1. 在 src/forge/clients/ 目录创建新的客户端类
2. 实现统一的异步接口方法
3. 注册到客户端注册表

添加 Guardrail

1. 在 Guardrails 类中添加新的检查逻辑
2. 在 AblationConfig 中添加开关
3. 创建消融实验预设进行验证

资料来源：CONTRIBUTING.md:14-22

总结

Forge 的架构围绕三个核心目标设计：

整个系统基于异步架构设计，所有核心组件（Runner、Client、ServerManager）均支持并发执行，确保在多任务场景下的高性能表现。

概述

Forge 是一个基于 LLM 的工具调用框架，其核心组件负责管理对话工作流、执行步骤追踪、上下文管理和后端服务调度。整个系统采用异步架构设计，通过 WorkflowRunner 协调各个组件完成复杂的多轮对话任务。

核心组件按照职责划分为以下几个模块：

资料来源：src/forge/core/workflow.py

概述

Forge 的工作流系统是一个异步优先的 LLM 工具调用编排框架，核心设计目标是将复杂的 LLM 交互与工具执行解耦。工作流由 Workflow 类定义，封装了工具定义、步骤要求、终端工具等元信息，通过 WorkflowRunner 驱动执行。

工作流机制包含三个核心子系统：

1. 工具定义与绑定系统 — 定义可用工具及其参数模式
2. 步骤追踪与前置条件系统 — 确保工具按正确顺序调用
3. Guardrails 保护系统 — 验证响应、处理重试、阻止过早终止

资料来源：src/forge/core/workflow.py:1-50

核心数据模型

Workflow 类

Workflow 是工作流的核心抽象，作为 dataclass 存储所有元信息。

@dataclass
class Workflow:
    name: str
    description: str
    tools: dict[str, ToolDef]
    required_steps: list[str]
    terminal_tool: str
    system_prompt_template: str

资料来源：src/forge/core/workflow.py:60-80

ToolDef 与 ToolSpec

ToolDef 将工具模式绑定到实际可执行函数，ToolSpec 定义工具的 JSON Schema 参数规范。

@dataclass
class ToolDef:
    spec: ToolSpec
    callable: Callable[..., Any]
    prerequisites: list[str | dict[str, str]] = field(default_factory=list)

prerequisites 字段支持两种前置条件表达方式：

资料来源：src/forge/core/workflow.py:85-110

ToolCall 数据模型

ToolCall 是 LLM 返回的工具调用验证结果。

class ToolCall(BaseModel):
    tool: str
    args: dict[str, Any] = {}
    reasoning: str | None = None

资料来源：src/forge/core/workflow.py:120-130

步骤追踪机制

StepTracker

StepTracker 维护在工作流运行器上，独立于消息历史，紧跟在上下文压缩之外。

@dataclass
class StepTracker:
    required_steps: list[str]
    completed_steps: dict[str, None] = field(default_factory=dict)
    executed_tools: dict[str, list[dict[str, Any]]] = field(default_factory=dict)

资料来源：src/forge/core/steps.py:30-55

PrerequisiteCheck

前置条件检查结果用于判断工具调用是否满足依赖要求。

@dataclass
class PrerequisiteCheck:
    satisfied: bool
    missing: list[str]

资料来源：src/forge/core/steps.py:15-25

Guardrails 系统

Guardrails 是 Forge 的保护中间件，将三个独立检查器组合成统一的两阶段 API：

graph LR
    A[LLM Response] --> B[Guardrails.check]
    B --> C{ResponseValidator}
    B --> D{StepEnforcer}
    B --> E{ErrorTracker}
    C --> F[CheckResult]
    D --> F
    E --> F

CheckResult 结果类型

action: Literal["execute", "retry", "step_blocked", "fatal"]
tool_calls: list[ToolCall] | None
nudge: Nudge | None
reason: str | None

资料来源：src/forge/guardrails/guardrails.py:140-160

ResponseValidator

验证 LLM 响应是否包含有效工具调用，支持救援解析。

class ResponseValidator:
    def __init__(
        self,
        tool_names: list[str],
        rescue_enabled: bool = True,
        retry_nudge_fn: Callable[[str], str] | None = None,
    )

救援解析支持以下格式：

• OpenAI 格式：{"name": ..., "arguments": ...}
• Qwen Coder XML 格式：<function=name><parameter=key>value</parameter></function>

资料来源：src/forge/guardrails/guardrails.py:40-80

StepEnforcer

强制执行步骤顺序，防止过早调用终端工具。

class StepEnforcer:
    def __init__(
        self,
        required_steps: list[str],
        terminal_tools: frozenset[str],
        max_premature_attempts: int = 3,
    )

超过 max_premature_attempts 次过早终止尝试后返回 fatal。

资料来源：src/forge/guardrails/guardrails.py:80-110

ErrorTracker

追踪连续错误和工具执行失败。

class ErrorTracker:
    def __init__(
        self,
        max_retries: int = 3,
        max_tool_errors: int = 2,
    )

资料来源：src/forge/guardrails/guardrails.py:110-140

提示模板与 Nudge

步骤阻止提示

当模型尝试在完成必需步骤前调用终端工具时，系统注入阻止提示：

def step_blocked_nudge(terminal_tool, pending_steps, tier=1):
    tier = max(1, min(3, tier))
    steps = ", ".join(pending_steps)
    if tier == 1:
        return f"You cannot call {terminal_tool} yet. You must first complete these required steps: {steps}."

资料来源：src/forge/prompts/nudges.py:1-30

前置条件提示

当工具调用缺少前置依赖时触发：

def prerequisite_nudge(tool_name, missing_prereqs):
    prereqs = ", ".join(missing_prereqs)
    return f"You cannot call {tool_name} yet. You must first call: {prereqs}."

资料来源：src/forge/prompts/nudges.py:35-55

工作流执行流程

完整执行状态机

graph TD
    A[初始化 WorkflowRunner] --> B[构建系统提示]
    B --> C[发送初始请求到 LLM]
    C --> D[接收 LLM 响应]
    D --> E{Guardrails.check}
    E -->|execute| F[执行工具]
    E -->|retry| G[注入 nudge]
    E -->|step_blocked| G
    E -->|fatal| H[终止工作流]
    F --> I{工具为终端工具?}
    I -->|是| J[返回最终结果]
    I -->|否| K[记录步骤]
    K --> C
    G --> C

分阶段 API 用法

对于外部编排循环，Forge 提供两种使用方式：

简化 API（全部集成）

from forge.guardrails import Guardrails

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

result = guardrails.check(response)
if result.action == "execute":
    executed = [tc.tool for tc in result.tool_calls]
    done = guardrails.record(executed)

分阶段 API（细粒度控制）

from forge.guardrails import ErrorTracker, ResponseValidator, StepEnforcer

validator = ResponseValidator(tool_names=[...], rescue_enabled=True)
enforcer = StepEnforcer(required_steps=[...], terminal_tool="answer")
errors = ErrorTracker(max_retries=3, max_tool_errors=2)

# 每阶段可自定义处理逻辑

资料来源：examples/foreign_loop.py:1-80

快速开始示例

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

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

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

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")
    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?")

资料来源：README.md:1-50

架构设计要点

异步优先设计

所有客户端方法和运行器均为 async 实现，支持高并发工具调用场景：

async def run(self, workflow: Workflow, user_input: str) -> WorkflowResult:
    ...

资料来源：src/forge/core/workflow.py:1-30

上下文管理隔离

ContextManager 独立于步骤追踪，允许上下文压缩不影响步骤完成状态：

graph LR
    subgraph WorkflowRunner
        A[StepTracker] --- B[ContextManager]
    end
    B --- C[消息历史]
    A --- D[completed_steps]

工具参数模式

使用 Pydantic 模型定义工具参数，确保类型安全：

parameters: type[BaseModel]  # 必须是 Pydantic 模型

可通过 get_json_schema() 方法导出 JSON Schema：

def get_json_schema(self) -> dict[str, Any]:
    return self.parameters.model_json_schema()

资料来源：src/forge/core/workflow.py:50-65

总结

Forge 的工作流机制通过清晰的职责分离实现了可靠的 LLM 工具调用编排：

该设计支持从简单的单工具调用到复杂的多步骤工作流，同时为外部编排系统提供了灵活的中间件接口。

概述

Guardrails 系统是 Forge 框架的核心安全与流程控制中间件，设计用于在外部编排循环中确保 LLM 响应符合预期的工作流约束。该系统通过三重验证机制（响应验证、步骤强制、错误追踪）保护工具调用流程的完整性和可靠性。 资料来源：src/forge/guardrails/guardrails.py:46-52

系统架构

Guardrails 系统由四个核心组件构成，它们协同工作以实现完整的中间件功能：

graph TB
    subgraph Guardrails系统
        RV[ResponseValidator<br/>响应验证器]
        SE[StepEnforcer<br/>步骤强制器]
        ET[ErrorTracker<br/>错误追踪器]
        NR[Nudge模块<br/>提示生成]
    end
    
    LLM[LLM响应] --> RV
    RV --> SE
    SE --> ET
    NR --> SE
    
    subgraph CheckResult
        AC[action]
        TC[tool_calls]
        ND[nudge]
        RS[reason]
    end
    
    ET --> CheckResult

资料来源：src/forge/guardrails/guardrails.py:24-45

CheckResult 数据模型

check() 方法返回 CheckResult 对象，包含四个字段：

class CheckResult(BaseModel):
    action: Literal["execute", "retry", "step_blocked", "fatal"]
    tool_calls: list[ToolCall] | None = None
    nudge: Nudge | None = None
    reason: str | None = None

资料来源：src/forge/guardrails/guardrails.py:67-74

Guardrails 核心 API

构造函数参数

资料来源：src/forge/guardrails/guardrails.py:76-91

主要方法

#### check() 方法

在每次 LLM 响应后、执行任何工具前调用此方法。它依次通过三个验证组件：

sequenceDiagram
    participant App as 调用方
    participant G as Guardrails
    participant RV as ResponseValidator
    participant SE as StepEnforcer
    participant ET as ErrorTracker
    
    App->>G: check(response)
    G->>RV: validate(response)
    RV-->>G: parsed_tool_calls
    G->>SE: enforce(tool_calls)
    SE-->>G: step_result
    G->>ET: track_errors()
    ET-->>G: error_status
    G-->>App: CheckResult

资料来源：src/forge/guardrails/guardrails.py:93-126

#### record() 方法

在工具执行完成后调用，用于更新错误追踪器状态：

def record(self, executed: list[str]) -> bool:
    """Record executed tools and check if workflow is done.
    
    Returns:
        True if the terminal tool was reached (workflow complete).
    """
    self._errors.tool_succeeded()
    return self._enforcer.is_terminal_reached(executed)

资料来源：src/forge/guardrails/guardrails.py:128-140

组件详解

ResponseValidator（响应验证器）

负责解析 LLM 返回的原始响应，提取结构化工具调用，并处理救援解析场景：

• 工具调用提取：从 JSON 格式或特定文本格式中提取工具名和参数
• 救援解析：当模型输出纯文本而非工具调用时，可配置地尝试恢复
• 自定义重试提示：支持通过 retry_nudge_fn 生成动态重试消息

资料来源：src/forge/guardrails/response_validator.py

StepEnforcer（步骤强制器）

确保工作流按正确的步骤顺序执行：

graph LR
    A[search] --> B[lookup]
    B --> C[answer]
    
    style A fill:#90EE90
    style B fill:#90EE90
    style C fill:#FFD700

关键特性：

• 验证 required_steps 列表中的所有工具已被调用
• 检测提前调用 terminal_tool 的情况（max_premature_attempts 控制）
• 支持通过 is_terminal_reached() 判断工作流是否完成

资料来源：src/forge/guardrails/step_enforcer.py

ErrorTracker（错误追踪器）

维护错误状态机，防止无限重试循环：

当工具成功执行时，两个计数器都会重置。

资料来源：src/forge/guardrails/error_tracker.py

Nudge 提示系统

Nudge 模块负责生成用户友好的引导消息，帮助模型理解并修正其行为。

step_nudge() 函数

当模型试图跳过必需步骤直接调用终端工具时触发：

def step_nudge(
    terminal_tool: str,
    pending_steps: list[str],
    tier: int = 1,
) -> str:
    tier = max(1, min(3, tier))  # 限制在 1-3 范围内

资料来源：src/forge/prompts/nudges.py:18-40

prerequisite_nudge() 函数

当模型调用带有前置依赖的工具但未满足前置条件时触发：

def prerequisite_nudge(tool_name: str, missing_prereqs: list[str]) -> str:
    prereqs = ", ".join(missing_prereqs)
    return (
        f"You cannot call {tool_name} yet. "
        f"You must first call: {prereqs}. "
        "Call the prerequisite tool now."
    )

资料来源：src/forge/prompts/nudges.py:42-56

使用模式

简化 API（推荐）

使用 Guardrails 类进行一站式验证：

from forge.guardrails import Guardrails

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

def handle_response(response):
    result = guardrails.check(response)
    
    if result.action == "fatal":
        return f"FATAL: {result.reason}"
    
    if result.action in ("retry", "step_blocked"):
        # 将 nudge.content 注入对话历史
        return f"{result.action}: {result.nudge.content}"
    
    # 执行工具
    executed = [tc.tool for tc in result.tool_calls]
    done = guardrails.record(executed)
    return f"executed {executed}" + (" -- DONE" if done else "")

资料来源：examples/foreign_loop.py:19-40

粒度 API（高级）

直接访问各个验证组件，实现自定义行为：

from forge.guardrails import ErrorTracker, ResponseValidator, StepEnforcer

validator = ResponseValidator(
    tool_names=["search", "lookup", "answer"],
    rescue_enabled=True,
)
enforcer = StepEnforcer(
    required_steps=["search", "lookup"],
    terminal_tool="answer",
)
errors = ErrorTracker(max_retries=3, max_tool_errors=2)

适用于需要日志记录、指标收集或条件性救援的场景。

资料来源：examples/foreign_loop.py:52-66

与 respond 工具集成

Forge 支持通过 respond() 工具让模型直接返回文本响应：

from forge.tools import RESPOND_TOOL_NAME, respond_spec

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

def handle_response(response):
    result = guardrails.check(response)
    # ...
    for tc in result.tool_calls:
        if tc.tool == RESPOND_TOOL_NAME:
            message = tc.args.get("message", "")
            return f"MODEL SAYS: {message}"

资料来源：examples/foreign_loop.py:95-115

Guardrails 工作流状态图

stateDiagram-v2
    [*] --> 等待响应
    等待响应 --> 验证响应: LLM 返回
    验证响应 --> 响应有效: 工具调用已提取
    验证响应 --> 响应无效: 解析失败
    响应无效 --> 检查重试次数: 递增 retry
    检查重试次数 --> 等待响应: retry_nudge
    检查重试次数 --> 致命错误: 达到上限
    响应有效 --> 检查步骤顺序
    检查步骤顺序 --> 步骤通过: 验证通过
    检查步骤顺序 --> 步骤受阻: 缺少步骤
    步骤受阻 --> 重置提前尝试: step_nudge
    步骤通过 --> 执行工具
    执行工具 --> 记录执行: record()
    记录执行 --> 工作流完成?: terminal reached
    记录执行 --> 工具错误?: 工具执行失败
    工作流完成? --> [*]: 是
    工作流完成? --> 等待响应: 否
    工具错误? --> 等待响应: 重试
    工具错误? --> [*]: 达到上限

配置与消融研究

Guardrails 的各个组件可通过 AblationConfig 独立切换，用于评估每个防护措施的性能影响：

# tests/eval/ablation.py 中的配置示例
@dataclass
class AblationConfig:
    enable_rescue: bool = True
    enable_step_enforcement: bool = True
    enable_error_tracker: bool = True

在 CONTRIBUTING.md 中提到，每个 guardrail 可以独立消融：

Guardrails live in the runner (src/forge/core/runner.py) and nudge templates (src/forge/prompts/nudges.py). Each guardrail can be independently toggled via ablation presets in tests/eval/ablation.py.

资料来源：CONTRIBUTING.md:3-8

总结

Guardrails 系统通过模块化的验证、强制和追踪机制，为 Forge 工作流提供了可靠的安全防护：

概述

上下文管理（Context Management）是 forge 框架中负责管理对话历史和令牌预算的核心系统。在与大语言模型（LLM）进行多轮交互时，上下文窗口的大小直接影响能保留多少历史信息和工具调用记录。forge 的上下文管理系统通过预算模式（Budget Mode）和压缩策略（Compaction Strategy）的组合，确保在有限的令牌预算内最大化有效信息的保留。

该系统位于 src/forge/context/ 模块下，包含三个核心子模块：

资料来源：CONTRIBUTING.md

概述

SlotWorker 是 forge 框架中用于管理 llama-server / llamafile 后端并发槽位 的核心机制。它通过 ServerManager 类实现，支持在同一进程中启动具有多个独立 KV Cache 槽位的推理服务器，从而支持多智能体（multi-agent）架构的并发执行。

槽位调度的核心目的是：在单个推理服务器实例上并行运行多个独立的 agent 工作流，每个槽位拥有独立的上下文状态，彼此互不干扰。

资料来源：src/forge/server.py:1-50

架构设计

槽位模型

forge 的槽位调度采用 硬分区 + 可选统一缓存 两种模式：

资料来源：src/forge/server.py:35-42

组件关系图

graph TD
    A[WorkflowRunner] --> B[ContextManager]
    A --> C[ServerManager]
    C --> D[llama-server / llamafile]
    D --> E[Slot 0: KV Cache]
    D --> F[Slot 1: KV Cache]
    D --> G[Slot N: KV Cache]
    E --> H[独立上下文]
    F --> H
    G --> H

ServerManager 槽位管理

初始化参数

ServerManager 类负责启动和管理带有槽位配置的推理服务器。其构造函数接受以下参数：

资料来源：src/forge/server.py:120-133

槽位状态追踪

ServerManager 内部维护以下状态字段用于槽位调度决策：

_current_model: str | None       # 当前加载的模型标识
_current_mode: str | None         # 运行模式 (native/prompt/reforged)
_current_ctx: int | None          # 上下文长度配置
_current_flags: tuple[str, ...]   # 额外 CLI 参数
_current_cache_type_k: str | None # KV Cache Key 量化类型
_current_cache_type_v: str | None # KV Cache Value 量化类型
_current_n_slots: int | None      # 并发槽位数量
_current_kv_unified: bool         # 是否启用统一 KV Cache

资料来源：src/forge/server.py:135-146

启动方法签名

async def start(
    self,
    model: str,
    gguf_path: str | Path,
    mode: str = "native",
    extra_flags: list[str] | None = None,
    ctx_override: int | None = None,
    cache_type_k: str | None = None,  # e.g. "q8_0", "q4_0"
    cache_type_v: str | None = None,  # e.g. "q8_0", "q4_0"
    n_slots: int | None = None,       # 并发槽位数量
    kv_unified: bool = False,         # 统一 KV Cache 模式
) -> None:

资料来源：src/forge/server.py:60-77

槽位配置与缓存量化

KV Cache 量化类型

forge 支持对 Key 和 Value 缓存分别进行量化，以节省显存：

槽位复用机制

ServerManager 实现了智能复用逻辑：当新请求的配置与当前运行配置完全一致时，跳过重启直接复用现有槽位：

if (
    self._current_model == model
    and self._current_mode == mode
    and self._current_ctx == ctx_override
    and self._current_flags == flags
    and self._current_cache_type_k == cache_type_k
    and self._current_cache_type_v == cache_type_v
    and self._current_n_slots == n_slots
    and self._current_kv_unified == kv_unified
):
    return  # 复用现有槽位

资料来源：src/forge/server.py:69-81

工作流程集成

槽位与 WorkflowRunner 的协作

WorkflowRunner 在执行工作流时通过 ContextManager 管理上下文，并通过 ServerManager 与后端交互：

sequenceDiagram
    participant User
    participant WorkflowRunner
    participant ContextManager
    participant ServerManager
    participant LlamaServer
    
    User->>WorkflowRunner: run(workflow, input)
    WorkflowRunner->>ContextManager: 请求上下文预算
    ContextManager->>ServerManager: resolve_budget(mode)
    ServerManager->>LlamaServer: 查询 /props 获取 n_ctx
    LlamaServer-->>ServerManager: 返回上下文长度
    ServerManager-->>ContextManager: 返回预算
    WorkflowRunner->>LlamaServer: 发送推理请求（槽位分配）

步骤追踪与槽位隔离

每个槽位在执行过程中会独立维护步骤状态：

@dataclass
class StepTracker:
    """追踪必需步骤完成情况和工具执行记录"""
    required_steps: list[str]
    completed_steps: dict[str, None] = field(default_factory=dict)
    executed_tools: dict[str, list[dict[str, Any]]] = field(default_factory=dict)

资料来源：src/forge/core/steps.py:22-30

槽位调度模式

模式对比

资料来源：src/forge/server.py:38-42

配置示例

from forge import ServerManager, BudgetMode

# 创建支持 4 个并发槽位的服务器管理器
server = ServerManager(backend="llamaserver", port=8080)

# 启动服务器，配置统一 KV Cache
await server.start(
    model="qwen3:14b-q4_K_M",
    gguf_path="/models/qwen3-14b-q4_k_m.gguf",
    mode="reforged",
    n_slots=4,
    kv_unified=True,
    cache_type_k="q8_0",
    cache_type_v="q4_0",
)

错误处理与超时

错误追踪机制

Guardrails 模块提供槽位级别的错误追踪：

@dataclass
class ErrorTracker:
    """追踪重试次数和工具错误"""
    max_retries: int
    max_tool_errors: int
    _consecutive_retries: int = 0
    _tool_errors: int = 0

资料来源：src/forge/guardrails/guardrails.py:45-52

超时配置

批量评估时，每个场景有 300 秒墙钟超时：

批量评估时，每个场景有 300s 墙钟超时；超时时运行记录为 completeness=False, error_type='Timeout'，批次继续执行。

资料来源：CHANGELOG.md:30-32

最佳实践

槽位数量选择

缓存量化建议

限制与已知问题

llama.cpp reasoning budget 挂起问题

已知问题：2026 年 4 月 10 日之后的构建版本中，Gemma 4、Qwen 3.5 和 Ministral Reasoning 模型使用无界 reasoning budget sampler 会导致静默挂起。

临时解决方案：使用 --reasoning-budget 0 参数禁用。

资料来源：CHANGELOG.md:24-28

后端兼容性

总结

SlotWorker 槽位调度是 forge 框架支持多智能体并发执行的核心基础设施。通过 ServerManager 类，开发者可以：

1. 配置并发槽位数量 (n_slots) 以支持多 agent 并行
2. 选择 KV Cache 模式 (kv_unified) 平衡隔离性与资源利用
3. 优化显存使用 通过 Key/Value 缓存量化
4. 实现智能复用 避免不必要的服务器重启

该机制使 forge 能够在单个推理服务器实例上高效运行多个独立工作流，特别适合需要并行评估或异步执行多个 agent 任务的场景。

概述

Forge 的内置工具系统是一套用于定义、注册和执行 LLM 工具调用的核心机制。该系统通过 ToolSpec、ToolDef 和 ToolCall 三个核心类实现工具的schema定义与实际执行逻辑的绑定，并为工作流引擎提供验证、执行和依赖管理能力。

工具系统在整个架构中扮演关键角色：

• 工具发现：工作流通过 ToolDef 字典声明可用工具
• LLM 接口：工具规格自动转换为 JSON Schema，供 LLM 理解调用格式
• 执行验证：GuardRails 组件验证工具调用合法性
• 依赖管理：通过前置条件（Prerequisites）表达工具间依赖关系

资料来源：src/forge/core/workflow.py:1-50

核心数据模型

ToolSpec

ToolSpec 是工具的静态声明，使用 Pydantic 模型定义参数schema：

@dataclass
class ToolSpec:
    """Tool specification with name, description, and parameters."""
    name: str
    description: str
    parameters: type[BaseModel]

parameters 字段接收一个 Pydantic BaseModel 子类，Forge 自动从中提取 JSON Schema：

def get_json_schema(self) -> dict[str, Any]:
    """Return JSON Schema dict for this tool's parameters."""
    return self.parameters.model_json_schema()

资料来源：src/forge/core/workflow.py:100-140

ToolDef

ToolDef 将工具规格与实现绑定，并声明执行依赖：

@dataclass
class ToolDef:
    """Binds a tool schema to its implementation."""
    spec: ToolSpec
    callable: Callable[..., Any]
    prerequisites: list[str | dict[str, str]] = field(default_factory=list)

prerequisites 支持两种表达方式：

``python prerequisites=["read_file"] ` 表示调用此工具前必须曾调用过 read_file`

1. 字符串形式（名称匹配）：

``python prerequisites=[{"tool": "read_file", "match_arg": "path"}] ` 表示调用此工具前必须调用过 read_file 且 path` 参数值相同

1. 字典形式（参数匹配）：

资料来源：src/forge/core/workflow.py:143-175

ToolCall

ToolCall 是经 LLM 返回并验证后的工具调用对象：

class ToolCall(BaseModel):
    """Validated tool invocation returned by an LLMClient."""
    tool: str

由 ResponseValidator 验证通过后生成，传递给执行层。

资料来源：src/forge/core/workflow.py:177-179

工具注册与使用

工作流中的工具声明

在创建工作流时，通过 tools 字典注册所有可用工具：

from forge import Workflow, ToolDef, ToolSpec

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. Use the available tools to answer the user.",
)

工作流内部从 tools 字典派生出：

• 工具规格列表：供 LLM 理解可调用工具
• 可调用对象映射：执行时的函数查找表

资料来源：README.md:1-40

参数模型定义

使用 Pydantic 定义工具参数，每个参数可带描述供 LLM 理解：

from pydantic import BaseModel, Field

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

Pydantic 自动处理类型验证和类型转换。

资料来源：README.md:5-15

前置依赖机制

依赖声明语法

工具的前置依赖通过 prerequisites 字段声明，支持细粒度控制：

def answer(question: str) -> str:
    """Answer a question."""
    return "The answer is 42."

workflow_tools = {
    "search": ToolDef(
        spec=ToolSpec(name="search", description="Search the web", parameters=SearchParams),
        callable=search,
    ),
    "lookup": ToolDef(
        spec=ToolSpec(name="lookup", description="Look up details", parameters=LookupParams),
        callable=lookup,
    ),
    "answer": ToolDef(
        spec=ToolSpec(name="answer", description="Answer the question", parameters=AnswerParams),
        callable=answer,
        prerequisites=["search", "lookup"],  # 简单依赖
    ),
}

依赖执行流程

graph TD
    A[LLM 返回工具调用] --> B[StepEnforcer 检查依赖]
    B --> C{依赖满足?}
    C -->|是| D[执行工具]
    C -->|否| E[返回 Nudge 提示]
    D --> F[记录执行结果]
    F --> G{所有必需步骤完成?}
    G -->|是| H[允许终止工具]
    G -->|否| A

StepEnforcer 组件负责验证依赖是否满足：

enforcer = StepEnforcer(
    required_steps=["search", "lookup"],
    terminal_tool="answer",
)

资料来源：examples/foreign_loop.py:1-80

工具验证流程

ResponseValidator

ResponseValidator 负责验证 LLM 返回的工具调用：

1. 工具名称验证：检查是否在允许列表中
2. 参数验证：通过 Pydantic 模型验证参数合法性
3. 未知工具检测：识别并提示未知工具名

validator = ResponseValidator(
    tool_names=["search", "lookup", "answer"],
    rescue_enabled=True,
)

资料来源：src/forge/guardrails/response_validator.py:1-60

验证结果处理

验证结果通过 ValidationResult 返回：

if result.action == "fatal":
    return f"FATAL: {result.reason}"

if result.action in ("retry", "step_blocked"):
    # 注入 nudge 到对话历史
    return f"{result.action}: {result.nudge.content[:80]}..."

# result.action == "execute"
# 执行工具并通知 forge
tool_calls = result.tool_calls
executed = [tc.tool for tc in tool_calls]
done = guardrails.record(executed)

资料来源：examples/foreign_loop.py:30-50

GuardRails 集成

GuardRails 是工具系统的顶层编排组件，整合验证、依赖检查和错误追踪：

@dataclass
class GuardRails:
    """Check LLM responses against tool-calling guardrails."""

    def __init__(
        self,
        tool_names: list[str],
        terminal_tool: str | frozenset[str],
        required_steps: list[str] | None = None,
        max_retries: int = 3,
        max_tool_errors: int = 2,
        rescue_enabled: bool = True,
        max_premature_attempts: int = 3,
        retry_nudge: Callable[[str], str] | None = None,
    ) -> None:

资料来源：src/forge/guardrails/guardrails.py:1-50

架构图

工具系统完整架构

graph TB
    subgraph "工具定义层"
        PS[ToolSpec<br/>工具规格]
        PD[ToolDef<br/>工具定义]
        PC[Callable<br/>可调用对象]
        PR[Prerequisites<br/>前置依赖]
    end
    
    subgraph "工作流层"
        WF[Workflow<br/>工作流]
        SE[StepEnforcer<br/>步骤执行器]
        CM[ContextManager<br/>上下文管理]
    end
    
    subgraph "验证层"
        RV[ResponseValidator<br/>响应验证器]
        ET[ErrorTracker<br/>错误追踪]
        GR[GuardRails<br/>防护栏]
    end
    
    subgraph "LLM 层"
        LL[LLM Client<br/>LLM 客户端]
        TC[ToolCall<br/>工具调用]
    end
    
    PS --> PD
    PC --> PD
    PR --> PD
    PD --> WF
    WF --> SE
    WF --> CM
    LL --> TC
    TC --> RV
    RV --> GR
    SE --> GR
    ET --> GR

最佳实践

工具命名规范

• 使用小写字母和下划线：get_weather, read_file
• 动词开头描述操作：search, lookup, answer
• 名词描述数据：user_info, document_content

参数设计

class SearchParams(BaseModel):
    query: str = Field(description="Search query string")
    limit: int = Field(default=10, description="Maximum results to return")

依赖声明

对于多步骤工作流，清晰声明依赖关系：

ToolDef(
    spec=...,
    callable=answer,
    prerequisites=[
        "search",                           # 必须执行过 search
        {"tool": "lookup", "match_arg": "id"}  # 必须用相同 id 执行过 lookup
    ],
)

总结

Forge 的内置工具系统通过类型安全的 Pydantic 模型、清晰的前置依赖声明和完善的验证机制，为构建可靠的 LLM 工具调用工作流提供了坚实基础。系统设计强调：

1. 类型安全：参数schema与执行逻辑强绑定
2. 可验证性：每个工具调用都经过多层验证
3. 可追踪性：完整记录工具执行历史和依赖满足状态
4. 可扩展性：通过 ToolDef 轻松注册新工具

Pitfall Log / 踩坑日志

项目：antoinezambelli/forge

摘要：发现 15 个潜在踩坑项，其中 0 个为 high/blocking；最高优先级：安装坑 - 来源证据：Client sampling params: thread top_p/top_k/min_p/repeat_penalty through request body。

1. 安装坑 · 来源证据：Client sampling params: thread top_p/top_k/min_p/repeat_penalty through request body

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：Client sampling params: thread top_p/top_k/min_p/repeat_penalty through request body
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_148dff87195e42549d0ffb88b99e9cbf | https://github.com/antoinezambelli/forge/issues/58 | 来源类型 github_issue 暴露的待验证使用条件。

2. 安装坑 · 来源证据：Investigate: integration paths with Hermes Agent

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：Investigate: integration paths with Hermes Agent
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_e3cbd2d1c9a84a1887887bf24b036865 | https://github.com/antoinezambelli/forge/issues/51 | 来源类型 github_issue 暴露的待验证使用条件。

3. 安装坑 · 来源证据：Per-model recommended sampling defaults (map keyed by HF model cards)

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：Per-model recommended sampling defaults (map keyed by HF model cards)
• 对用户的影响：可能阻塞安装或首次运行。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_057ca2af912e4a608259ffb2a3654d4f | https://github.com/antoinezambelli/forge/issues/59 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

4. 安装坑 · 来源证据：Rescue-parse ChatGPT-style XML tool calls

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：Rescue-parse ChatGPT-style XML tool calls
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_471c674c8d73451da75d6b8c9349aabf | https://github.com/antoinezambelli/forge/issues/55 | 来源类型 github_issue 暴露的待验证使用条件。

5. 配置坑 · 来源证据：Proxy external mode hardcodes native FC — no prompt-injection fallback

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个配置相关的待验证问题：Proxy external mode hardcodes native FC — no prompt-injection fallback
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_f3a85ec8447a4838b3bc4c846cd9e7a0 | https://github.com/antoinezambelli/forge/issues/53 | 来源类型 github_issue 暴露的待验证使用条件。

6. 能力坑 · 能力判断依赖假设

• 严重度：medium
• 证据强度：source_linked
• 发现：README/documentation is current enough for a first validation pass.
• 对用户的影响：假设不成立时，用户拿不到承诺的能力。
• 建议检查：将假设转成下游验证清单。
• 防护动作：假设必须转成验证项；没有验证结果前不能写成事实。
• 证据：capability.assumptions | hn_item:48192383 | https://news.ycombinator.com/item?id=48192383 | README/documentation is current enough for a first validation pass.

7. 维护坑 · 维护活跃度未知

• 严重度：medium
• 证据强度：source_linked
• 发现：未记录 last_activity_observed。
• 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
• 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
• 防护动作：维护活跃度未知时，推荐强度不能标为高信任。
• 证据：evidence.maintainer_signals | hn_item:48192383 | https://news.ycombinator.com/item?id=48192383 | last_activity_observed missing

8. 安全/权限坑 · 下游验证发现风险项

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：下游已经要求复核，不能在页面中弱化。
• 建议检查：进入安全/权限治理复核队列。
• 防护动作：下游风险存在时必须保持 review/recommendation 降级。
• 证据：downstream_validation.risk_items | hn_item:48192383 | https://news.ycombinator.com/item?id=48192383 | no_demo; severity=medium

9. 安全/权限坑 · 存在评分风险

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：风险会影响是否适合普通用户安装。
• 建议检查：把风险写入边界卡，并确认是否需要人工复核。
• 防护动作：评分风险必须进入边界卡，不能只作为内部分数。
• 证据：risks.scoring_risks | hn_item:48192383 | https://news.ycombinator.com/item?id=48192383 | no_demo; severity=medium

10. 安全/权限坑 · 来源证据：Hardware detection: AMD unified-memory rigs fall through to 4K Ollama budget

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Hardware detection: AMD unified-memory rigs fall through to 4K Ollama budget
• 对用户的影响：可能影响授权、密钥配置或安全边界。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_4ad226a6d1fa4a5f89fa7702bec11188 | https://github.com/antoinezambelli/forge/issues/61 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

11. 安全/权限坑 · 来源证据：Sub-agent support: dynamic slot splitting

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Sub-agent support: dynamic slot splitting
• 对用户的影响：可能影响授权、密钥配置或安全边界。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_5b35873cf63c4647bca8a0611d441189 | https://github.com/antoinezambelli/forge/issues/28 | 来源类型 github_issue 暴露的待验证使用条件。

12. 安全/权限坑 · 来源证据：Sub-agent support: slot pool

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Sub-agent support: slot pool
• 对用户的影响：可能影响授权、密钥配置或安全边界。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_070d9a3d20d24123b62d7d76ee16078a | https://github.com/antoinezambelli/forge/issues/29 | 来源类型 github_issue 暴露的待验证使用条件。

13. 安全/权限坑 · 来源证据：llama.cpp reasoning budget sampler causes silent hangs after April 10 builds

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：llama.cpp reasoning budget sampler causes silent hangs after April 10 builds
• 对用户的影响：可能阻塞安装或首次运行。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_673be4a583984219bab90cbadff631fe | https://github.com/antoinezambelli/forge/issues/54 | 来源类型 github_issue 暴露的待验证使用条件。

14. 维护坑 · issue/PR 响应质量未知

• 严重度：low
• 证据强度：source_linked
• 发现：issue_or_pr_quality=unknown。
• 对用户的影响：用户无法判断遇到问题后是否有人维护。
• 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
• 防护动作：issue/PR 响应未知时，必须提示维护风险。
• 证据：evidence.maintainer_signals | hn_item:48192383 | https://news.ycombinator.com/item?id=48192383 | issue_or_pr_quality=unknown

15. 维护坑 · 发布节奏不明确

• 严重度：low
• 证据强度：source_linked
• 发现：release_recency=unknown。
• 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
• 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
• 防护动作：发布节奏未知或过期时，安装说明必须标注可能漂移。
• 证据：evidence.maintainer_signals | hn_item:48192383 | https://news.ycombinator.com/item?id=48192383 | release_recency=unknown