Forge中的项目管理构建LLM驱动的任务管理系统【免费下载链接】forgeA Python framework for self-hosted LLM tool-calling and multi-step agentic workflows项目地址: https://gitcode.com/GitHub_Trending/forge54/forgeForge是一个功能强大的Python框架专为自托管LLM工具调用和多步骤代理工作流设计。它提供了完整的任务管理生命周期包括系统提示、工具执行、上下文压缩和护栏机制帮助开发者构建可靠的LLM驱动任务管理系统。核心功能概览Forge的任务管理系统建立在几个关键组件之上这些组件协同工作确保LLM能够可靠地完成复杂任务WorkflowRunner定义工具、选择后端、运行结构化代理循环管理完整的任务生命周期工具规范ToolSpec描述LLM可以调用的Python函数及其参数上下文管理自动处理对话历史防止上下文窗口溢出护栏机制包括步骤执行、先决条件检查、重试提示和错误恢复三种集成模式Forge提供了灵活的集成方式可根据项目需求选择最合适的模式特性WorkflowRunner代理模式中间件模式验证和救援解析是是是重试提示是是是响应工具调用者添加自动注入调用者添加步骤执行是否是调用者连接先决条件是否是调用者连接上下文压缩是是调用者连接ContextManager快速开始构建你的第一个任务管理系统安装与设置首先克隆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 f72°F and sunny in {city} def report_weather(city: str, weather: str) - str: return fWeather report: {weather} # 定义工具参数模型 class GetWeatherParams(BaseModel): city: str Field(descriptionCity name) class ReportWeatherParams(BaseModel): city: str Field(descriptionCity name) weather: str Field(descriptionWeather description) # 创建工作流 workflow Workflow( nameweather, descriptionLook up weather and report it., tools{ get_weather: ToolDef( specToolSpec( nameget_weather, descriptionGet current weather for a city, parametersGetWeatherParams, ), callableget_weather, ), report_weather: ToolDef( specToolSpec( namereport_weather, descriptionReport the weather, parametersReportWeatherParams, ), callablereport_weather, ), }, required_steps[get_weather], # 必须调用的步骤 terminal_toolreport_weather, # 结束工作流的工具 ) # 设置后端和上下文管理器 server, ctx await setup_backend( backendllamaserver, gguf_pathpath/to/Ministral-3-8B-Instruct-2512-Q8_0.gguf, budget_modeBudgetMode.FORGE_FULL, ) # 创建客户端和运行器 client LlamafileClient( gguf_pathpath/to/Ministral-3-8B-Instruct-2512-Q8_0.gguf, modenative, recommended_samplingTrue, ) runner WorkflowRunner(clientclient, context_managerctx, streamTrue) # 运行工作流 await runner.run(workflow, Whats the weather in Paris?) await server.stop()任务管理的核心组件工作流定义工作流是Forge任务管理的基础它定义了完成特定任务所需的工具、步骤和规则。通过Workflow类你可以指定工具集合及其调用方式必须执行的步骤required_steps结束工作流的终端工具terminal_tool工具之间的依赖关系先决条件上下文管理长时间运行的任务会累积大量对话历史可能超出LLM的上下文窗口限制。Forge的上下文管理系统自动处理这一问题from forge.context import ContextManager, TieredCompact # 默认分层压缩保留最近2条消息 ctx ContextManager(strategyTieredCompact(keep_recent2), budget_tokens8192) # 无压缩适用于不会超出限制的短工作流 ctx ContextManager(strategyNoCompact(), budget_tokens8192)上下文压缩分三个阶段进行总结旧的工具结果保持最近的消息完整压缩对话中间部分保留系统提示和最近上下文激进压缩只保留系统提示和最后几个对话护栏机制Forge内置了多种护栏机制确保任务可靠执行护栏功能步骤执行验证在终端工具执行前是否调用了所有必需的工具先决条件强制工具之间的依赖关系例如必须先读取再编辑重试提示当工具调用验证失败时提示LLM重试救援循环从LLM的文本输出中恢复格式错误的工具调用错误恢复工具执行错误后重新提示而不是崩溃压缩防止长对话中的上下文溢出多终端工具工作流可以有多个有效的结束点通过将终端工具指定为列表实现workflow Workflow( ... terminal_tool[set_ac, no_action], # 任一工具都可以结束工作流 )高级任务管理功能工具先决条件工具可以声明依赖关系确保在调用特定工具前必须先调用其他工具# 名称匹配任何对read_file的调用都满足先决条件 ToolDef( specedit_spec, callableedit_file, prerequisites[read_file], ) # 参数匹配必须使用相同的path调用过read_file ToolDef( specedit_spec, callableedit_file, prerequisites[{tool: read_file, match_arg: path}], )多轮对话通过on_message回调和initial_messages参数Forge支持构建持续的多轮对话系统# 消费者管理对话历史 conversation: list[Message] [] # 第一轮 - 正常运行on_message收集所有内容 runner WorkflowRunner(clientclient, context_managerctx, on_messagelambda msg: conversation.append(msg)) await runner.run(workflow, first question) # 后续轮次 - 用完整历史作为种子添加新的用户消息 turn_messages: list[Message] [] runner WorkflowRunner(clientclient, context_managerctx, on_messagelambda 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_messagesseed) conversation.extend(turn_messages)优先级队列与任务抢占SlotWorker提供了基于优先级的任务排队和自动抢占功能适用于多任务共享单个推理资源的场景from forge import SlotWorker, WorkflowRunner # 创建运行器和工作器 runner WorkflowRunner(clientclient, context_managerctx) worker SlotWorker(runner) await worker.start() # 定义优先级 USER 0 # 最高优先级 ESCALATED 1 ROUTINE 2 # 最低优先级 # 提交任务 result await worker.submit(calendar_wf, whats on my schedule?, priorityUSER) result await worker.submit(ac_wf, check temperature, priorityROUTINE)当高优先级任务提交时正在运行的低优先级任务会被自动取消确保重要任务优先执行。最佳实践与性能优化过滤瞬态消息在长时间运行的会话中过滤掉重试提示和错误消息等瞬态内容保持上下文清洁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_pathpath/to/model.gguf, modenative, recommended_samplingTrue, # 启用推荐采样参数 )取消机制使用cancel_event实现协作式取消允许在任务运行时中断import asyncio cancel asyncio.Event() # 在另一个协程或回调中触发取消 cancel.set() try: result await runner.run(workflow, task, cancel_eventcancel) except WorkflowCancelledError as e: print(fCancelled at iteration {e.iteration}) print(fCompleted 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考