到目前为止，已经有了：

• 基于向量检索（含多模态）的 RAG；
• 拆分良好的多智能体与中台；
• 图式工作流（Graph）来编排节点；
• 完整的调用监控与 Trace；
• 工作流级的 A/B 测试与效果评估。

但现在还有一个问题没有解决：

「知道哪种配置更好」是一回事，
「让系统自己发现更优配置并逐步采用」是另一回事。

这篇就来做这件事——在你现有框架上，再加一层自动化调优环，大致实现：

1. 自动从线上日志中发现“表现不佳”的场景；
2. 自动生成若干「候选配置」（比如不同参数、不同节点组合）；
3. 自动进行小流量试验与评估；
4. 自动更新「当前默认配置」或给出建议。

整篇从工程视角出发，你可以理解为：
在现有工作流引擎外面，再包一圈“控制器（Controller）”。​

---

一、整体思路：在执行引擎外再加一层「控制环」

可以把你现在的大模型系统抽象成两层：

• 执行层（Execution Layer）​：

  • Graph 工作流引擎
  • 各种 Agent / 工具 / RAG 模块
  • 负责“按当前配置跑起来”

• 控制层（Control Layer）​：

  • 读取监控指标和 Trace
  • 设计并执行 A/B 测试
  • 生成、筛选新配置
  • 决定何时切换配置

用文字版架构表示：

               ┌────────────────────────┐
               │      控制层 Controller  │
               │  - 指标聚合             │
               │  - 生成候选配置         │
               │  - 试验与选择           │
               └─────────┬──────────────┘
                         │
           配置（workflow_config_X.json）
                         │
               ┌─────────▼──────────────┐
               │      执行层 Engine      │
               │  - Graph 工作流         │
               │  - 多智能体/工具调用     │
               │  - RAG 检索             │
               └─────────┬──────────────┘
                         │
              Trace / 日志 / 反馈 / 评分

我们要做的是：实现那个「控制层」。

---

二、先把“可调参数”和“可替换模块”列出来

自动化调优之前，先要明确什么是可调的，否则控制层无从下手。

2.1 工作流里的典型可调项

1. 流程结构

   • 是否开启某个节点（例如是否使用工具 Agent）
   • 分支条件的阈值（例如 Router 的分类信心阈值）

2. LLM 相关参数

   • 温度（temperature）
   • 最大输出长度（max_tokens）
   • top_p / top_k 等采样策略

3. RAG 相关参数

   • top_k（检索条数）
   • 是否启用多轮 Query 重写
   • Embedding 模型版本

4. 节点策略

   • 某 Agent 是否启用“草稿 + 复核”模式
   • 是否允许某些问题直接走「快捷路径」

你可以先在配置文件中做一个结构化定义，例如：

workflow_name: "qa_pipeline"

llm:
  model: "gpt-4.1-mini"
  temperature: 0.2
  max_tokens: 512

rag:
  top_k: 4
  rerank: false

nodes:
  qa_agent:
    enabled: true
  qa_with_tool_agent:
    enabled: false
  summary_agent:
    enabled: true

router:
  summary_keywords: ["总结", "概括", "归纳"]

控制层的任务，就是在这个参数空间里自动搜索更优组合。

---

三、控制层核心流程：观察 → 生成候选 → 实验 → 选择

可以把控制器逻辑拆成四步：

1. 观察（Observe）​：从 Trace / 日志中聚合指标，识别“表现不佳”的场景；
2. 生成候选（Generate）​：针对这些场景，生成几组新的配置候选；
3. 实验（Experiment）​：用 A/B 测试框架给候选配置分配小流量，收集结果；
4. 选择（Select）​：对比指标，决定是否把某个候选提升为新的默认配置。

下面逐步实现每一步的骨架。

---

四、观察：从日志中识别“问题热点”

可以复用你之前的 TraceStore，在此基础上写一个简单分析器：

from collections import defaultdict
from datetime import datetime, timedelta

