面向长时程任务的智能体系统需要具备规划、正确使用工具以及逐步执行的能力。大多数现代智能体系统依赖推理，每次组件调用都从零开始，且缺乏事先训练，这会显著增加长时程任务中任一步骤出现错误规划或错误工具调用的概率。GRPO算法能够持续训练智能体在长任务中进行规划与正确执行。一个典型的基于GRPO的智能体训练系统大致如下所示……

GRPO如何影响智能体训练：

1. 基于组的评估：GRPO会针对同一查询评估多条轨迹，让智能体比较不同策略，而不是依赖于单步奖励。
2. 相对优势学习：成功的轨迹会相对于组平均水平被强化，从而提升正确规划与执行的概率。
3. 错误抑制：较差的轨迹会收到负向优势信号，从而减少幻觉与错误工具使用。
4. 迭代式改进：通过反复生成轨迹，智能体将持续改进，更可靠地应对长时程任务规划。
5. 跨子智能体协同：在组上下文中训练，GRPO有助于多个子智能体对齐其行动，从而提升整个多智能体系统的表现。

目录

• GRPO算法在智能体系统中的角色
• 智能体数据预处理
• 构建多智能体架构
  • 定义智能体思维
  • 创建工具集
  • Solver、Planner、Executor与Verifier
  • 编排智能体循环
• 分析查询中的幻觉
• 智能体GRPO算法实现
  • 初始化策略模型（QLoRA & PEFT）
  • 用于轨迹生成的智能体系统包装器
  • 使用GPT-4o进行奖励建模
  • 创建优势与PPO损失
• 运行优化后的规划智能体

GRPO算法在智能体系统中的角色

几乎所有强化学习算法都基于奖励机制：智能体在环境中采取动作，并根据动作获得奖励或惩罚；目标是在时间维度上最大化累计奖励。

然而在多智能体系统中，多个智能体相互作用并与环境交互，传统强化学习在协调与优化这些动作方面常常力不从心。

GRPO正是为了解决多智能体场景下传统强化学习面临的一些挑战。在基于智能体的GRPO中，不再对每个单独动作逐一打分，而是让智能体对同一个问题进行多次尝试（形成一个“组”），比较这些结果，并强化那些优于组平均水平的策略。

我们用一个例子来理解（也是本文实现的核心）：

1. 系统接收一个复杂查询，例如：“计算12的平方，然后用Wikipedia找出该年份发生的主要历史事件。”
2. 规划器策略针对这个单一查询生成一组不同的轨迹（例如4次尝试）。由于模型使用非零温度，它会探索不同策略：某条轨迹可能正确用Python算出12^2=144，另一条可能猜错数值，还有一条可能不检索就幻觉出历史事件。
3. 外部评判器将每条轨迹的最终答案与真实答案进行评估。成功算出144并找到正确历史事件的轨迹获得奖励=1.0，而猜错、工具执行失败或幻觉答案的轨迹获得奖励=0.0。
4. 算法通过与组平均表现比较计算相对优势。如果组均值是0.25，则成功的轨迹（1.0）获得高正优势（+0.75），失败的（0.0）获得负优势（-0.25）。
5. 策略模型根据这些优势信号更新权重。由于成功的轨迹超过了组平均，它大幅提升成功轨迹所用规划步骤的概率，同时抑制失败策略。

我们将在多智能体系统中编码实现上述GRPO算法，以改进规划阶段并减少幻觉与跑题结果。

智能体数据预处理

多智能体系统通常由不同目的的子智能体协作完成任务。例如，可以有网络搜索智能体、规划智能体、任务执行智能体等。

在推理时，这些智能体的表现高度依赖“规划阶段”。如果规划不佳，每轮迭代后智能体可能偏离正轨，产出无关结果，甚至让整个管线“幻觉化”。

本文将模拟一个真实世界的多智能体系统，包含多类子智能体，其中有些子智能体目的相同但方法不同。这样我们就能实测GRPO如何降低幻觉与跑题风险。

我们将使用两个Hugging Face数据集来准备训练数据：

