# https://github.com/antoinezambelli/forge 项目说明书
生成时间：2026-05-19 20:04:23 UTC
## 目录
- [Forge 简介](#page-introduction)
- [安装与配置](#page-installation)
- [快速开始教程](#page-quickstart)
- [架构概述](#page-architecture)
- [核心组件详解](#page-core-components)
- [工作流内部机制](#page-workflow-internals)
- [Guardrails 系统](#page-guardrails)
- [上下文管理](#page-context-management)
- [SlotWorker 槽位调度](#page-slot-worker)
- [内置工具系统](#page-tools)
## Forge 简介
### 相关页面
相关主题：[安装与配置](#page-installation), [架构概述](#page-architecture)
相关源码文件
以下源码文件用于生成本页说明：
- [README.md](https://github.com/antoinezambelli/forge/blob/main/README.md)
- [CONTRIBUTING.md](https://github.com/antoinezambelli/forge/blob/main/CONTRIBUTING.md)
- [src/forge/clients/sampling_defaults.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/clients/sampling_defaults.py)
- [src/forge/server.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
- [src/forge/guardrails/guardrails.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py)
- [src/forge/core/workflow.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
- [src/forge/proxy/__main__.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/proxy/__main__.py)
- [examples/foreign_loop.py](https://github.com/antoinezambelli/forge/blob/main/examples/foreign_loop.py)
- [CHANGELOG.md](https://github.com/antoinezambelli/forge/blob/main/CHANGELOG.md)
# Forge 简介
## 概述
Forge 是一个基于 LLM（大型语言模型）的工具调用框架，专注于构建可靠、可预测的多步骤 AI 工作流程。该项目由 Antoine Zambelli 开发，旨在解决 LLM 在工具调用场景中的常见问题，如响应格式不稳定、上下文长度失控、错误恢复困难等。
Forge 的核心设计理念是将 LLM 响应校验、上下文压缩、错误重试和步骤强制等机制封装为可配置的保护栏（Guardrails），使开发者能够专注于业务逻辑而非基础设施细节。资料来源：[README.md]()
## 核心概念
### Workflow（工作流）
Workflow 是 Forge 中的核心抽象，定义了一个完整的多步骤任务。它包含以下关键组件：
| 属性 | 类型 | 说明 |
|------|------|------|
| `name` | str | 工作流名称 |
| `description` | str | 任务描述 |
| `tools` | dict[str, ToolDef] | 工具定义字典 |
| `required_steps` | list[str] | 必须执行的关键步骤 |
| `terminal_tool` | str | 终止工具（执行后工作流结束） |
| `system_prompt_template` | str | 系统提示词模板 |
资料来源：[src/forge/core/workflow.py:1-50]()
### ToolDef 和 ToolSpec
ToolDef 将工具的模式定义与实际实现绑定在一起：
```python
@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 通过统一的客户端接口支持多个后端：
| 后端 | 客户端类 | 说明 |
|------|----------|------|
| Ollama | OllamaClient | 本地模型支持，推荐采样参数 |
| Llamafile | LlamafileClient | 单文件可执行模型 |
| Anthropic | AnthropicClient | Claude 系列模型 |
| Proxy | ProxyServer | OpenAI 兼容代理 |
资料来源：[README.md]()
资料来源：[CONTRIBUTING.md]()
资料来源：[src/forge/proxy/__main__.py:1-50]()
## 架构设计
### 系统组件图
```mermaid
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
```
### 工作流程
```mermaid
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]()
### 采样策略四象限
| strict | 模型在映射表中 | 行为 |
|--------|---------------|------|
| True | 是 | 返回字典副本 |
| True | 否 | 抛出 UnsupportedModelError |
| False | 是 | 单次 INFO 日志；返回 {} |
| False | 否 | 返回 {}（静默） |
资料来源：[src/forge/clients/sampling_defaults.py:40-60]()
### recommended_sampling 参数
```python
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 负责管理对话历史的长度，防止超出模型的上下文窗口。
```python
from forge import ContextManager, TieredCompact
ctx = ContextManager(
strategy=TieredCompact(keep_recent=2),
budget_tokens=8192
)
```
### BudgetMode 预算模式
| 模式 | 说明 |
|------|------|
| FORGE_FAST | 快速响应预算 |
| FORGE_BALANCED | 平衡模式 |
| FORGE_DEEP | 深度推理预算 |
| MANUAL | 手动指定 token 数 |
资料来源：[src/forge/server.py:80-100]()
### KV Cache 量化
Forge 支持 KV 缓存量化以减少显存占用：
| 量化类型 | 显存节省 | 说明 |
|----------|----------|------|
| q8_0 | ~50% | 高质量量化 |
| q4_0 | ~75% | 更高压缩率 |
资料来源：[src/forge/server.py:30-35]()
### KV Unified 模式
当 `kv_unified=True` 时，所有槽位共享单一 KV 缓存池，每个槽位可使用完整上下文长度。
## Guardrails 系统
Guardrails 是 Forge 的核心保护机制，包含三个子组件：
### 组件架构
```mermaid
graph LR
subgraph Guardrails
V[ResponseValidator]
E[StepEnforcer]
T[ErrorTracker]
end
```
### ResponseValidator
验证 LLM 响应的格式和内容，支持 XML 救援解析（针对 Qwen Coder 等模型）。
### StepEnforcer
强制执行必需的步骤顺序：
```python
Guardrails(
tool_names=["search", "lookup", "answer"],
required_steps=["search", "lookup"],
terminal_tool="answer"
)
```
### ErrorTracker
追踪错误并控制重试次数：
| 参数 | 默认值 | 说明 |
|------|--------|------|
| max_retries | 3 | 最大重试次数 |
| max_tool_errors | 2 | 最大工具错误数 |
| max_premature_attempts | 3 | 提前终止尝试次数 |
资料来源：[src/forge/guardrails/guardrails.py:1-60]()
## 快速开始
### 安装
```bash
git clone https://github.com/antoinezambelli/forge.git
cd forge
python -m venv .venv
pip install -e ".[dev]"
```
### 基本使用示例
```python
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 后端：
```python
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]()
## 测试
### 单元测试
```bash
# 完整套件（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
```
### 集成测试
集成测试需要运行中的后端，跳过方式：
```bash
python -m pytest tests/ -m "not integration"
```
资料来源：[CONTRIBUTING.md]()
## 相关资源
| 资源 | 链接 |
|------|------|
| GitHub 仓库 | https://github.com/antoinezambelli/forge |
| 用户指南 | docs/USER_GUIDE.md |
| 后端设置 | docs/BACKEND_SETUP.md |
| 模型指南 | docs/MODEL_GUIDE.md |
| 评估仪表板 | docs/results/dashboard.html |
| 架构决策记录 | docs/decisions/ |
---
## 安装与配置
### 相关页面
相关主题：[Forge 简介](#page-introduction)
相关源码文件
以下源码文件用于生成本页说明：
- [README.md](https://github.com/antoinezambelli/forge/blob/main/README.md)
- [CONTRIBUTING.md](https://github.com/antoinezambelli/forge/blob/main/CONTRIBUTING.md)
- [src/forge/server.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
- [src/forge/clients/sampling_defaults.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/clients/sampling_defaults.py)
- [src/forge/proxy/__main__.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/proxy/__main__.py)
- [docs/BACKEND_SETUP.md](https://github.com/antoinezambelli/forge/blob/main/docs/BACKEND_SETUP.md)
# 安装与配置
## 项目概述
Forge 是一个基于 Python 的 LLM 工作流框架，支持多种后端（Ollama、llamafile、llamaserver），提供工作流管理、上下文管理和工具调用等功能。安装与配置涉及环境准备、后端选择、客户端初始化以及可选的代理服务器部署。
资料来源：[README.md:1-45]()
## 环境要求
### 系统要求
| 组件 | 要求 |
|------|------|
| Python | 3.12+ |
| 操作系统 | Linux/macOS/Windows |
| 依赖管理 | pip |
### Python 版本
Forge 使用现代 Python 语法，包括类型联合操作符（`|`）等特性，明确要求 Python 3.12 或更高版本。
资料来源：[CONTRIBUTING.md:1-15]()
## 安装步骤
### 标准安装
```bash
git clone https://github.com/antoinezambelli/forge.git
cd forge
python -m venv .venv
pip install -e ".[dev]"
```
`.[dev]` 标志安装包含开发依赖的完整包，包括测试框架 pytest。
资料来源：[CONTRIBUTING.md:1-12]()
### 可选依赖
| 额外功能 | 安装方式 |
|----------|----------|
| 开发依赖 | `pip install -e ".[dev]"` |
| 仅运行时 | `pip install -e .` |
## 后端配置
Forge 支持三种 LLM 后端，后端配置通过 `ServerManager` 类统一管理。
### 支持的后端类型
| 后端 | 标识符 | 特点 |
|------|--------|------|
| Ollama | `"ollama"` | 独立服务，使用 `model` 参数 |
| llamafile | `"llamafile"` | 单文件可执行，使用 `gguf_path` |
| llamaserver | `"llamaserver"` | llama.cpp 服务器，使用 `gguf_path` |
资料来源：[src/forge/server.py:1-50]()
### 后端初始化参数
`ServerManager` 类的构造函数接受以下参数：
```python
def __init__(
self,
backend: str, # 后端类型
port: int = 8080, # 服务端口
models_dir: str | Path | None = None, # GGUF 文件目录
) -> None:
```
资料来源：[src/forge/server.py:120-140]()
### 服务启动配置
`start()` 方法支持丰富的配置选项：
| 参数 | 类型 | 说明 |
|------|------|------|
| `model` | `str` | Ollama 模型名称 |
| `gguf_path` | `str \| Path` | GGUF 模型文件路径 |
| `mode` | `str` | 模式（默认 `"native"`） |
| `extra_flags` | `list[str]` | 额外 CLI 参数 |
| `ctx_override` | `int` | 上下文长度覆盖 |
| `cache_type_k` | `str` | KV 缓存量化类型（键） |
| `cache_type_v` | `str` | KV 缓存量化类型（值） |
| `n_slots` | `int` | 并发槽位数量 |
| `kv_unified` | `bool` | 是否使用统一 KV 缓存 |
```python
# 启动示例
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]()
### 上下文长度解析
| 后端 | 上下文获取方式 |
|------|----------------|
| llamaserver/llamafile | 查询 `/props` 端点获取 `n_ctx` |
| ollama | 使用 `ollama stop` 清理 VRAM |
资料来源：[src/forge/server.py:200-220]()
## 客户端配置
### OllamaClient
```python
from forge.clients import OllamaClient
client = OllamaClient(
model="ministral-3:8b-instruct-2512-q4_K_M",
recommended_sampling=True # 使用推荐采样参数
)
```
资料来源：[README.md:20-35]()
### LlamafileClient
```python
from forge.clients import LlamafileClient
client = LlamafileClient(
gguf_path="/path/to/model.gguf",
recommended_sampling=True
)
```
### 推荐采样参数
`forge.clients.sampling_defaults` 模块提供经过验证的模型推荐采样参数：
| 模型系列 | 支持的模型 |
|----------|------------|
| Qwen | Qwen3, Qwen3.5, Qwen3.6, Qwen3-Coder |
| Gemma | Gemma 4 |
| Mistral | Mistral Small 3.2, Mistral Nemo |
| Devstral | Devstral Small 2 |
| Ministral | Ministral 3 Instruct + Reasoning |
| Granite | Granite 4.0 (h-micro, h-tiny) |
每个配置行包含指向 HuggingFace 模型卡的 URL，参数值经过逐一验证。
资料来源：[src/forge/clients/sampling_defaults.py:1-80]()
### 采样参数策略
| 模式 | 已知模型 | 未知模型 |
|------|----------|----------|
| `strict=True` | 返回推荐参数 | 抛出 `UnsupportedModelError` |
| `strict=False` | 单次 INFO 日志，返回 `{}` | 返回 `{}`（静默） |
```python
# 严格模式示例
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` 字典参数，字段逐一合并到客户端实例级采样：
```python
response = await client.send(
messages,
sampling={"temperature": 0.8, "top_p": 0.9}
)
```
资料来源：[CHANGELOG.md:1-50]()
## 代理服务器配置
Forge 提供 `ProxyServer` 用于转发 OpenAI 兼容请求到后端。
### CLI 参数
```bash
python -m forge.proxy --backend ollama --model qwen3:8b
```
| 参数 | 默认值 | 说明 |
|------|--------|------|
| `--backend` | 必需 | 后端类型 |
| `--model` | 必需（Ollama） | 模型名称 |
| `--gguf` | 必需（非Ollama） | GGUF 文件路径 |
| `--backend-url` | 必需 | 后端服务器 URL |
| `--backend-port` | 8080 | 后端端口 |
| `--host` | 127.0.0.1 | 代理监听地址 |
| `--port` | 8081 | 代理监听端口 |
| `--budget-mode` | - | 预算模式 |
| `--budget-tokens` | - | 手动 token 预算 |
| `--extra-flags` | - | 额外后端 CLI 参数 |
| `--serialize` | None | 强制请求序列化 |
| `--max-retries` | 3 | 单请求最大重试次数 |
| `--no-rescue` | False | 禁用救援解析 |
| `--verbose` | False | 详细日志 |
资料来源：[src/forge/proxy/__main__.py:1-80]()
### 代理采样穿透
代理将 OpenAI 兼容的采样参数字段透传到后端：
| 透传字段 |
|----------|
| `temperature` |
| `top_p` |
| `top_k` |
| `min_p` |
| `repeat_penalty` |
| `presence_penalty` |
| `seed` |
资料来源：[CHANGELOG.md:1-50]()
## 工作流配置
### Workflow 组件
```python
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="你是一个有帮助的助手。",
)
```
### 上下文管理器配置
| 策略 | 说明 |
|------|------|
| `TieredCompact` | 分层压缩策略 |
| `budget_tokens` | Token 预算上限 |
```python
ctx = ContextManager(
strategy=TieredCompact(keep_recent=2),
budget_tokens=8192
)
```
资料来源：[README.md:1-50]()
## 完整运行示例
```python
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]()
## 常见问题
### 后端选择
| 场景 | 推荐后端 |
|------|----------|
| 快速原型/测试 | Ollama |
| 生产部署 | llamaserver |
| 单文件分发 | llamafile |
### 版本信息
通过 `importlib.metadata` 暴露版本：
```python
from importlib.metadata import version
print(version("forge"))
```
资料来源：[CHANGELOG.md:1-50]()
---
## 快速开始教程
### 相关页面
相关主题：[工作流内部机制](#page-workflow-internals), [Guardrails 系统](#page-guardrails)
相关源码文件
以下源码文件用于生成本页说明：
- [README.md](https://github.com/antoinezambelli/forge/blob/main/README.md)
- [src/forge/core/workflow.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
- [src/forge/core/runner.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/runner.py)
- [src/forge/guardrails/guardrails.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py)
- [src/forge/clients/llamafile.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/clients/llamafile.py)
- [examples/foreign_loop.py](https://github.com/antoinezambelli/forge/blob/main/examples/foreign_loop.py)
- [CONTRIBUTING.md](https://github.com/antoinezambelli/forge/blob/main/CONTRIBUTING.md)
# 快速开始教程
本教程将帮助你在 5 分钟内上手 Forge 框架，构建第一个 LLM 驱动的工具调用工作流。Forge 是一个用于管理 LLM 与工具交互的框架，提供了工作流管理、上下文管理、护栏校验和多后端支持等核心功能。
## 环境准备
### 系统要求
| 要求 | 说明 |
|------|------|
| Python 版本 | 3.12+ |
| 操作系统 | macOS, Linux, Windows |
| LLM 后端 | Ollama、llama.cpp server 或 llamafile |
### 安装步骤
```bash
git clone https://github.com/antoinezambelli/forge.git
cd forge
python -m venv .venv
pip install -e ".[dev]"
```
资料来源：[CONTRIBUTING.md:1-10]()
### 运行测试验证安装
```bash
# 完整单元测试（865 个测试，无需 LLM 后端）
python -m pytest tests/unit/ -v --tb=short
# 单文件测试
python -m pytest tests/unit/test_runner.py -v
```
资料来源：[CONTRIBUTING.md:14-22]()
## 核心概念
### 组件架构概览
```mermaid
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
```
### 核心组件表
| 组件 | 文件位置 | 职责 |
|------|----------|------|
| `Workflow` | `src/forge/core/workflow.py` | 定义工作流的工具、步骤和提示模板 |
| `ToolDef` | `src/forge/core/workflow.py:60-70` | 绑定工具 Schema 与实际可调用函数 |
| `ToolSpec` | `src/forge/core/workflow.py` | 定义工具的名称、描述和参数 Schema |
| `WorkflowRunner` | `src/forge/core/runner.py` | 执行工作流，协调 LLM 与工具调用 |
| `OllamaClient` | `src/forge/clients/ollama.py` | Ollama 后端适配器 |
| `ContextManager` | `src/forge/core/context.py` | 管理上下文窗口和上下文压缩 |
| `Guardrails` | `src/forge/guardrails/guardrails.py` | 验证 LLM 响应，执行步骤校验 |
资料来源：[README.md:1-35](), [src/forge/core/workflow.py:1-75]()
## 快速开始示例
### 基础天气查询工作流
以下是一个完整的基础示例，展示了如何使用 Forge 构建天气查询工具调用工作流：
```python
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]()
## 工作流执行流程
### 内部执行步骤
```mermaid
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 与实际实现绑定：
```python
@dataclass
class ToolDef:
"""将工具 Schema 绑定到其实现。
Prerequisites 表达条件依赖："如果你调用这个工具，
必须先调用工具 X"。
"""
spec: ToolSpec # 工具规范（名称、描述、参数）
callable: Callable # 实际执行的函数
prerequisites: list[str | dict[str, str]] = field(default_factory=list)
```
| 字段 | 类型 | 说明 |
|------|------|------|
| `spec` | `ToolSpec` | 工具的规范定义 |
| `callable` | `Callable` | Python 可调用对象 |
| `prerequisites` | `list` | 前置依赖工具列表 |
资料来源：[src/forge/core/workflow.py:60-75]()
## 多步骤工作流
### 带步骤要求的工作流
对于复杂任务，可以定义必须按顺序执行的步骤：
```python
# 定义工具
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 响应：
```mermaid
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[执行工具]
```
| 护栏组件 | 职责 |
|----------|------|
| `ResponseValidator` | 验证响应格式，提取工具调用 |
| `StepEnforcer` | 确保必需步骤已完成 |
| `ErrorTracker` | 跟踪错误次数和重试状态 |
资料来源：[src/forge/guardrails/guardrails.py:1-80]()
## 上下文管理
### TieredCompact 策略
`TieredCompact` 是推荐的上下文压缩策略：
```python
from forge import ContextManager, TieredCompact
ctx = ContextManager(
strategy=TieredCompact(keep_recent=2), # 保留最近 2 轮完整
budget_tokens=8192
)
```
| 参数 | 说明 |
|------|------|
| `keep_recent` | 保留最近的完整消息轮数 |
| `budget_tokens` | 上下文预算（令牌数）|
### 预算模式
| 模式 | 说明 |
|------|------|
| `FORGE_FAST` | 快速模式，较小预算 |
| `FORGE_BALANCED` | 平衡模式 |
| `FORGE_DEEP` | 深度模式，较大预算 |
| `MANUAL` | 手动指定令牌数 |
资料来源：[src/forge/server.py:1-50]()
## 常用配置
### OllamaClient 配置
```python
client = OllamaClient(
model="ministral-3:8b-instruct-2512-q4_K_M",
recommended_sampling=True # 使用推荐的采样参数
)
```
### 推荐采样参数
`recommended_sampling=True` 时，Forge 会自动应用模型卡推荐的采样参数：
| 参数 | 说明 |
|------|------|
| `temperature` | 生成温度 |
| `top_p` | Top-p 采样 |
| `top_k` | Top-k 采样 |
| `min_p` | 最小概率阈值 |
| `repeat_penalty` | 重复惩罚 |
资料来源：[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 的护栏组件进行验证：
```python
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 服务已启动：
```bash
ollama serve
```
### 上下文超出限制
调整 `budget_tokens` 或使用更激进的压缩策略：
```python
ctx = ContextManager(
strategy=TieredCompact(keep_recent=1), # 减少保留轮数
budget_tokens=4096 # 减小预算
)
```
### 模型不支持
检查 `MODEL_SAMPLING_DEFAULTS` 是否包含该模型，或禁用推荐采样：
```python
client = OllamaClient(
model="your-model",
recommended_sampling=False # 不使用推荐参数
)
```
## 下一步
- 阅读 [用户指南](docs/USER_GUIDE.md) 了解高级特性
- 查看 [MODEL_GUIDE.md](docs/MODEL_GUIDE.md) 了解支持的模型
- 运行评估：`python -m tests.eval.eval_runner --scenarios your_scenario --runs 5`
---
## 架构概述
### 相关页面
相关主题：[核心组件详解](#page-core-components), [工作流内部机制](#page-workflow-internals)
相关源码文件
以下源码文件用于生成本页说明：
- [src/forge/core/runner.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/runner.py)
- [src/forge/context/manager.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/manager.py)
- [src/forge/server.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
- [src/forge/core/workflow.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
- [src/forge/guardrails/guardrails.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py)
- [CONTRIBUTING.md](https://github.com/antoinezambelli/forge/blob/main/CONTRIBUTING.md)
- [README.md](https://github.com/antoinezambelli/forge/blob/main/README.md)
# 架构概述
Forge 是一个基于 LLM（大型语言模型）的自动化工作流框架，专注于工具调用（Tool Calling）和上下文管理。其核心设计理念是通过结构化的 Workflow、可靠的工具执行机制以及智能的上下文压缩，实现可预测、可重复的 LLM 驱动任务自动化。
## 核心设计原则
| 原则 | 说明 |
|------|------|
| **异步优先** | 全部采用 `asyncio` 实现，所有客户端方法和运行器均为异步 资料来源：[CONTRIBUTING.md:32]() |
| **类型安全** | 使用 Pydantic 定义工具参数模式和 API 响应模型 |
| **可拔插架构** | 支持多种 LLM 后端适配器，客户端层完全解耦 |
| **Guardrail 机制** | 通过验证器和强制器确保工作流执行的正确性 |
## 系统架构图
```mermaid
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` 是任务定义的中心模型，包含工具注册、步骤约束和终止条件：
| 属性 | 类型 | 说明 |
|------|------|------|
| `name` | str | 工作流名称 |
| `description` | str | 工作流描述 |
| `tools` | dict[str, ToolDef] | 工具定义字典 |
| `required_steps` | list[str] | 必须按顺序执行的步骤 |
| `terminal_tool` | str | 终止工具（工作流结束时必须调用） |
资料来源：[src/forge/core/workflow.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
#### ToolDef 结构
```mermaid
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 交互、工具调用和上下文管理：
```mermaid
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 预算：
| 模式 | 说明 |
|------|------|
| `TieredCompact` | 分层压缩策略，保留最近 N 轮对话 |
| `MANUAL` | 手动指定 token 预算 |
| `FORGE_FAST` | 快速模式预算 |
| `FULL` | 完整上下文模式 |
资料来源：[src/forge/server.py:140-150]()
```mermaid
graph LR
A[用户输入] --> B[构建消息]
B --> C{Token 超限?}
C -->|否| D[发送给 LLM]
C -->|是| E[执行压缩]
E --> D
D --> F[收集响应]
F --> G[工具执行]
G --> B
```
#### TieredCompact 策略
```python
# 示例配置
ctx = ContextManager(
strategy=TieredCompact(keep_recent=2),
budget_tokens=8192
)
```
资料来源：[README.md:18]()
### 4. Guardrails（安全护栏）
Guardrails 是确保工作流正确执行的核心机制，由三个子组件构成：
```mermaid
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 参数
| 参数 | 默认值 | 说明 |
|------|--------|------|
| `tool_names` | — | 可用工具名称列表 |
| `terminal_tool` | — | 终止工具名称 |
| `required_steps` | None | 必须按顺序执行的步骤列表 |
| `max_retries` | 3 | 最大重试次数 |
| `max_tool_errors` | 2 | 最大工具错误数 |
| `rescue_enabled` | True | 启用救援解析 |
| `max_premature_attempts` | 3 | 提前终止尝试次数上限 |
| `retry_nudge` | None | 自定义重试提示函数 |
#### CheckResult 动作
| 动作 | 说明 |
|------|------|
| `proceed` | 继续执行工具调用 |
| `retry` | 需要重试并显示 nudge 提示 |
| `step_blocked` | 步骤被阻止，等待正确步骤 |
| `fatal` | 致命错误，终止工作流 |
### 5. LLM 后端适配器
Forge 支持多种 LLM 后端，通过统一接口抽象差异：
```mermaid
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
```
#### 支持的后端
| 后端 | 配置文件 | 说明 |
|------|----------|------|
| Ollama | `model` 参数 | 本地模型服务 |
| Llamafile | `gguf_path` 参数 | 单文件 GGUF 格式 |
| Llama Server | `gguf_path` 参数 | llama.cpp 服务器 |
| Anthropic | API Key | Claude 系列模型 |
资料来源：[src/forge/server.py:90-110]()
### 6. ServerManager（服务器管理器）
`ServerManager` 负责后端进程的生命周期管理：
```mermaid
stateDiagram-v2
[*] --> Idle: 创建实例
Idle --> Running: start()
Running --> Idle: stop()
Running --> Running: start() (不同模型)
note right of Running: 自动复用相同配置
note right of Idle: 进程已终止
```
#### 缓存优化机制
ServerManager 会缓存当前运行的配置，仅在配置变更时重启服务器：
```python
# 配置相等性检查
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 模式解析
```mermaid
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[获取服务器完整上下文]
```
## 数据流图
```mermaid
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
```
## 快速启动示例
```python
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 的架构围绕三个核心目标设计：
| 目标 | 实现方式 |
|------|----------|
| **可靠性** | Guardrails 确保工具调用正确性和工作流完整性 |
| **效率** | TieredCompact 上下文压缩保持在 token 预算内 |
| **可扩展性** | 插件化客户端支持多种 LLM 后端 |
整个系统基于异步架构设计，所有核心组件（Runner、Client、ServerManager）均支持并发执行，确保在多任务场景下的高性能表现。
---
## 核心组件详解
### 相关页面
相关主题：[架构概述](#page-architecture), [上下文管理](#page-context-management), [SlotWorker 槽位调度](#page-slot-worker)
相关源码文件
以下源码文件用于生成本页说明：
- [src/forge/core/runner.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/runner.py)
- [src/forge/core/steps.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/steps.py)
- [src/forge/core/workflow.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
- [src/forge/core/messages.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/messages.py)
- [src/forge/context/manager.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/manager.py)
- [src/forge/guardrails/guardrails.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py)
- [src/forge/server.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
# 核心组件详解
## 概述
Forge 是一个基于 LLM 的工具调用框架，其核心组件负责管理对话工作流、执行步骤追踪、上下文管理和后端服务调度。整个系统采用异步架构设计，通过 `WorkflowRunner` 协调各个组件完成复杂的多轮对话任务。
核心组件按照职责划分为以下几个模块：
| 模块 | 文件路径 | 主要职责 |
|------|----------|----------|
| 工作流引擎 | `src/forge/core/runner.py` | 执行 Workflow、管理会话生命周期 |
| 步骤追踪 | `src/forge/core/steps.py` | 追踪必需步骤和前置依赖 |
| 消息管理 | `src/forge/core/messages.py` | 管理对话历史和消息结构 |
| 上下文管理 | `src/forge/context/manager.py` | 上下文压缩和预算控制 |
| 守卫检查 | `src/forge/guardrails/guardrails.py` | 响应验证和重试机制 |
| 服务管理 | `src/forge/server.py` | LLM 后端服务的生命周期管理 |
资料来源：[src/forge/core/workflow.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
---
## 工作流引擎 (WorkflowRunner)
### 架构概述
`WorkflowRunner` 是 Forge 框架的核心执行引擎，负责协调整个对话工作流的生命周期。它接收用户输入、与 LLM 后端交互、执行工具调用，并管理上下文预算。
```mermaid
graph TD
A[用户输入] --> B[WorkflowRunner.run]
B --> C[初始化 StepTracker]
C --> D[LLM 推理]
D --> E{Guardrails 检查}
E -->|通过| F{工具调用?}
E -->|失败| G[重试或终止]
F -->|是| H[执行工具]
F -->|否| I[返回结果]
H --> J[上下文压缩]
J --> D
I --> K[返回最终响应]
```
资料来源：[src/forge/core/runner.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/runner.py)
### 核心数据结构
#### Workflow
`Workflow` 是工作流的定义容器，包含工具配置、必需步骤和终端条件：
```python
@dataclass
class Workflow:
name: str # 工作流名称
description: str # 描述
tools: dict[str, ToolDef] # 工具定义字典
required_steps: list[str] # 必需执行步骤列表
terminal_tool: str | frozenset[str] # 终端工具名称
system_prompt_template: str # 系统提示模板
```
资料来源：[src/forge/core/workflow.py:1-50](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
#### ToolDef
`ToolDef` 将工具模式绑定到具体实现：
```python
@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:85-100](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
#### ToolSpec
`ToolSpec` 定义工具的外部接口规范：
```python
class ToolSpec(BaseModel):
"""LLM 看到的工具规格"""
name: str
description: str
parameters: type[BaseModel] # Pydantic 模型
```
资料来源：[src/forge/core/workflow.py:20-30](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
---
## 步骤追踪系统 (StepTracker)
### 功能概述
`StepTracker` 负责追踪必需步骤的完成状态和工具执行历史，用于强制执行前置条件约束。该组件独立于消息历史存在，上下文压缩不会影响步骤完成状态。
```mermaid
graph LR
A[ToolCall] --> B{检查前置条件}
B -->|满足| C[记录执行]
B -->|不满足| D[返回错误]
C --> E[更新 completed_steps]
C --> F[更新 executed_tools]
```
资料来源：[src/forge/core/steps.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/steps.py)
### 核心数据结构
| 类/数据结构 | 职责 |
|------------|------|
| `StepTracker` | 追踪已完成步骤和已执行工具 |
| `PrerequisiteCheck` | 前置条件检查结果 |
#### StepTracker
```python
@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)
def record(self, tool_name: str, args: dict[str, Any] | None = None) -> None:
"""记录成功的工具执行"""
self.completed_steps[tool_name] = None
self.executed_tools.setdefault(tool_name, []).append(args or {})
def is_satisfied(self) -> bool:
"""所有必需步骤是否都已调用"""
return all(s in self.completed_steps for s in self.required_steps)
```
#### PrerequisiteCheck
```python
@dataclass
class PrerequisiteCheck:
"""工具调用的前置条件检查结果"""
satisfied: bool
missing: list[str] # 未满足的前置工具名列表
```
资料来源：[src/forge/core/steps.py:1-50](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/steps.py)
---
## 上下文管理器 (ContextManager)
### 功能概述
`ContextManager` 负责管理对话上下文的大小和 token 预算。通过上下文压缩策略，确保长对话不会超出模型的上下文窗口限制。
### 压缩策略
框架支持多种上下文压缩策略，主要包括：
| 策略 | 描述 |
|------|------|
| `TieredCompact` | 分层压缩，保留最近 N 条消息 |
| `KeepRecentStrategy` | 仅保留最近的固定数量消息 |
### 预算模式
| 模式 | 说明 |
|------|------|
| `FORGE_FAST` | 快速模式，使用较小的上下文预算 |
| `FORGE_BALANCED` | 平衡模式 |
| `MANUAL` | 手动指定 token 数量 |
资料来源：[src/forge/context/manager.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/manager.py)
---
## 守卫系统 (Guardrails)
### 架构概述
守卫系统通过多层检查确保 LLM 响应符合预期，包括响应验证、前置步骤强制和错误追踪。
```mermaid
graph TD
A[LLM Response] --> B[ResponseValidator]
B --> C{格式正确?}
C -->|是| D[StepEnforcer]
C -->|否| E[Rescue Parser]
E --> F{解析成功?}
F -->|是| D
F -->|否| G[Retry Nudge]
D --> H{步骤满足?}
H -->|是| I[Tool Execution]
H -->|否| J[ErrorTracker]
G --> K[重新发送 LLM]
```
资料来源：[src/forge/guardrails/guardrails.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py)
### 核心组件
| 组件 | 职责 |
|------|------|
| `ResponseValidator` | 验证响应格式和工具调用提取 |
| `StepEnforcer` | 强制执行必需步骤 |
| `ErrorTracker` | 追踪重试次数和错误类型 |
### Guardrails 配置参数
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `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` | 自定义重试提示 |
资料来源：[src/forge/guardrails/guardrails.py:50-80](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py)
### CheckResult 返回值
`Guardrails.check()` 方法返回 `CheckResult`，包含：
| 字段 | 类型 | 说明 |
|------|------|------|
| `status` | `str` | `"pass"`, `"retry"`, `"fatal"` |
| `message` | `str \| None` | 状态消息 |
| `nudge` | `str \| None` | 重试时发送给 LLM 的提示 |
---
## 服务管理器 (ServerManager)
### 功能概述
`ServerManager` 负责管理 LLM 后端服务的生命周期，包括启动、停止和配置。
### 支持的后端
| 后端 | 配置要求 | 说明 |
|------|----------|------|
| `ollama` | `model` 参数 | 使用 Ollama 模型 |
| `llamaserver` | `gguf_path` 参数 | 本地 GGUF 文件 |
| `llamafile` | `gguf_path` 参数 | Llamafile 可执行文件 |
### 核心功能
```python
class ServerManager:
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,
cache_type_v: str | None = None,
n_slots: int | None = None,
kv_unified: bool = False,
) -> None:
"""启动后端服务"""
```
### 配置参数
| 参数 | 类型 | 说明 |
|------|------|------|
| `model` | `str` | 模型标识符 |
| `gguf_path` | `str \| Path` | GGUF 文件路径 |
| `mode` | `str` | 运行模式 (`native`, `reasoning` 等) |
| `extra_flags` | `list[str] \| None` | 额外的 CLI 参数 |
| `ctx_override` | `int \| None` | 上下文长度覆盖 |
| `cache_type_k` | `str \| None` | KV 缓存键量化类型 |
| `cache_type_v` | `str \| None` | KV 缓存值量化类型 |
| `n_slots` | `int \| None` | 并发槽位数 |
| `kv_unified` | `bool` | 统一 KV 缓存 |
资料来源：[src/forge/server.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
---
## 消息系统 (Messages)
### 消息类型
框架定义了标准化的消息类型用于通信：
| 消息类型 | 描述 |
|----------|------|
| `UserMessage` | 用户输入消息 |
| `AssistantMessage` | 助手响应消息 |
| `ToolMessage` | 工具执行结果消息 |
| `SystemMessage` | 系统级消息 |
### ToolCall 结构
`ToolCall` 是 LLM 返回的标准化工具调用表示：
```python
class ToolCall(BaseModel):
"""LLM 返回的已验证工具调用"""
tool: str
```
资料来源：[src/forge/core/workflow.py:110-115](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
---
## 组件协作流程
### 完整执行流程
```mermaid
sequenceDiagram
participant User as 用户
participant Runner as WorkflowRunner
participant LLM as LLM Client
participant Step as StepTracker
participant Context as ContextManager
participant Guard as Guardrails
participant Tool as 工具执行
User->>Runner: run(workflow, user_input)
Runner->>Step: 初始化 StepTracker
Runner->>Context: 初始化 ContextManager
Runner->>LLM: 发送推理请求
LLM-->>Runner: LLMResponse
Runner->>Guard: check(response)
Guard->>Step: 检查前置条件
Step-->>Guard: PrerequisiteCheck
alt 响应通过
Guard-->>Runner: CheckResult.pass
Runner->>Tool: 执行工具
Tool-->>Runner: 工具结果
Runner->>Context: 压缩上下文
Runner->>LLM: 发送下一轮
else 需要重试
Guard-->>Runner: CheckResult.retry + nudge
Runner->>LLM: 重试
else 致命错误
Guard-->>Runner: CheckResult.fatal
Runner-->>User: 返回错误
end
```
### 错误恢复机制
| 错误类型 | 处理策略 | 最大重试 |
|----------|----------|----------|
| 格式错误 | 触发救援解析 | 由 `max_retries` 控制 |
| 工具执行失败 | 记录错误，重试 | 由 `max_tool_errors` 控制 |
| 过早终止 | 发送提示重试 | 由 `max_premature_attempts` 控制 |
| 超时 | 记录超时错误 | 300s 固定超时 |
---
## 快速开始示例
```python
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, "巴黎天气怎么样？")
asyncio.run(main())
```
资料来源：[README.md](https://github.com/antoinezambelli/forge/blob/main/README.md)
---
## 总结
Forge 框架的核心组件通过清晰的职责划分和异步架构，提供了强大的 LLM 工具调用能力：
- **WorkflowRunner** 作为中央协调器，管理整个执行生命周期
- **StepTracker** 确保必需步骤的执行顺序和前置条件满足
- **ContextManager** 处理长对话的上下文压缩
- **Guardrails** 提供多层次的响应验证和错误恢复
- **ServerManager** 抽象了不同 LLM 后端的差异
这些组件协同工作，使得 Forge 能够可靠地执行复杂的多轮对话任务。
---
## 工作流内部机制
### 相关页面
相关主题：[核心组件详解](#page-core-components), [内置工具系统](#page-tools)
相关源码文件
以下源码文件用于生成本页说明：
- [src/forge/core/workflow.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
- [src/forge/core/steps.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/steps.py)
- [src/forge/guardrails/guardrails.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py)
- [src/forge/prompts/nudges.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/prompts/nudges.py)
- [examples/foreign_loop.py](https://github.com/antoinezambelli/forge/blob/main/examples/foreign_loop.py)
# 工作流内部机制
## 概述
Forge 的工作流系统是一个异步优先的 LLM 工具调用编排框架，核心设计目标是将复杂的 LLM 交互与工具执行解耦。工作流由 `Workflow` 类定义，封装了工具定义、步骤要求、终端工具等元信息，通过 `WorkflowRunner` 驱动执行。
工作流机制包含三个核心子系统：
1. **工具定义与绑定系统** — 定义可用工具及其参数模式
2. **步骤追踪与前置条件系统** — 确保工具按正确顺序调用
3. **Guardrails 保护系统** — 验证响应、处理重试、阻止过早终止
资料来源：[src/forge/core/workflow.py:1-50]()
## 核心数据模型
### Workflow 类
`Workflow` 是工作流的核心抽象，作为 dataclass 存储所有元信息。
```python
@dataclass
class Workflow:
name: str
description: str
tools: dict[str, ToolDef]
required_steps: list[str]
terminal_tool: str
system_prompt_template: str
```
| 属性 | 类型 | 说明 |
|------|------|------|
| `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 参数规范。
```python
@dataclass
class ToolDef:
spec: ToolSpec
callable: Callable[..., Any]
prerequisites: list[str | dict[str, str]] = field(default_factory=list)
```
`prerequisites` 字段支持两种前置条件表达方式：
| 类型 | 示例 | 含义 |
|------|------|------|
| `str` | `"read_file"` | 调用过任意参数的 read_file 即可 |
| `dict` | `{"tool": "read_file", "match_arg": "path"}` | 必须调用过 path 参数相同的 read_file |
资料来源：[src/forge/core/workflow.py:85-110]()
### ToolCall 数据模型
`ToolCall` 是 LLM 返回的工具调用验证结果。
```python
class ToolCall(BaseModel):
tool: str
args: dict[str, Any] = {}
reasoning: str | None = None
```
| 字段 | 类型 | 说明 |
|------|------|------|
| `tool` | `str` | 工具名称 |
| `args` | `dict[str, Any]` | 工具参数 |
| `reasoning` | `str \| None` | 推理过程（用于带 thinking 的模型） |
资料来源：[src/forge/core/workflow.py:120-130]()
## 步骤追踪机制
### StepTracker
`StepTracker` 维护在工作流运行器上，独立于消息历史，紧跟在上下文压缩之外。
```python
@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)
```
| 方法 | 功能 |
|------|------|
| `record(tool_name, args)` | 记录工具已执行及参数 |
| `is_satisfied()` | 检查所有必需步骤是否完成 |
| `pending()` | 返回未完成步骤列表 |
资料来源：[src/forge/core/steps.py:30-55]()
### PrerequisiteCheck
前置条件检查结果用于判断工具调用是否满足依赖要求。
```python
@dataclass
class PrerequisiteCheck:
satisfied: bool
missing: list[str]
```
资料来源：[src/forge/core/steps.py:15-25]()
## Guardrails 系统
Guardrails 是 Forge 的保护中间件，将三个独立检查器组合成统一的两阶段 API：
```mermaid
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 结果类型
```python
action: Literal["execute", "retry", "step_blocked", "fatal"]
tool_calls: list[ToolCall] | None
nudge: Nudge | None
reason: str | None
```
| Action | 含义 | 后续处理 |
|--------|------|----------|
| `execute` | 响应有效，可执行工具 | 执行 tool_calls |
| `retry` | 响应无效需重试 | 注入 nudge 后重新请求 |
| `step_blocked` | 违反步骤顺序 | 注入 nudge 后重新请求 |
| `fatal` | 达到最大重试次数 | 终止工作流 |
资料来源：[src/forge/guardrails/guardrails.py:140-160]()
### ResponseValidator
验证 LLM 响应是否包含有效工具调用，支持救援解析。
```python
class ResponseValidator:
def __init__(
self,
tool_names: list[str],
rescue_enabled: bool = True,
retry_nudge_fn: Callable[[str], str] | None = None,
)
```
| 参数 | 默认值 | 说明 |
|------|--------|------|
| `tool_names` | — | 有效工具名列表 |
| `rescue_enabled` | `True` | 是否启用救援解析 |
| `retry_nudge_fn` | `None` | 自定义重试提示生成器 |
救援解析支持以下格式：
- OpenAI 格式：`{"name": ..., "arguments": ...}`
- Qwen Coder XML 格式：`value`
资料来源：[src/forge/guardrails/guardrails.py:40-80]()
### StepEnforcer
强制执行步骤顺序，防止过早调用终端工具。
```python
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
追踪连续错误和工具执行失败。
```python
class ErrorTracker:
def __init__(
self,
max_retries: int = 3,
max_tool_errors: int = 2,
)
```
| 参数 | 默认值 | 说明 |
|------|--------|------|
| `max_retries` | `3` | 连续错误响应后终止 |
| `max_tool_errors` | `2` | 连续工具执行失败后终止 |
资料来源：[src/forge/guardrails/guardrails.py:110-140]()
## 提示模板与 Nudge
### 步骤阻止提示
当模型尝试在完成必需步骤前调用终端工具时，系统注入阻止提示：
```python
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}."
```
| 层级 | 语气强度 | 使用场景 |
|------|----------|----------|
| 1 | 礼貌提示 | 首次违规 |
| 2 | 直接命令 | 第二次违规 |
| 3 | 强烈警告 | 第三次违规 |
资料来源：[src/forge/prompts/nudges.py:1-30]()
### 前置条件提示
当工具调用缺少前置依赖时触发：
```python
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]()
## 工作流执行流程
### 完整执行状态机
```mermaid
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（全部集成）**
```python
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（细粒度控制）**
```python
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]()
## 快速开始示例
```python
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` 实现，支持高并发工具调用场景：
```python
async def run(self, workflow: Workflow, user_input: str) -> WorkflowResult:
...
```
资料来源：[src/forge/core/workflow.py:1-30]()
### 上下文管理隔离
`ContextManager` 独立于步骤追踪，允许上下文压缩不影响步骤完成状态：
```mermaid
graph LR
subgraph WorkflowRunner
A[StepTracker] --- B[ContextManager]
end
B --- C[消息历史]
A --- D[completed_steps]
```
### 工具参数模式
使用 Pydantic 模型定义工具参数，确保类型安全：
```python
parameters: type[BaseModel] # 必须是 Pydantic 模型
```
可通过 `get_json_schema()` 方法导出 JSON Schema：
```python
def get_json_schema(self) -> dict[str, Any]:
return self.parameters.model_json_schema()
```
资料来源：[src/forge/core/workflow.py:50-65]()
## 总结
Forge 的工作流机制通过清晰的职责分离实现了可靠的 LLM 工具调用编排：
| 组件 | 职责 |
|------|------|
| `Workflow` | 定义工作流元信息和工具集合 |
| `StepTracker` | 追踪步骤完成状态和前置条件 |
| `Guardrails` | 统一验证、重试、阻止逻辑 |
| `WorkflowRunner` | 协调执行流程和上下文管理 |
该设计支持从简单的单工具调用到复杂的多步骤工作流，同时为外部编排系统提供了灵活的中间件接口。
---
## Guardrails 系统
### 相关页面
相关主题：[工作流内部机制](#page-workflow-internals), [核心组件详解](#page-core-components)
相关源码文件
以下源码文件用于生成本页说明：
- [src/forge/guardrails/__init__.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/__init__.py)
- [src/forge/guardrails/response_validator.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/response_validator.py)
- [src/forge/guardrails/step_enforcer.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/step_enforcer.py)
- [src/forge/guardrails/error_tracker.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/error_tracker.py)
- [src/forge/guardrails/nudge.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/nudge.py)
- [src/forge/guardrails/guardrails.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py)
- [src/forge/prompts/nudges.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/prompts/nudges.py)
# Guardrails 系统
## 概述
Guardrails 系统是 Forge 框架的核心安全与流程控制中间件，设计用于在外部编排循环中确保 LLM 响应符合预期的工作流约束。该系统通过三重验证机制（响应验证、步骤强制、错误追踪）保护工具调用流程的完整性和可靠性。 资料来源：[src/forge/guardrails/guardrails.py:46-52]()
## 系统架构
Guardrails 系统由四个核心组件构成，它们协同工作以实现完整的中间件功能：
```mermaid
graph TB
subgraph Guardrails系统
RV[ResponseValidator
响应验证器]
SE[StepEnforcer
步骤强制器]
ET[ErrorTracker
错误追踪器]
NR[Nudge模块
提示生成]
end
LLM[LLM响应] --> RV
RV --> SE
SE --> ET
NR --> SE
subgraph CheckResult
AC[action]
TC[tool_calls]
ND[nudge]
RS[reason]
end
ET --> CheckResult
```
| 组件 | 职责 | 文件位置 |
|------|------|----------|
| ResponseValidator | 解析 LLM 响应，提取工具调用，处理文本救援 | `response_validator.py` |
| StepEnforcer | 强制执行必需步骤顺序，阻止提前终止 | `step_enforcer.py` |
| ErrorTracker | 追踪连续重试和工具错误次数 | `error_tracker.py` |
| Nudge | 生成用户友好的引导提示消息 | `nudge.py` / `prompts/nudges.py` |
资料来源：[src/forge/guardrails/guardrails.py:24-45]()
## CheckResult 数据模型
`check()` 方法返回 `CheckResult` 对象，包含四个字段：
| 字段 | 类型 | 说明 |
|------|------|------|
| `action` | `Literal["execute", "retry", "step_blocked", "fatal"]` | 下一步操作指令 |
| `tool_calls` | `list[ToolCall] \| None` | 提取的工具调用列表 |
| `nudge` | `Nudge \| None` | 注入的提示消息（retry/step_blocked 时设置） |
| `reason` | `str \| None` | 人类可读的解释（仅 fatal 时设置） |
```python
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
### 构造函数参数
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `tool_names` | `list[str]` | 必需 | 有效工具名称列表 |
| `terminal_tool` | `str \| frozenset[str]` | 必需 | 可结束工作流的工具 |
| `required_steps` | `list[str] \| None` | `None` | 终端工具前必须调用的步骤 |
| `max_retries` | `int` | `3` | 连续错误响应后返回 fatal 的阈值 |
| `max_tool_errors` | `int` | `2` | 工具执行失败后耗尽的阈值 |
| `rescue_enabled` | `bool` | `True` | 是否启用文本救援解析 |
| `max_premature_attempts` | `int` | `3` | 提前调用终端工具的次数上限 |
| `retry_nudge` | `Callable[[str], str] \| None` | `None` | 自定义重试提示生成器 |
资料来源：[src/forge/guardrails/guardrails.py:76-91]()
### 主要方法
#### check() 方法
在每次 LLM 响应后、执行任何工具前调用此方法。它依次通过三个验证组件：
```mermaid
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() 方法
在工具执行完成后调用，用于更新错误追踪器状态：
```python
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（步骤强制器）
确保工作流按正确的步骤顺序执行：
```mermaid
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（错误追踪器）
维护错误状态机，防止无限重试循环：
| 错误类型 | 计数器 | 达到上限后果 |
|----------|--------|--------------|
| 连续重试 | `_consecutive_retries` | `check()` 返回 `"fatal"` |
| 工具错误 | `_consecutive_tool_errors` | 工作流标记为"耗尽" |
当工具成功执行时，两个计数器都会重置。
资料来源：[src/forge/guardrails/error_tracker.py]()
## Nudge 提示系统
Nudge 模块负责生成用户友好的引导消息，帮助模型理解并修正其行为。
### step_nudge() 函数
当模型试图跳过必需步骤直接调用终端工具时触发：
| tier 值 | 语气强度 | 示例消息 |
|---------|----------|----------|
| 1 | 礼貌 | "You cannot call answer yet. You must first complete these required steps: search, lookup." |
| 2 | 直接 | "You must call one of these tools now: search, lookup. Pick one." |
| 3 | 强制 | "STOP. You MUST call one of: search, lookup. Do NOT call answer." |
```python
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() 函数
当模型调用带有前置依赖的工具但未满足前置条件时触发：
```python
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` 类进行一站式验证：
```python
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（高级）
直接访问各个验证组件，实现自定义行为：
```python
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()` 工具让模型直接返回文本响应：
```python
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 工作流状态图
```mermaid
stateDiagram-v2
[*] --> 等待响应
等待响应 --> 验证响应: LLM 返回
验证响应 --> 响应有效: 工具调用已提取
验证响应 --> 响应无效: 解析失败
响应无效 --> 检查重试次数: 递增 retry
检查重试次数 --> 等待响应: retry_nudge
检查重试次数 --> 致命错误: 达到上限
响应有效 --> 检查步骤顺序
检查步骤顺序 --> 步骤通过: 验证通过
检查步骤顺序 --> 步骤受阻: 缺少步骤
步骤受阻 --> 重置提前尝试: step_nudge
步骤通过 --> 执行工具
执行工具 --> 记录执行: record()
记录执行 --> 工作流完成?: terminal reached
记录执行 --> 工具错误?: 工具执行失败
工作流完成? --> [*]: 是
工作流完成? --> 等待响应: 否
工具错误? --> 等待响应: 重试
工具错误? --> [*]: 达到上限
```
## 配置与消融研究
Guardrails 的各个组件可通过 `AblationConfig` 独立切换，用于评估每个防护措施的性能影响：
```python
# 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 工作流提供了可靠的安全防护：
| 特性 | 描述 |
|------|------|
| **模块化设计** | ResponseValidator、StepEnforcer、ErrorTracker 可独立使用 |
| **灵活配置** | 支持自定义重试次数、错误阈值、救援行为 |
| **可观测性** | 通过 `nudge` 机制提供清晰的错误引导 |
| **状态管理** | 内置错误计数和步骤追踪，自动防止循环 |
| **消融支持** | 完整的独立切换能力，便于性能分析 |
---
## 上下文管理
### 相关页面
相关主题：[核心组件详解](#page-core-components), [架构概述](#page-architecture)
相关源码文件
以下源码文件用于生成本页说明：
- [src/forge/context/manager.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/manager.py)
- [src/forge/context/strategies.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/strategies.py)
- [src/forge/context/hardware.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/hardware.py)
- [src/forge/server.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
# 上下文管理
## 概述
上下文管理（Context Management）是 forge 框架中负责管理对话历史和令牌预算的核心系统。在与大语言模型（LLM）进行多轮交互时，上下文窗口的大小直接影响能保留多少历史信息和工具调用记录。forge 的上下文管理系统通过预算模式（Budget Mode）和压缩策略（Compaction Strategy）的组合，确保在有限的令牌预算内最大化有效信息的保留。
该系统位于 `src/forge/context/` 模块下，包含三个核心子模块：
| 模块 | 职责 |
|------|------|
| `manager.py` | 上下文管理的主入口，负责预算解析和消息压缩协调 |
| `strategies.py` | 定义压缩策略，如 `TieredCompact`（分层压缩） |
| `hardware.py` | 硬件感知配置，根据设备资源调整上下文参数 |
资料来源：[CONTRIBUTING.md](https://github.com/antoinezambelli/forge/blob/main/CONTRIBUTING.md)
---
## 核心概念
### 令牌预算（Budget Tokens）
令牌预算是上下文管理的基础参数，定义了允许保留在上下文窗口中的最大令牌数量。在初始化 `ContextManager` 时通过 `budget_tokens` 参数指定：
```python
from forge import ContextManager, TieredCompact
ctx = ContextManager(
strategy=TieredCompact(keep_recent=2),
budget_tokens=8192
)
```
资料来源：[README.md](https://github.com/antoinezambelli/forge/blob/main/README.md)
### 预算模式（Budget Mode）
预算模式决定了如何确定令牌预算的来源。forge 支持多种预算解析策略：
| 模式 | 说明 | 适用场景 |
|------|------|----------|
| `FORGE_FAST` | 快速模式，使用较小的上下文预算 | 简单查询、快速响应 |
| `FORGE_BALANCED` | 平衡模式，在速度和深度间取得平衡 | 常规工作流 |
| `FORGE_DEEP` | 深度模式，使用更大的上下文预算 | 复杂推理任务 |
| `MANUAL` | 手动指定预算值 | 精确控制令牌使用 |
```python
from forge.server import BudgetMode
budget = await server.start_with_budget(
model=identity,
gguf_path=gguf_path or "",
mode=mode,
budget_mode=BudgetMode.FORGE_FAST,
client=client,
)
```
资料来源：[src/forge/server.py:1-50](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
---
## 架构设计
### 系统架构图
```mermaid
graph TD
A[WorkflowRunner] --> B[ContextManager]
B --> C[压缩策略 Strategy]
B --> D[预算解析 Budget Resolution]
C --> E[TieredCompact]
C --> F[其他策略...]
D --> G[ServerManager]
G --> H[服务端上下文查询]
G --> I[手动预算指定]
J[硬件配置 hardware.py] --> B
```
### 预算解析流程
当使用服务端管理时，预算的解析遵循以下逻辑：
```mermaid
graph TD
A[resolve_budget 调用] --> B{预算模式}
B -->|MANUAL| C[使用 manual_tokens]
B -->|OLLAMA| D{模式判断}
B -->|LLAMASERVER| E[查询服务端上下文]
B -->|LLAMAFILE| E
D -->|MANUAL| C
D -->|其他| F[查询服务端上下文]
C --> G[返回预算令牌数]
E --> G
F --> G
```
资料来源：[src/forge/server.py:200-280](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
---
## 分层压缩策略（TieredCompact）
### 核心思想
`TieredCompact` 是一种分层保留策略，其核心思想是保留最近的消息完整不变，同时对较早的消息进行摘要或丢弃。这种策略在保持对话连贯性的同时，最大化地利用有限的令牌预算。
### 参数配置
| 参数 | 类型 | 说明 |
|------|------|------|
| `keep_recent` | `int` | 保留最近 N 轮完整对话 | 默认值：2 |
| `budget_tokens` | `int` | 总令牌预算 | 继承自 ContextManager |
```python
# 保留最近 2 轮完整消息，剩余预算用于早期消息摘要
strategy = TieredCompact(keep_recent=2)
```
资料来源：[README.md](https://github.com/antoinezambelli/forge/blob/main/README.md)
### 工作流程
```mermaid
graph LR
A[消息列表] --> B[识别最近 N 轮]
B --> C[保留完整]
D[早期消息] --> E{是否可摘要}
E -->|是| F[生成摘要]
E -->|否| G[丢弃]
F --> H[放入上下文]
C --> H
G --> H
```
---
## 硬件感知配置
### 硬件适配
`hardware.py` 模块负责根据运行环境的硬件资源（主要是 GPU 显存）调整上下文配置。这对于在消费级硬件上运行大模型尤为重要。
### 主要考量因素
| 因素 | 影响 |
|------|------|
| GPU 显存大小 | 决定最大并发槽数（n_slots） |
| KV Cache 类型 | 影响上下文长度和质量 |
| 上下文分区策略 | 共享 vs 分区 KV Cache |
```python
# 多槽位配置示例
server = ServerManager(...)
await server.start_with_budget(
n_slots=4, # 4 个并发槽位
kv_unified=True, # 共享 KV Cache
)
```
资料来源：[src/forge/server.py:30-40](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
---
## 服务端上下文管理
### 上下文长度查询
服务端通过 `/props` 端点报告其配置的上下文长度：
```python
async def get_server_context(self) -> int:
"""获取服务端报告的上下文长度"""
try:
props = await self.query_props()
except (httpx.HTTPError, BackendError) as exc:
raise BudgetResolutionError(cause=exc) from exc
ctx = props.get("default_generation_settings", {}).get("n_ctx")
if ctx is None:
raise BudgetResolutionError()
return ctx
```
资料来源：[src/forge/server.py:180-195](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
### KV Cache 配置
对于非 Ollama 后端（llamaserver / llamafile），支持细粒度的 KV Cache 量化配置：
| 参数 | 说明 | 示例值 |
|------|------|--------|
| `cache_type_k` | Key 缓存量化类型 | `"q8_0"`, `"q4_0"` |
| `cache_type_v` | Value 缓存量化类型 | `"q8_0"`, `"q4_0"` |
```python
server = ServerManager(backend="llamaserver", port=8080)
await server.start_with_budget(
cache_type_k="q4_0",
cache_type_v="q8_0",
)
```
资料来源：[src/forge/server.py:25-35](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
---
## 使用指南
### 基础用法
```python
import asyncio
from forge import (
WorkflowRunner,
OllamaClient,
ContextManager,
TieredCompact,
)
async def main():
# 创建上下文管理器
ctx = ContextManager(
strategy=TieredCompact(keep_recent=2),
budget_tokens=8192
)
# 创建客户端和运行器
client = OllamaClient(model="ministral-3:8b-q4_K_M")
runner = WorkflowRunner(client=client, context_manager=ctx)
# 运行工作流
# await runner.run(workflow, user_message)
asyncio.run(main())
```
### 与服务端集成
```python
from forge.server import ServerManager, BudgetMode
async def main():
# 启动服务端并自动解析预算
server, ctx = await ServerManager.start_with_budget(
backend="llamaserver",
gguf_path="/path/to/model.gguf",
budget_mode=BudgetMode.FORGE_BALANCED,
)
# ... 使用 server 和 ctx ...
await server.stop()
asyncio.run(main())
```
---
## 与守卫系统的协同
上下文管理与守卫系统（Guardrails）紧密配合，共同确保工作流的正确执行：
```mermaid
graph LR
A[LLM 响应] --> B[Guardrails.check]
B --> C{响应类型}
C -->|工具调用| D[执行工具]
C -->|文本响应| E[上下文压缩]
C -->|错误| F[重试/提示]
D --> G[ContextManager 压缩]
E --> G
```
守卫系统会拦截 LLM 的响应，进行验证后交由上下文管理器决定是否需要压缩历史记录：
```python
class 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:40-60](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py)
---
## 错误处理
### BudgetResolutionError
当无法解析预算时会抛出此异常，通常发生在服务端不可达或未返回有效上下文长度时：
```python
try:
ctx_length = await server.get_server_context()
except BudgetResolutionError as e:
# 回退到手动指定
ctx = ContextManager(
strategy=TieredCompact(keep_recent=2),
budget_tokens=4096 # 手动指定
)
```
资料来源：[src/forge/server.py:170-180](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py)
---
## 最佳实践
1. **选择合适的预算模式**：简单查询使用 `FORGE_FAST`，复杂推理任务使用 `FORGE_DEEP`
2. **合理设置 keep_recent**：通常 2-3 轮即可保留对话上下文
3. **监控令牌使用**：定期检查实际使用的令牌数，避免溢出
4. **硬件感知配置**：在资源受限环境使用较小的 n_slots 和共享 KV Cache
5. **与服务端配合使用**：利用服务端自动解析上下文长度，减少手动配置
---
## SlotWorker 槽位调度
### 相关页面
相关主题：[核心组件详解](#page-core-components), [架构概述](#page-architecture)
相关源码文件
以下源码文件用于生成本页说明：
- [src/forge/server.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/server.py) — ServerManager 槽位启动参数定义
- [src/forge/core/runner.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/runner.py) — WorkflowRunner 槽位上下文管理引用
- [src/forge/core/steps.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/steps.py) — 步骤追踪与工具执行记录
- [src/forge/guardrails/guardrails.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py) — 响应验证与错误处理
- [CONTRIBUTING.md](https://github.com/antoinezambelli/forge/blob/main/CONTRIBUTING.md) — 槽位调度相关架构说明
# SlotWorker 槽位调度
## 概述
SlotWorker 是 forge 框架中用于管理 **llama-server / llamafile 后端并发槽位** 的核心机制。它通过 `ServerManager` 类实现，支持在同一进程中启动具有多个独立 KV Cache 槽位的推理服务器，从而支持多智能体（multi-agent）架构的并发执行。
槽位调度的核心目的是：**在单个推理服务器实例上并行运行多个独立的 agent 工作流，每个槽位拥有独立的上下文状态，彼此互不干扰。**
资料来源：[src/forge/server.py:1-50]()
## 架构设计
### 槽位模型
forge 的槽位调度采用 **硬分区 + 可选统一缓存** 两种模式：
| 模式 | `kv_unified` | 行为描述 | 适用场景 |
|------|-------------|----------|----------|
| 独立槽位 | `False`（默认） | 每个槽位拥有独立的 KV Cache 切片，上下文长度在槽位创建时硬性划分 | 资源受限环境、需要严格隔离的并发任务 |
| 统一缓存 | `True` | 所有槽位共享单一 KV Cache 池，每个槽位可使用完整上下文容量 | 需要灵活分配上下文的高并发场景 |
资料来源：[src/forge/server.py:35-42]()
### 组件关系图
```mermaid
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` 类负责启动和管理带有槽位配置的推理服务器。其构造函数接受以下参数：
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `backend` | `str` | — | 后端类型：`"ollama"`、`"llamaserver"` 或 `"llamafile"` |
| `port` | `int` | `8080` | 服务端口（llama-server / llamafile 专用） |
| `models_dir` | `str \| Path` | `None` | GGUF 模型文件目录 |
资料来源：[src/forge/server.py:120-133]()
### 槽位状态追踪
`ServerManager` 内部维护以下状态字段用于槽位调度决策：
```python
_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]()
### 启动方法签名
```python
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 缓存分别进行量化，以节省显存：
| 量化类型 | 说明 | 显存节省 | 精度损失 |
|----------|------|----------|----------|
| `q8_0` | 8-bit 量化 | 中等 | 较低 |
| `q4_0` | 4-bit 量化 | 高 | 中等 |
| `q4_K_M` | 混合 4-bit（中等块大小） | 高 | 较低-中等 |
### 槽位复用机制
`ServerManager` 实现了**智能复用逻辑**：当新请求的配置与当前运行配置完全一致时，跳过重启直接复用现有槽位：
```python
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` 与后端交互：
```mermaid
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: 发送推理请求（槽位分配）
```
### 步骤追踪与槽位隔离
每个槽位在执行过程中会独立维护步骤状态：
```python
@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]()
## 槽位调度模式
### 模式对比
| 模式 | 上下文分配 | 资源利用率 | 隔离性 | 配置参数 |
|------|-----------|-----------|--------|----------|
| **独立槽位** | 硬性分区，每个槽位固定容量 | 中等 | 强 | `n_slots=N, kv_unified=False` |
| **统一缓存** | 动态共享，任意槽位可用全部容量 | 高 | 中等 | `n_slots=N, kv_unified=True` |
资料来源：[src/forge/server.py:38-42]()
### 配置示例
```python
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` 模块提供槽位级别的错误追踪：
```python
@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]()
## 最佳实践
### 槽位数量选择
| 模型大小 | 推荐槽位数 | 说明 |
|----------|-----------|------|
| ≤ 8B | 2-4 | 显存充裕时可增加并发 |
| 14B-32B | 1-2 | 需更多显存用于单槽 |
| > 32B | 1 | 大模型建议独占槽位 |
### 缓存量化建议
| 量化类型 | 推荐场景 | 显存占用 |
|----------|----------|----------|
| `q8_0/q8_0` | 高精度需求 | 较高 |
| `q4_K_M/q4_K_M` | 平衡场景 | 中等 |
| `q4_0/q8_0` | 内存受限 | 较低 |
## 限制与已知问题
### llama.cpp reasoning budget 挂起问题
> **已知问题**：2026 年 4 月 10 日之后的构建版本中，Gemma 4、Qwen 3.5 和 Ministral Reasoning 模型使用无界 reasoning budget sampler 会导致静默挂起。
**临时解决方案**：使用 `--reasoning-budget 0` 参数禁用。
资料来源：[CHANGELOG.md:24-28]()
### 后端兼容性
| 后端 | 槽位支持 | 说明 |
|------|---------|------|
| Ollama | ❌ 不支持 | 使用原生模型管理 |
| llama-server | ✅ 完全支持 | 推荐用于多槽位场景 |
| llamafile | ✅ 完全支持 | 与 llama-server 行为一致 |
## 总结
SlotWorker 槽位调度是 forge 框架支持多智能体并发执行的核心基础设施。通过 `ServerManager` 类，开发者可以：
1. **配置并发槽位数量** (`n_slots`) 以支持多 agent 并行
2. **选择 KV Cache 模式** (`kv_unified`) 平衡隔离性与资源利用
3. **优化显存使用** 通过 Key/Value 缓存量化
4. **实现智能复用** 避免不必要的服务器重启
该机制使 forge 能够在单个推理服务器实例上高效运行多个独立工作流，特别适合需要并行评估或异步执行多个 agent 任务的场景。
---
## 内置工具系统
### 相关页面
相关主题：[工作流内部机制](#page-workflow-internals)
相关源码文件
以下源码文件用于生成本页说明：
- [src/forge/tools/__init__.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/tools/__init__.py)
- [src/forge/tools/respond.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/tools/respond.py)
- [src/forge/core/workflow.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/workflow.py)
- [src/forge/guardrails/guardrails.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/guardrails.py)
- [src/forge/guardrails/response_validator.py](https://github.com/antoinezambelli/forge/blob/main/src/forge/guardrails/response_validator.py)
- [examples/foreign_loop.py](https://github.com/antoinezambelli/forge/blob/main/examples/foreign_loop.py)
# 内置工具系统
## 概述
Forge 的内置工具系统是一套用于定义、注册和执行 LLM 工具调用的核心机制。该系统通过 `ToolSpec`、`ToolDef` 和 `ToolCall` 三个核心类实现工具的schema定义与实际执行逻辑的绑定，并为工作流引擎提供验证、执行和依赖管理能力。
工具系统在整个架构中扮演关键角色：
- **工具发现**：工作流通过 `ToolDef` 字典声明可用工具
- **LLM 接口**：工具规格自动转换为 JSON Schema，供 LLM 理解调用格式
- **执行验证**：GuardRails 组件验证工具调用合法性
- **依赖管理**：通过前置条件（Prerequisites）表达工具间依赖关系
资料来源：[src/forge/core/workflow.py:1-50]()
## 核心数据模型
### ToolSpec
`ToolSpec` 是工具的静态声明，使用 Pydantic 模型定义参数schema：
```python
@dataclass
class ToolSpec:
"""Tool specification with name, description, and parameters."""
name: str
description: str
parameters: type[BaseModel]
```
| 字段 | 类型 | 说明 |
|------|------|------|
| `name` | `str` | 工具唯一标识名 |
| `description` | `str` | 工具功能描述，供 LLM 理解用途 |
| `parameters` | `type[BaseModel]` | Pydantic 模型类，定义参数结构 |
`parameters` 字段接收一个 Pydantic `BaseModel` 子类，Forge 自动从中提取 JSON Schema：
```python
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` 将工具规格与实现绑定，并声明执行依赖：
```python
@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)
```
| 字段 | 类型 | 说明 |
|------|------|------|
| `spec` | `ToolSpec` | 工具规格（名称、描述、参数schema） |
| `callable` | `Callable` | 实际执行的 Python 函数 |
| `prerequisites` | `list` | 前置依赖条件列表 |
`prerequisites` 支持两种表达方式：
1. **字符串形式**（名称匹配）：
```python
prerequisites=["read_file"]
```
表示调用此工具前必须曾调用过 `read_file`
2. **字典形式**（参数匹配）：
```python
prerequisites=[{"tool": "read_file", "match_arg": "path"}]
```
表示调用此工具前必须调用过 `read_file` 且 `path` 参数值相同
资料来源：[src/forge/core/workflow.py:143-175]()
### ToolCall
`ToolCall` 是经 LLM 返回并验证后的工具调用对象：
```python
class ToolCall(BaseModel):
"""Validated tool invocation returned by an LLMClient."""
tool: str
```
由 `ResponseValidator` 验证通过后生成，传递给执行层。
资料来源：[src/forge/core/workflow.py:177-179]()
## 工具注册与使用
### 工作流中的工具声明
在创建工作流时，通过 `tools` 字典注册所有可用工具：
```python
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 理解：
```python
from pydantic import BaseModel, Field
class GetWeatherParams(BaseModel):
city: str = Field(description="City name")
```
Pydantic 自动处理类型验证和类型转换。
资料来源：[README.md:5-15]()
## 前置依赖机制
### 依赖声明语法
工具的前置依赖通过 `prerequisites` 字段声明，支持细粒度控制：
```python
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"], # 简单依赖
),
}
```
### 依赖执行流程
```mermaid
graph TD
A[LLM 返回工具调用] --> B[StepEnforcer 检查依赖]
B --> C{依赖满足?}
C -->|是| D[执行工具]
C -->|否| E[返回 Nudge 提示]
D --> F[记录执行结果]
F --> G{所有必需步骤完成?}
G -->|是| H[允许终止工具]
G -->|否| A
```
`StepEnforcer` 组件负责验证依赖是否满足：
```python
enforcer = StepEnforcer(
required_steps=["search", "lookup"],
terminal_tool="answer",
)
```
资料来源：[examples/foreign_loop.py:1-80]()
## 工具验证流程
### ResponseValidator
`ResponseValidator` 负责验证 LLM 返回的工具调用：
1. **工具名称验证**：检查是否在允许列表中
2. **参数验证**：通过 Pydantic 模型验证参数合法性
3. **未知工具检测**：识别并提示未知工具名
```python
validator = ResponseValidator(
tool_names=["search", "lookup", "answer"],
rescue_enabled=True,
)
```
资料来源：[src/forge/guardrails/response_validator.py:1-60]()
### 验证结果处理
验证结果通过 `ValidationResult` 返回：
| 结果 | 说明 |
|------|------|
| `tool_calls` | 验证通过的工具调用列表 |
| `nudge` | 需要重试时的提示信息 |
| `needs_retry` | 是否需要 LLM 重试 |
```python
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` 是工具系统的顶层编排组件，整合验证、依赖检查和错误追踪：
```python
@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:
```
| 参数 | 默认值 | 说明 |
|------|--------|------|
| `tool_names` | - | 允许的工具名称列表 |
| `terminal_tool` | - | 终止工具名称 |
| `required_steps` | `None` | 必需的执行步骤 |
| `max_retries` | `3` | 最大重试次数 |
| `max_tool_errors` | `2` | 最大工具错误数 |
| `rescue_enabled` | `True` | 是否启用救援解析 |
| `max_premature_attempts` | `3` | 提前终止最大次数 |
| `retry_nudge` | `None` | 自定义重试提示函数 |
资料来源：[src/forge/guardrails/guardrails.py:1-50]()
## 架构图
### 工具系统完整架构
```mermaid
graph TB
subgraph "工具定义层"
PS[ToolSpec
工具规格]
PD[ToolDef
工具定义]
PC[Callable
可调用对象]
PR[Prerequisites
前置依赖]
end
subgraph "工作流层"
WF[Workflow
工作流]
SE[StepEnforcer
步骤执行器]
CM[ContextManager
上下文管理]
end
subgraph "验证层"
RV[ResponseValidator
响应验证器]
ET[ErrorTracker
错误追踪]
GR[GuardRails
防护栏]
end
subgraph "LLM 层"
LL[LLM Client
LLM 客户端]
TC[ToolCall
工具调用]
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`
### 参数设计
```python
class SearchParams(BaseModel):
query: str = Field(description="Search query string")
limit: int = Field(default=10, description="Maximum results to return")
```
### 依赖声明
对于多步骤工作流，清晰声明依赖关系：
```python
ToolDef(
spec=...,
callable=answer,
prerequisites=[
"search", # 必须执行过 search
{"tool": "lookup", "match_arg": "id"} # 必须用相同 id 执行过 lookup
],
)
```
## 总结
Forge 的内置工具系统通过类型安全的 Pydantic 模型、清晰的前置依赖声明和完善的验证机制，为构建可靠的 LLM 工具调用工作流提供了坚实基础。系统设计强调：
1. **类型安全**：参数schema与执行逻辑强绑定
2. **可验证性**：每个工具调用都经过多层验证
3. **可追踪性**：完整记录工具执行历史和依赖满足状态
4. **可扩展性**：通过 `ToolDef` 轻松注册新工具
---
---
## Doramagic 踩坑日志
项目：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