This section is recommended to be studied together with the tutorial Branch Structure.

​
What is Router Server

In complex RAG reasoning tasks, we often need to dynamically decide the next execution path based on intermediate results (such as the model’s current generated content or retrieval results). Router Server is designed for this purpose—it is responsible for judging the current state based on input information and returning a branch label (state identifier) to drive the subsequent flow’s branch jump.

​
How to Implement a Router Tool?

We demonstrate how to implement a Router Tool with a simple example and use it in a Pipeline to control the flow branching. Suppose in the current RAG process, we want the model to generate answers for each question. If the model’s answer contains “I don’t know”, we consider it a failure and should regenerate; if the answer is clear, the task is considered completed and the process terminates. We can write the Router Tool like this:
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}
This Tool will tag each answer with a state label:
  • "retry": indicates the answer failed and needs to be regenerated;
  • "finish": indicates the answer succeeded and the process can end.

​
How to Use Router Tool in Pipeline?

Router Tool must be used together with the branch: and router: control structures.
This section is recommended to be studied together with the tutorial Branch Structure.
pipeline:
  - generation.generate
  - loop:
      times: 3
      steps:
        - branch:
            router:
              - router.check_dont_know
            branches:
              retry:
                - generation.generate
              finish: []
  - evaluation.evaluate