class PerformanceAnalyzer:
    def __init__(self, db_conn):
        self.conn = db_conn

    def get_hot_problems(self, since_minutes: int = 60, min_calls: int = 50):
        """
        找出过去 since_minutes 分钟里，表现明显不佳的组合：
        - 某个 workflow / 场景 / 节点 的平均耗时 或 错误率 明显偏高
        """
        cursor = self.conn.cursor()
        since_ts = (datetime.utcnow() - timedelta(minutes=since_minutes)).isoformat()

        cursor.execute("""
        SELECT 
          workflow,
          node,
          COUNT(*) as cnt,
          AVG(duration_ms) as avg_ms,
          SUM(CASE WHEN status LIKE 'error%' THEN 1 ELSE 0 END) * 1.0 / COUNT(*) as error_rate
        FROM traces
        WHERE timestamp >= ?
        GROUP BY workflow, node
        """, (since_ts,))

        problems = []
        for row in cursor.fetchall():
            workflow, node, cnt, avg_ms, err = row
            if cnt < min_calls:
                continue
            # 简化：如果平均耗时>3000ms 或 错误率>5%，认为是“热点问题”
            if avg_ms > 3000 or err > 0.05:
                problems.append({
                    "workflow": workflow,
                    "node": node,
                    "count": cnt,
                    "avg_ms": avg_ms,
                    "error_rate": err
                })
        return problems

控制层可以定时（比如每 10 分钟）跑一次 get_hot_problems，决定要不要针对这些节点发起调优。

---

五、生成候选配置：策略 + 搜索空间

针对某个 hotspot（如 qa_pipeline 下的 qa_agent），我们可以预设一些「可尝试的调优方向」：

• 增大或减小 rag.top_k；
• 开启/关闭 qa_with_tool_agent；
• 微调 llm.temperature 等。

可以用一个简单的「配置变异器」来生成候选：

import copy
import random

class ConfigMutator:
    def __init__(self, base_config: dict):
        self.base = base_config

    def mutate_temperature(self, delta: float = 0.1):
        cfg = copy.deepcopy(self.base)
        t = cfg["llm"]["temperature"]
        t_new = max(0.0, min(1.0, t + random.choice([-delta, delta])))
        cfg["llm"]["temperature"] = t_new
        return cfg

    def mutate_rag_topk(self, choices=(3, 4, 5, 6)):
        cfg = copy.deepcopy(self.base)
        cfg["rag"]["top_k"] = random.choice(choices)
        return cfg

    def toggle_tool_agent(self):
        cfg = copy.deepcopy(self.base)
        # 在 QA 和 QA_with_tool 之间做切换
        use_tool = cfg["nodes"].get("qa_with_tool_agent", {}).get("enabled", False)
        cfg["nodes"]["qa_with_tool_agent"]["enabled"] = not use_tool
        cfg["nodes"]["qa_agent"]["enabled"] = use_tool  # 保证二者互斥
        return cfg

    def generate_candidates(self, n: int = 3) -> list[dict]:
        """生成 n 个候选配置"""
        candidates = []
        ops = [self.mutate_temperature, self.mutate_rag_topk, self.toggle_tool_agent]
        for _ in range(n):
            op = random.choice(ops)
            candidates.append(op())
        return candidates

你可以根据场景细化变异策略，比如只针对某个节点做特定改动。

---

六、实验：把候选配置接入 A/B 测试框架

一旦生成了候选配置，需要为它们创建对应的工作流实例，并放入 AB Runner 中。

6.1 把配置映射成 Graph

假设你已经有一个函数 build_graph_from_config(config)，可以根据配置生成一张 Graph：

def build_graph_from_config(config: dict) -> Graph:
    g = Graph(name=config.get("workflow_name", "qa_pipeline"))

    # 根据 config["nodes"] 动态添加节点
    # 这里只演示结构，具体实现按你的 Agent 定义来
    if config["nodes"]["qa_agent"]["enabled"]:
        g.add_node(QAAgent(llm_client), is_start=True)
        # ...
    if config["nodes"]["qa_with_tool_agent"]["enabled"]:
        g.add_node(QAWithToolAgent(llm_client), is_start=True)
        # ...
    # 添加 Retrieve / Summary 等节点和边
    return g

6.2 注册到 ABTestWorkflowRunner 中

控制层可以为每个候选配置生成一个唯一名字，例如：

• qa_pipeline_base
• qa_pipeline_candidate_1
• qa_pipeline_candidate_2

然后在 AB Runner 中增加这些 workflow，并为它们分配一定流量（比如每个候选 5%，基线 85%）：

