Forge中的项目管理：构建LLM驱动的任务管理系统

Forge中的项目管理：构建LLM驱动的任务管理系统

【免费下载链接】forgeA Python framework for self-hosted LLM tool-calling and multi-step agentic workflows项目地址: https://gitcode.com/GitHub_Trending/forge54/forge

Forge是一个功能强大的Python框架，专为自托管LLM工具调用和多步骤代理工作流设计。它提供了完整的任务管理生命周期，包括系统提示、工具执行、上下文压缩和护栏机制，帮助开发者构建可靠的LLM驱动任务管理系统。

核心功能概览

Forge的任务管理系统建立在几个关键组件之上，这些组件协同工作，确保LLM能够可靠地完成复杂任务：

• WorkflowRunner：定义工具、选择后端、运行结构化代理循环，管理完整的任务生命周期
• 工具规范（ToolSpec）：描述LLM可以调用的Python函数及其参数
• 上下文管理：自动处理对话历史，防止上下文窗口溢出
• 护栏机制：包括步骤执行、先决条件检查、重试提示和错误恢复

三种集成模式

Forge提供了灵活的集成方式，可根据项目需求选择最合适的模式：

快速开始：构建你的第一个任务管理系统

安装与设置

首先，克隆Forge仓库并安装依赖：

git clone https://gitcode.com/GitHub_Trending/forge54/forge cd forge pip install .

创建简单的天气查询工作流

下面是一个完整的示例，展示如何创建一个天气查询任务管理系统：

from pydantic import BaseModel, Field from forge.core.workflow import Workflow, ToolDef, ToolSpec from forge.core.runner import WorkflowRunner from forge.clients.llamafile import LlamafileClient from forge.server import setup_backend, BudgetMode # 定义工具函数 def get_weather(city: str) -> str: return f"72°F and sunny in {city}" def report_weather(city: str, weather: str) -> str: return f"Weather report: {weather}" # 定义工具参数模型 class GetWeatherParams(BaseModel): city: str = Field(description="City name") class ReportWeatherParams(BaseModel): city: str = Field(description="City name") weather: str = Field(description="Weather description") # 创建工作流 workflow = Workflow( name="weather", description="Look up weather and report it.", tools={ "get_weather": ToolDef( spec=ToolSpec( name="get_weather", description="Get current weather for a city", parameters=GetWeatherParams, ), callable=get_weather, ), "report_weather": ToolDef( spec=ToolSpec( name="report_weather", description="Report the weather", parameters=ReportWeatherParams, ), callable=report_weather, ), }, required_steps=["get_weather"], # 必须调用的步骤 terminal_tool="report_weather", # 结束工作流的工具 ) # 设置后端和上下文管理器 server, ctx = await setup_backend( backend="llamaserver", gguf_path="path/to/Ministral-3-8B-Instruct-2512-Q8_0.gguf", budget_mode=BudgetMode.FORGE_FULL, ) # 创建客户端和运行器 client = LlamafileClient( gguf_path="path/to/Ministral-3-8B-Instruct-2512-Q8_0.gguf", mode="native", recommended_sampling=True, ) runner = WorkflowRunner(client=client, context_manager=ctx, stream=True) # 运行工作流 await runner.run(workflow, "What's the weather in Paris?") await server.stop()

任务管理的核心组件

工作流定义

工作流是Forge任务管理的基础，它定义了完成特定任务所需的工具、步骤和规则。通过Workflow类，你可以指定：

• 工具集合及其调用方式
• 必须执行的步骤（required_steps）
• 结束工作流的终端工具（terminal_tool）
• 工具之间的依赖关系（先决条件）

上下文管理

长时间运行的任务会累积大量对话历史，可能超出LLM的上下文窗口限制。Forge的上下文管理系统自动处理这一问题：

from forge.context import ContextManager, TieredCompact # 默认：分层压缩，保留最近2条消息 ctx = ContextManager(strategy=TieredCompact(keep_recent=2), budget_tokens=8192) # 无压缩（适用于不会超出限制的短工作流） ctx = ContextManager(strategy=NoCompact(), budget_tokens=8192)

上下文压缩分三个阶段进行：

