函数调用 Agent 偏好优化流水线

DPO 数据长什么样？

LlamaFactory DPO trainer 直接消费的 JSONL，每行一个样本：

{
  "system": "你是一个智能AI助手，可以通过调用工具来帮助用户完成各种任务。\n\n工具调用格式：\n<function_call>\n{\n  \"name\": \"工具名称@版本\",\n  \"arguments\": {...}\n}\n</function_call>\n\n最终回答用 <final> 标签包裹。",
  "tools": "[{\"name\": \"get_weather@v1\", \"description\": \"...\", \"parameters\": {...}}, ...]",
  "messages": [
    {"role": "user", "content": "北京今天的天气怎么样？"}
  ],
  "chosen": "<function_call>\n{\"name\": \"get_weather@v1\", \"arguments\": {\"city\": \"北京\"}}\n</function_call>",
  "rejected": "我不知道，你可以查天气预报。"
}

工具名带 @v1 版本号——schema 演化后旧轨迹仍然可解析。<function_call> 包工具调用、<final> 包最终面向用户的回答。

工具库（backend/configs/tools_registry.json）

仓库 ship 了 10 个工具，覆盖时间 / 天气 / 计算 / 搜索 / 翻译 / 邮件 / 股价 / 提醒 / 新闻 / 货币 10 类：

{
  "tools": [
    {"name": "get_current_time", "version": "v1", "category": "time",
     "parameters": {"type": "object", "properties": {}, "required": []}},
    {"name": "get_weather", "version": "v1", "category": "weather",
     "parameters": {"properties": {"city": {"type": "string"}}, "required": ["city"]}},
    {"name": "calculate", "version": "v1", "category": "math",
     "parameters": {"properties": {"expression": {"type": "string"}}, "required": ["expression"]}},
    {"name": "web_search", "version": "v1", "category": "search",
     "parameters": {"properties": {"query": {"type": "string"}, "max_results": {"type": "integer", "default": 10}}, "required": ["query"]}},
    // ... 共 10 个
  ]
}

加新工具 = 改这份 JSON，不改代码。

6 模块后端流水线（backend/core/）

TaskGenerator → DataSynthesizer (chosen + rejected 并发) → LLM 自评 → Validator → ConcurrentEngine → Exporter

TaskGenerator · 76 个 task 模板 × 8 类

# backend/core/task_generator.py:73
TASK_TEMPLATES = {
    "天气查询": [
        "请帮我查询{city}的天气情况",
        "{city}今天天气怎么样？",
        "我想知道{city}最近的天气预报",
        # … 共 12 条
    ],
    "时间查询": [...10 条...],
    "计算":     [...10 条...],
    "搜索":     [...12 条...],
    "翻译":     [...10 条...],
    "货币转换": [...7 条...],
    "新闻获取": [...7 条...],
    "通用":     [...8 条...],
}

PARAMS = {
    "cities":         ["北京", "上海", "广州", "深圳", "杭州", ...],  # ×20
    "expressions":    ["1+1", "25*4", "100/5", "2^10", ...],        # ×15
    "search_queries": ["人工智能", "机器学习", "量子计算", ...],     # ×18
    "target_langs":   ["英语", "日语", "法语", "德语", ...],         # ×8
    # ...
}

MULTI_TURN_CONNECTORS = ["然后", "接着", "同时", "另外", "还有", "并且", "以及"]

multi_ratio 默认 0.3，30% 多轮对话用上面 7 个连接词把两个 task 拼起来。76 个模板 × 多个参数池 = 数千 unique queries。

DataSynthesizer · 5 步智能 rejected 策略

简单生成 chosen 和 rejected 不够——LLM 经常生成「假对比」（rejected 跟 chosen 差不多），DPO 训练用不了。所以加了一套 5 步策略：