1. DeepMath-103K：包含10万余道数学题及其解答。可帮助智能体在代用户执行计算时学会规划，从而在规划阶段形成结构化的逐步推理。
2. Natural Questions：包含来自真实用户的提问，有助于提升智能体对全流程的规划能力。

首先导入必要依赖并为训练/验证数据创建输出目录。

“`python

Standard library imports for interacting with the operating system and handling JSON data.

import os
import json

Core data science libraries for data manipulation and numerical computation.

import pandas as pd
import numpy as np

Hugging Face library for dataset loading and processing.

from datasets import load_dataset, concatenate_datasets, Dataset

Utility for displaying progress bars, making long-running operations more informative.

from tqdm import tqdm
“`

创建训练与验证数据目录：

“`python

Define the path for the training data output directory.

train_output_dir = “./data/train”

Define the path for the validation data output directory.

val_output_dir = “./data/val”
“`

训练目录将存放 DeepMath 与 NQ 合并后的训练集，验证目录将存放 AIME 2024 验证集。

我们需要将两个数据集合并成一个训练集，这样智能体系统就能从多样化问题中学习，更好地规划与执行。

先处理 DeepMath-103K，载入并查看结构：

“`python
print(“n=== Loading DeepMath-103K ===”)

math_dataset = load_dataset(
“zwhe99/DeepMath-103K”,
split=”train”
)
“`

在处理前，先检查列名、样本量以及单条记录：

python
print("Columns:", math_dataset.column_names)
print("Total samples:", len(math_dataset))

输出示例：
Columns: ['question', 'final_answer', 'difficulty', 'topic', 'r1_solution_1', 'r1_solution_2', 'r1_solution_3']
Total samples: 103022

可见每题有 question 与三个不同解法以及 final_answer（将作为训练的 ground truth）。

查看一条样本：

python
sample = math_dataset[0]
...
print(json.dumps(truncated_sample, indent=2))

输出示例省略。这里的 r1_solution_* 我们不会用于训练，仅使用 question 与 final_answer，因为我们的智能体会执行代码并尝试得到最终答案。

将每条记录转为标准格式，便于后续与其他数据集合并。目标 schema：
* id：唯一标识
* question：问题文本
* chain：Chain-of-Thought 或推理步骤占位（暂留空）
* result：最终答案
* source：原始数据来源
* extra_info：保留原始元数据

python
print("n=== Processing MathHard ===")
...

处理完成后验证样本数量与结构：

python
print("Processed math samples:", len(math_rows))
print(json.dumps(math_rows[0], indent=2))

将列表转为 Hugging Face Dataset：

python
ds_math = Dataset.from_pandas(
pd.DataFrame(math_rows),
preserve_index=False
)

接着处理 Natural Questions（NQ）。载入并查看一条样本：

“`python
nq_dataset = load_dataset(
“RUC-NLPIR/FlashRAG_datasets”,
“nq”,
split=”train”
)

print(json.dumps(nq_dataset[0], indent=2))
“`

可以看到 question 是用户查询，golden_answers 是答案列表。对 NQ 做清洗：
1. 统一给问题补上问号。
2. 处理 golden_answers 的各种格式（list、ndarray、string、number…），提取为字符串。
3. 多个答案以分号连接为单字符串。

python
print("n=== Processing NQ ===")
...

验证处理结果并转为 Dataset：

“`python
print(json.dumps(nq_rows[0], indent=2))

ds_nq = Dataset.from_pandas(
pd.DataFrame(nq_rows),
preserve_index=False
)
“`

合并两个数据集，打乱并重排 id：

“`python
combined = concatenate_datasets([ds_nq, ds_math])
print(“Combined size:”, len(combined))

combined = combined.shuffle(seed=42)
combined = combined.map(
lambda _, idx: {“id”: idx},
with_indices=True
)
“`

保存为 Parquet 格式：

python
output_path = os.path.join(train_output_dir, "combined_train.parquet")
combined.to_parquet(output_path)

核对样本总数：

python
train_count = len(combined)
print(f"nTotal train samples: {train_count}")