1. 总结旧的工具结果，保持最近的消息完整
2. 压缩对话中间部分，保留系统提示和最近上下文
3. 激进压缩，只保留系统提示和最后几个对话

护栏机制

Forge内置了多种护栏机制，确保任务可靠执行：

多终端工具

工作流可以有多个有效的结束点，通过将终端工具指定为列表实现：

workflow = Workflow( ... terminal_tool=["set_ac", "no_action"], # 任一工具都可以结束工作流 )

高级任务管理功能

工具先决条件

工具可以声明依赖关系，确保在调用特定工具前必须先调用其他工具：

# 名称匹配：任何对read_file的调用都满足先决条件 ToolDef( spec=edit_spec, callable=edit_file, prerequisites=["read_file"], ) # 参数匹配：必须使用相同的path调用过read_file ToolDef( spec=edit_spec, callable=edit_file, prerequisites=[{"tool": "read_file", "match_arg": "path"}], )

多轮对话

通过on_message回调和initial_messages参数，Forge支持构建持续的多轮对话系统：

# 消费者管理对话历史 conversation: list[Message] = [] # 第一轮 - 正常运行，on_message收集所有内容 runner = WorkflowRunner(client=client, context_manager=ctx, on_message=lambda msg: conversation.append(msg)) await runner.run(workflow, "first question") # 后续轮次 - 用完整历史作为种子，添加新的用户消息 turn_messages: list[Message] = [] runner = WorkflowRunner(client=client, context_manager=ctx, on_message=lambda msg: turn_messages.append(msg)) seed = list(conversation) seed.append(Message(MessageRole.USER, "follow-up question", MessageMeta(MessageType.USER_INPUT))) await runner.run(workflow, "follow-up question", initial_messages=seed) conversation.extend(turn_messages)

优先级队列与任务抢占

SlotWorker提供了基于优先级的任务排队和自动抢占功能，适用于多任务共享单个推理资源的场景：

from forge import SlotWorker, WorkflowRunner # 创建运行器和工作器 runner = WorkflowRunner(client=client, context_manager=ctx) worker = SlotWorker(runner) await worker.start() # 定义优先级 USER = 0 # 最高优先级 ESCALATED = 1 ROUTINE = 2 # 最低优先级 # 提交任务 result = await worker.submit(calendar_wf, "what's on my schedule?", priority=USER) result = await worker.submit(ac_wf, "check temperature", priority=ROUTINE)

当高优先级任务提交时，正在运行的低优先级任务会被自动取消，确保重要任务优先执行。

最佳实践与性能优化

过滤瞬态消息

在长时间运行的会话中，过滤掉重试提示和错误消息等瞬态内容，保持上下文清洁：

from forge.core.messages import MessageType TRANSIENT_TYPES = { MessageType.RETRY_NUDGE, MessageType.STEP_NUDGE, MessageType.PREREQUISITE_NUDGE, MessageType.TEXT_RESPONSE, } def on_message(self, msg: Message) -> None: if msg.metadata.type not in TRANSIENT_TYPES: self.messages.append(msg)

采样参数优化

不同模型家族有不同的推荐温度、top_p和top_k参数。启用推荐采样可显著提高性能：

client = LlamafileClient( gguf_path="path/to/model.gguf", mode="native", recommended_sampling=True, # 启用推荐采样参数 )

取消机制

使用cancel_event实现协作式取消，允许在任务运行时中断：

import asyncio cancel = asyncio.Event() # 在另一个协程或回调中触发取消 cancel.set() try: result = await runner.run(workflow, "task", cancel_event=cancel) except WorkflowCancelledError as e: print(f"Cancelled at iteration {e.iteration}") print(f"Completed steps: {e.completed_steps}")

总结

Forge提供了一个强大而灵活的框架，用于构建LLM驱动的任务管理系统。通过其直观的API和内置的可靠性特性，开发者可以专注于业务逻辑而非基础设施细节。无论是简单的单步工具调用还是复杂的多步骤工作流，Forge都能提供一致且可靠的执行环境。

要深入了解更多高级功能，请参阅官方文档：

• 用户指南
• 架构文档
• 模型指南
• 评估指南

【免费下载链接】forgeA Python framework for self-hosted LLM tool-calling and multi-step agentic workflows项目地址: https://gitcode.com/GitHub_Trending/forge54/forge