# backend/core/data_synthesizer.py:89-200
async def synthesize_sample_with_smart_rejected(self, task: Task):
    # Step 1: 并发跑 chosen + rejected（省一轮 LLM 等待）
    chosen, rejected = await asyncio.gather(
        self._generate_chosen(task),
        self._generate_rejected(task)
    )

    # Step 2: 构造临时样本
    temp_sample = {..., "chosen": chosen, "rejected": rejected}

    # Step 3: LLM 自评 → quality_score (0-10) + similarity_score (0-100)
    llm_result = await self.llm_client.validate_and_correct(temp_sample)
    quality_score    = llm_result.get("quality_score", 7.0)
    similarity_score = llm_result.get("similarity_score", 50.0)

    # Step 4: 策略 1 · 如果 quality < 5 且 LLM 能修正 → 用 corrected_chosen 当新 chosen
    if quality_score < 5.0 and llm_result.get("corrected_chosen"):
        final_chosen = llm_result["corrected_chosen"]
        # 原 rejected 保留作真实错误案例

    # Step 5: 策略 2 · 如果 rejected 太像 chosen (>80%) → 用 temperature=1.2 重生成更差的
    if similarity_score > 80.0:
        final_rejected = await self.llm_client.generate_rejected_response(
            ..., temperature=1.2  # 更高的温度 → 更随机 → 差异更大
        )

    return {
        "task_id": task.task_id,
        "task_type": task.task_type,
        "system": task.system_prompt,
        "tools": task.to_dict()["tools"],
        "messages": messages,
        "chosen": final_chosen,
        "rejected": final_rejected,
        "quality_score": quality_score,
        "similarity_score": similarity_score
    }

LLM 自评 prompt（services/llm_client.py:269）

4 个评估轴：

validation_prompt = f"""
请检查以下DPO训练样本的质量：

用户问题：{sample['messages'][0]['content']}
可用工具：{sample['tools']}
Chosen回复：{sample['chosen']}
Rejected回复：{sample['rejected']}

请评估以下方面：
1. Chosen 回复质量（是否正确调用了工具，参数是否准确）
2. Rejected 回复质量（是否确实比 chosen 更差）
3. 两者差异度（差异是否明显，是否具有学习价值）
4. 格式规范性（是否符合 function_call 格式要求）

请以 JSON 格式回复：
{
  "is_valid": true/false,
  "quality_score": 8.5,      # 0-10
  "similarity_score": 25.0,  # 0-100，<80% 为佳
  "issues": ["问题1", "问题2"],
  "corrected_chosen": "...",
  "corrected_rejected": "..."
}

评分标准：
- quality_score: 9-10 极好 / 7-8 良好 / 5-6 一般 / <5 差
- similarity_score: <50% 优秀 / 50-80% 良好 / >80% 需要改进（rejected 不够差）
"""

JSON parse 兜底

DeepSeek 偶尔会用 ```json ... ``` 把 JSON 包起来。客户端先 strip 三种围栏再 json.loads()，失败就回退到默认分数，不让单次抖动掐死整批生成：

cleaned = response.strip()
if cleaned.startswith("```json"):
    cleaned = cleaned[7:]
if cleaned.startswith("```"):
    cleaned = cleaned[3:]
if cleaned.endswith("```"):
    cleaned = cleaned[:-3]

try:
    result = json.loads(cleaned)
except json.JSONDecodeError:
    return {"is_valid": True, "quality_score": 7.0,
            "similarity_score": 50.0, "issues": ["LLM返回格式错误"]}

ConcurrentEngine · 限流 + 指数退避

class ConcurrentEngine:
    def __init__(self, concurrency=10):
        self.semaphore = asyncio.Semaphore(concurrency)
        # 进度推送回调（接 WebSocket）
        self.progress_callbacks = []

ProgressStats 暴露 progress_percent / generation_rate (样本/秒) / validation_success_rate 给前端实时显示。

价值点

• 在工具调用决策层度量 Agent 质量，不只是文本流畅度
• 偏好数据骨架：轨迹采集 → 配对标注 → 评估准则——做的就是现代 DPO 真正需要的「数据基础设施」
• 关注生产细节：并发控制（Semaphore=10）、指数退避（2/4/8s 普通 → 3/9/27s 超时）、JSONL 严格性（LlamaFactory dataset_info.json columns 必须对齐）