本小节建议结合教程 分支型结构 一起学习。

什么是 Router Server

在复杂的 RAG 推理任务中,我们常常需要根据中间结果(如模型当前的生成内容或检索结果)动态决定下一步的执行路径。Router Server 正是为此设计的组件——它负责根据输入信息判断当前状态,并返回一个分支标签(状态标识),用于驱动后续流程的分支跳转。

如何实现一个 Router Tool?

我们通过一个简单的例子来演示如何实现 Router Tool,并在 Pipeline 中使用它完成流程分支控制。 假设当前的 RAG 流程中,我们希望模型对每个问题生成答案。如果模型的回答中包含 “I don’t know”,我们认为它失败了,应该重新生成;如果回答明确,则认为任务完成,流程终止。 我们可以这样编写 Router Tool:
servers/router/src/router.py
@app.tool(output="ans_ls->ans_ls")
def check_dont_know(ans_ls: List[str]) -> Dict[str, List[Dict[str, str]]]:
    ans_ls = [
            {
                "data": ans,
                "state": "retry" if "i don't know" in ans.lower() else "finish"
            }
            for ans in ans_ls
     ]
    return {"ans_ls": ans_ls}
该 Tool 会为每条回答打上状态标签:
  • "retry":表示回答失败,需要再次生成;
  • "finish":表示回答成功,可以结束流程。

如何在 Pipeline 中使用 Router Tool?

Router Tool 必须配合 branch:router: 控制结构使用。
本小节建议结合教程 分支型结构 一起学习。
pipeline:
  - generation.generate
  - loop:
      times: 3
      steps:
        - branch:
            router:
              - router.check_dont_know
            branches:
              retry:
                - generation.generate
              finish: []
  - evaluation.evaluate