我们成功将 DeepMath-103K 与 NQ 合并为一个包含 182,190 条样本的训练集，供后续训练使用。

构建 Multi-Agentic 架构

Agentic 工作流（多智能体系统）是指由一系列专门化组件（智能体）协同解决问题的框架。与依赖单次大语言模型调用不同，它将复杂问题拆解为多个可管理的阶段：

Multi-Agent System (Created by Fareed Khan)

1. 规划：分析初始查询，制定高层计划或决定下一步动作。
2. 工具使用：选择并调用专用工具（如代码解释器、网络搜索、数据库查询等）。
3. 执行：生成精确的工具调用命令。
4. 观察与反思：观察工具执行结果，判断是否达成目标或需要更多步骤。
5. 迭代：循环往复，直至验证器判定问题得到完整解答。
6. 综合：汇总整个过程的信息，给出最终答案。

本文的目标是改进贯穿全流程的“规划阶段”，以降低模型幻觉与跑题的风险。为了微调规划阶段，我们需要可训练的模型权重，因此本项目采用高吞吐、低延迟的 vLLM 服务。

实验环境使用 1 块 A100 80GB GPU，规划模型选用 Qwen/Qwen2.5-7B-Instruct；其他环节可按需更换模型。

安装并启动 vLLM 服务：
bash
vllm serve Qwen/Qwen2.5-7B-Instruct
--api-key a-secret-key
--port 8000
--max-model-len 8192

安装项目所需依赖：
bash
!pip install -q openai pydantic tenacity beautifulsoup4 requests wikipedia google-genai numpy json_repair

导入模块并配置 vLLM 与外部服务的连接：
“`python

Standard library imports

…
import json_repair
“`

配置 vLLM 服务端点与模型，并设置外部 API 密钥：
python
VLLM_BASE_URL = "http://localhost:8000"
VLLM_API_KEY = "a-secret-key"
MODEL_NAME = "Qwen/Qwen2.5-7B-Instruct"
os.environ['OPENAI_API_KEY'] = 'YOUR_OPENAI_API_KEY_HERE'
os.environ['GOOGLE_API_KEY'] = 'YOUR_GOOGLE_API_KEY_HERE'

为了与 LLM 进行交互，我们创建一个包装器类。这有利于实现抽象、增强健壮性（如自动重试）以及支持结构化输出等功能。

定义智能体思维组件

首先定义 EngineLM 抽象基类：

Agent thoughts (Created by Fareed Khan)

python
class EngineLM(ABC):
...

接着实现连接 vLLM 的 ChatVLLM 类，引入结构化输出（基于 Pydantic Schema）、JSON 修复等增强功能：
python
class ChatVLLM(EngineLM):
...

随后定义工厂函数并测试与 vLLM 服务的连接：
python
def create_llm_engine(...): ...
...
test_response = test_engine.generate("Ping")

若输出为 “Pong”，则说明连接正常。

为规范多智能体间的交互，我们定义一系列结构化输出模型（基于 Pydantic）：
* QueryAnalysis：初始查询分析
* NextStep：每轮规划器的下一步行动决策
* ToolCommand：执行器生成的具体指令
* MemoryVerification：验证器的停止/继续判断
* Select_Relevant_Queries：供 Wikipedia 工具内部使用

python
class QueryAnalysis(BaseModel): ...
class NextStep(BaseModel): ...
class ToolCommand(BaseModel): ...
class MemoryVerification(BaseModel): ...
class Select_Relevant_Queries(BaseModel): ...

实现 Memory 类作为行为日志，用于保存每一步的工具、子目标、命令与结果；同时提供一个序列化与截断的工具方法，以避免上下文过长：
python
class Memory: ...
def make_json_serializable_truncated(...): ...

创建工具集

设计 BaseTool 抽象类，以统一工具的元数据定义与执行接口：

Tool Definitions (Created by Fareed Khan)

python
class BaseTool(ABC):
...