class WeightedABRunner:
    def __init__(self, workflows: Dict[str, Graph], weights: Dict[str, int]):
        """
        :param workflows: {name: graph}
        :param weights: {name: weight}
        """
        self.workflows = workflows
        self.weights = weights
        self.total_weight = sum(weights.values())

    def choose_workflow(self, request_id: str) -> Graph:
        h = int(hashlib.md5(request_id.encode()).hexdigest(), 16) % self.total_weight
        cur = 0
        for name, w in self.weights.items():
            cur += w
            if h < cur:
                return self.workflows[name]
        # fallback
        return list(self.workflows.values())[0]

控制层负责：

• 把候选配置转换为 Graph；
• 放到 workflows 里；
• 设置权重（流量占比）；

然后在一段时间内（例如 1～3 天）收集它们的指标。

---

七、选择：用简单规则决定是否“晋升”候选配置

实验周期结束后，控制层从 TraceStore / 指标库里拉出各配置表现，做一次对比。

一个简单但实用的规则可以是：

• 在关键指标上显著不差于基线，就可以考虑晋升；
• 指标包括：
  • 质量维度：好评率 / 有效回答率
  • 性能维度：平均耗时 & 超时率
  • 成本维度：平均 token 消耗

伪代码示例：

def should_promote(candidate_metrics, baseline_metrics,
                   max_latency_increase_ratio=1.2,
                   min_quality_improve=0.02):
    """
    :param candidate_metrics/baseline_metrics: dict，包含 keys:
      - avg_latency_ms
      - success_rate
      - positive_feedback_rate
    """
    # 质量提升至少 2 个百分点
    if candidate_metrics["positive_feedback_rate"] < baseline_metrics["positive_feedback_rate"] + min_quality_improve:
        return False

    # 延迟不能恶化太多（不超过 20%）
    if candidate_metrics["avg_latency_ms"] > baseline_metrics["avg_latency_ms"] * max_latency_increase_ratio:
        return False

    # 可以加更多约束，如错误率、超时率
    return True

一旦某个候选被认定“值得晋升”，控制层就可以：

1. 把它的配置写入「线上默认配置文件」（例如 workflow_config_active.yaml）；
2. 调整权重：基线配置权重降低，候选权重提高；
3. 记录一次「变更事件」，便于回溯。

---

八、落地建议：如何在你现有项目中逐步启用“自我调优”

不需要一口气把整个自动化控制环写完，可以按以下步骤渐进式落地：

1. 先做“半自动”

   • 用 PerformanceAnalyzer 定期输出「热点问题节点列表」；
   • 人工根据这些信息手动修改配置，观察效果。

2. 加上候选配置生成器（ConfigMutator）​

   • 仍然由开发/运维手动指定「哪种变异方向」；
   • 控制层只负责帮你生成多套配置文件，人工挑选后上线。

3. 接入 A/B + 指标聚合，做到“自动试验、人工决策”

   • 控制层自动跑实验；
   • 把对比报告（Excel / Markdown）发到群里，由人拍板是否切换。

4. 最后才考虑全自动晋升

   • 在有充足监控与回滚能力的前提下，允许控制层在某些安全场景下自动调高某个配置的权重。
   • 高风险场景依然保留人工审批。

整个演进过程中，你做的本质上是：

• 把「配置」抽象成可序列化的对象（YAML/JSON）；
• 把「评估」抽象成一组指标函数；
• 用一个简单的控制循环不断迭代它们。

---

九、总结

这篇的目标是让你的多智能体 + 工作流系统，从：

“可以跑、多配置、有人调”

进化成：

“会看数据、会做实验、能给出更优配置建议，甚至在低风险场景中自动调优”。

关键步骤包括：

1. 明确工作流里的可调参数与可替换模块；
2. 强化 Trace 结构，让每次执行都有可用于评估的数据；
3. 用「控制层」包装执行层，实现：
   • 热点识别（观察）；
   • 候选生成（变异/搜索）；
   • 流量试验（A/B）；
   • 配置晋升（选择与回滚）。

你可以先在一个很小的、风险极低的场景试一试（例如“日报自动生成”的工作流），
等这套“自我调优环”稳定之后，再逐步推广到其它场景。