什么是Sever?

在典型 RAG 系统中,我们可以将系统拆解为多个功能模块,例如:检索器(Retriever)、生成器(Generation)等。这些模块承担不同任务,通过某种流程编排,协同完成复杂的问答与推理任务。 在 UR-2.0 中,我们基于 MCP(Model Context Protocol) 架构,对功能模块进行了统一封装,提出了更加标准化的实现方式——Server。
Server 本质上就是一个具备独立功能的 RAG 模块组件。
每个 Server 封装一类核心任务逻辑(如检索、生成、评估等),并通过工具函数(Tool)对外提供标准化接口,支持在完整的推理流程(pipeline)中被灵活调度和组合使用。

快速上手:开发一个 Server

为了帮助你理解 Server 的使用方式,我们将以实现一个简单的 “计算器” 模块为例,展示 Server 的完整开发流程。

Step1:创建 Calculator Server

我们首先通过 UltraRAG_MCP_Server 实例化一个名为 calculator 的 Server:
servers/calculator/src/calculator.py
from ultrarag_mcp.server import UltraRAG_MCP_Server

app = UltraRAG_MCP_Server("calculator")

if __name__ == "__main__":
    app.run(transport="stdio")

Step2:实现工具函数(Tool)

使用 @app.tool 装饰器注册工具函数,这些函数将在 pipeline 中被调用:
servers/calculator/src/calculator.py
from ultrarag_mcp.server import UltraRAG_MCP_Server

app = UltraRAG_MCP_Server("calculator")

@app.tool(output="a,b->result")
def add(a: float, b: float) -> Dict[str, float]:
    """Return the sum of a and b"""
    return {"result": a + b}

@app.tool(output="result")
def minus(a: float, b: float) -> Dict[str, float]:
    """Return a - b"""
    return {"result": a - b}
    
@app.tool(output="a,b->None")
def log(a: float, b: float) -> None:
   """Logging a, b"""
   app.logger.info(f"a: {a}, b:{b}")

if __name__ == "__main__":
    app.run(transport="stdio")

Step3:配置参数文件

创建参数配置文件 servers/calculator/parameter.yaml,请将你实现的工具需要用到的参数写入该 server 路径下的 parameter 文件,用于标明参数并注入默认值:
/images/yaml.svgservers/calculator/parameter.yaml
a: 1
b: 2

注意事项:参数注册机制

UR-2.0 在 build 阶段会自动读取每个 Server 目录下的 parameter.yaml 文件,并据此感知并注册工具函数所需的参数。这一机制带来了极大的便利,但也需注意以下事项:
  • 支持参数共享:多个工具函数如需共用某个参数(如 template、path 等),在parameter.yaml 中只声明一次并复用,无需重复定义。
  • 注意字段覆盖风险:若多个工具需要的参数在含义或默认值上存在冲突,请使用不同字段名显式区分,避免在生成的配置文件中被覆盖。
  • 上下文自动推断机制:若工具函数中的某些输入参数未出现在 parameter.yaml 中,UltraRAG 会默认尝试从运行时上下文中获取这些参数(即视为由上游 Tool 的输出提供)。因此,只有无法通过上下文自动传入的参数才需要显式写入 parameter.yaml。

封装为类:支持状态管理

在某些场景我们可能希望在 Server 内部维护一些共享的状态或变量,比如初始化模型,可以将 Server 封装为一个类。以下是将 Calculator Server 封装为类的示例:
servers/calculator/src/calculator.py
from ultrarag_mcp.server import UltraRAG_MCP_Server

app = UltraRAG_MCP_Server("calculator")

class Calculator:
    def __init__(self, mcp_inst: UltraRAG_MCP_Server)
        mcp_inst.tool(self.add, output="b->result")
        mcp_inst.tool(self.minus, output="result")
        mcp_inst.tool(self.log, output="none")
        self.a = 10  # 模拟全局变量
        
    def add(self, b: float) -> Dict[str, float]:
        return {"result": self.a + b}
        
    def minus(self, b: float) -> Dict[str, float]:
        return {"result": self.a - b}
    
    def log(self, b: float):
        app.logger.info(f"a: {a}, b:{b}")
 
if __name__ == "__main__":
    Calculator(app)
    app.run(transport="stdio")
以上内容展示了如何开发一个自定义 Server,并实现其中的工具函数(Tool)。你可以使用这种方式将任意功能封装为模块组件,并在 UR-2.0 的 pipeline 中进行组合调用。 在接下来的小节中,我们将介绍 UR-2.0 中最常用的几个内置 Server,帮助你快速搭建自己的 RAG 流程。