实现若干具体工具：
* Base_Generator_Tool：通用 LLM 直接答复工具。
* Python_Coder_Tool：生成并在沙箱中执行 Python 代码（包含超时、输出捕获等安全机制）。
* Google_Search_Tool：基于 Google Gemini 的联网搜索工具（若无 API 密钥则使用模拟模式）。
* Web_Search_Tool：面向单一 URL 的 RAG 流水线（包含抓取、切块、嵌入、召回、综合等步骤）。
* Wikipedia_Search_Tool：组合工具，先查询 Wikipedia 条目，筛选相关条目，再使用 Web_RAG 搜索并聚合结果。

上述工具的具体实现均包含在项目代码中（为保持技术一致性与可复现性，此处不逐条翻译，保留原始英文代码与注释）。

求解器、规划器、执行器与验证器

系统通过 Initializer 初始化所有工具并缓存其元数据；再由 Solver 统一编排整个智能体工作流，分别为规划器（主/固定）、验证器、执行器准备 LLM 引擎与记忆模块。

核心步骤函数如下：
* solve_step_0_analyze：对查询进行初始分析（使用 planner_fixed_engine）。
* solve_step_1_plan：基于查询、分析结果、可用工具与记忆，选择最优工具与子目标（使用 planner_main_engine）。
* solve_step_2_generate_command：生成可执行的 Python 命令（使用 executor_engine）。
* solve_step_3_execute_command：执行命令，并将结果写入记忆。
* solve_step_4_verify：验证器判断是否应停止（使用 verifier_engine）。
* solve_final_step_synthesize：任务完成后，综合输出最终答案（使用 planner_fixed_engine）。

主循环 solve() 将以上步骤按顺序组织执行，直到达到预设的最大步数或验证器发出停止信号。

编排 Agentic 循环

上述步骤的提示词（prompts）在代码中均有详细说明，包括需要提供给 LLM 的上下文、输出结构及严格的格式约束。

分析查询中的幻觉

我们运行一个复杂查询，涉及多工具多轮调用，并观察 agent 的行为与失误点。示例日志（节选）显示：

• 先用搜索工具再算数学，导致基于错误年份搜索历史事件（规划顺序错误）
• 过早做出假设（例如 120 AD）
• 将“未命中”误判为“该年没有大事件”
• 最后使用通用生成工具给出未经验证的历史描述

这些问题表明，即便有多工具与验证环节，若规划阶段不稳健，仍可能出现严重错误。接下来我们用 GRPO 训练改进其规划与推理能力。

Agentic GRPO 算法实现

我们将实现 Flow-GRPO 训练环节：生成 trajectories、计算 reward，并基于相对优势更新策略模型。

GRPO Training (Created by Fareed Khan)

导入依赖、配置训练超参与模型路径（使用 transformers + peft 进行 QLoRA/LoRA 训练）：

python
from transformers import ...
from peft import ...
@dataclass
class TrainingConfig: ...

关键点：

• base_model_name（可训练的 Policy/Planner）：较小模型（如 Qwen2-1.5B-Instruct），利于效率训练
• fixed_model_name：更强大的固定模型（如 Qwen2.5-7B-Instruct）用于执行/验证
• reward_model_name（Judge）：高质量评审模型（如 gpt-4o）

解释部分超参含义（train_batch_size、rollout_n、ppo_clip_eps、kl_coef、max_grad_norm 等），详见原文。

初始化策略模型（QLoRA & PEFT）

加载 tokenizer、以 4-bit 量化加载 base 模型（QLoRA），再用 LoRA 挂载可训练适配器，仅训练极少参数比例以保证效率与稳定性。初始化固定 LLM（执行/验证）与 reward LLM（Judge）。

python
tokenizer = AutoTokenizer.from_pretrained(...)
bnb_config = BitsAndBytesConfig(...)
policy_model = AutoModelForCausalLM.from_pretrained(..., quantization_config=bnb_config, ...)
policy_model = prepare_model_for_kbit_training(policy_model)
peft_config = LoraConfig(...)
policy_model = get_peft_model(policy_model, peft_config)
ref_model = policy_model
fixed_llm = create_llm_engine(...)
reward_llm = create_llm_engine(...)

输出显示仅约 1.1% 参数可训练。

用于 Rollouts 的 Agentic System 包装器

AgenticSystem 封装训练 rollout 所需环境：

• 工具加载与管理
• 基于 Query+Memory 构建 Planner 的状态提示词
• 由 Policy 生成动作与计算对应 log 概率（log_probs），这是 PPO/GRPO 的基础

python
class AgenticSystem:
def _load_tools(...): ...
def build_planner_prompt(...): ...
@torch.no_grad()
def generate_planner_action(...): ...
def run_executor_verifier(...): ...
def run_trajectory(...): ...

• generate_planner_action：用采样（非零温度、top-p）生成多样化动作，并从输出 logits 计算对应 token 的 log_probs。
• run_executor_verifier：用固定 LLM 生成工具命令并执行，再由固定 LLM 验证是否停止。
• run_trajectory：组织完整多轮 rollouts，收集每一 turn 的 TurnData（包含 prompt、action、ids、log_probs），并最终得到 final_answer。

使用 GPT-4o 进行 Reward 建模

引入外部 Judge（如 GPT-4o）判断最终答案是否正确，输出二元 reward（1.0/0.0）。若 Judge 失败则回退为字符串匹配。

python
def compute_reward(query, ground_truth, final_answer) -> float: ...

创建 Advantages 与 PPO Loss

实现 PPO/GRPO 损失函数：

• 计算新/参考策略的 log_probs
• 基于采样时旧策略的 log_prob 构建重要性比率 ratio
• 引入剪切的 surrogate loss（防止策略更新过大）
• 加 KL 正则防止策略漂移

python
def compute_ppo_loss(policy_model, ref_model, tokenizer, trajectories, advantages):
...

载入训练数据并用 DataLoader 迭代：

python
full_dataset = load_dataset("parquet", data_files=config.data_file, split="train")
train_dataloader = DataLoader(..., batch_size=config.train_batch_size, shuffle=True)

组装训练流程：

• 对每个 batch 的每条 query 进行 N=rollout_n 次 group rollouts
• 计算该组 rewards 的均值/标准差，并得到每条 trajectory 的 Advantage（相对优势）
• 用 PPO/GRPO loss 更新策略模型（支持梯度累积与学习率调度）

“`python
agent_system = AgenticSystem(…)
optimizer = AdamW(…)
scheduler = get_scheduler(…)

for epoch in …:
for step, batch in …:
for query in batch:
group_trajectories, group_rewards = [], []
for _ in range(config.rollout_n):
trajectory, final_answer = agent_system.run_trajectory(query)
reward = compute_reward(query, ground_truth, final_answer)
…
advantages = (rewards – mean) / std
policy_loss, kl_div = compute_ppo_loss(…)
loss = policy_loss + config.kl_coef * kl_div
…
“`

训练日志显示损失值逐步降低，平均奖励稳步上升。以上示例仅展示了一个训练周期内的部分步骤，实际应用中需要更长时间的训练才能获得可验证的性能提升。

运行优化后的规划智能体

训练完成后，加载模型检查点并使用 vLLM 启动推理服务：

bash
vllm serve ./agentflow_checkpoints/epoch_1
--served-model-name Qwen/Qwen2-1.5B-Instruct
--quantization bitsandbytes
--enable-lora
--port 8000

使用训练后的策略模型处理相同的复杂查询，可以观察到其在规划、工具切换以及最终答案质量方面的显著改善。示例运行结果（节选）表明：

• 规划顺序更合理：例如，优先执行数学计算，再进行历史信息检索。
• 具备容错能力：当 Wikipedia_RAG 或搜索工具返回错误或未命中时，智能体能够执行备用路径（fallback）。
• 验证更稳健：验证器对流程逻辑与可行性给出了更为可靠的判断。

尽管某些工具调用可能遇到 422 错误，但智能体仍能通过利用其他工具和综合推理步骤，生成更合理的最终答案。这证明了其规划阶段的稳健性以及工具切换策略的有效性已得到提升。

关注“鲸栖”小程序，掌握最新AI资讯