跳转至

MCP 工具

cg_rag_mcp 将 CG_RAG 服务包装为 FastMCP 工具。工具返回统一结构，区分 MCP 调用是否成功以及底层 CG_RAG 操作是否成功。

工具概览

  • cg_health：返回 CG_RAG 服务健康状态。
  • cg_list_profiles：列出可用检索 profile。
  • cg_retrieve_rerank：执行召回和重排，只返回候选条文。
  • cg_constrained_generate：基于候选条文执行受约束生成。
  • cg_rag：执行完整 CG_RAG 流程。

cg_rag_mcp

make_cg_rag_mcp_lifespan(get_mcp: Callable[[], FastMCP | None]) -> Callable[[Any], AsyncIterator[None]]

源代码位于： eval_rag_results/scripts/question_generated_app/cg_rag_mcp.py 
73
74
75
76
77
78
79
80
81
82
83
def make_cg_rag_mcp_lifespan(get_mcp: Callable[[], FastMCP | None]) -> Callable[[Any], AsyncIterator[None]]:
    @asynccontextmanager
    async def lifespan(_app: Any) -> AsyncIterator[None]:
        mcp = get_mcp()
        if mcp is None:
            yield
            return
        async with mcp.session_manager.run():
            yield

    return lifespan

build_cg_rag_mcp(service: Any) -> FastMCP

源代码位于： eval_rag_results/scripts/question_generated_app/cg_rag_mcp.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def build_cg_rag_mcp(service: Any) -> FastMCP:
    mcp = FastMCP("CG_RAG", json_response=True, streamable_http_path="/")

    @mcp.tool()
    async def cg_health() -> dict[str, Any]:
        """返回 CG_RAG 服务健康状态、active scope、可用 profile、rerank 和 LLM 配置状态。"""
        return await _call_service(service.health)

    @mcp.tool()
    async def cg_list_profiles() -> dict[str, Any]:
        """列出 CG_RAG 可用检索范围 profile，例如 usual、usual_plus_law 和 full。"""
        return await _call_service(lambda: {"profiles": service.profiles()})

    @mcp.tool()
    async def cg_retrieve_rerank(
        query: TextInput = None, scope: TextInput = "usual", topk: IntInput = 78
    ) -> dict[str, Any]:
        """执行 CG_RAG 召回和重排，只返回候选条文，不调用 LLM 生成。"""
        return await _call_service(
            lambda: service.retrieve_rerank(CgRetrieveRerankRequest(query=query, scope=scope, topk=topk))
        )

    @mcp.tool()
    async def cg_constrained_generate(
        query: TextInput = None,
        retrieval_docs: DocsInput = None,
        max_items: IntInput = 10,
        include_debug: BoolInput = False,
    ) -> dict[str, Any]:
        """基于 CG_RAG 调用方传入的候选条文执行受约束生成，返回候选索引和格式化条文。"""
        return await _call_service(
            lambda: service.constrained_generate(
                CgConstrainedGenerateRequest(
                    query=query,
                    retrieval_docs=retrieval_docs,
                    max_items=max_items,
                    include_debug=include_debug,
                )
            )
        )

    @mcp.tool()
    async def cg_rag(
        query: TextInput = None,
        scope: TextInput = "usual",
        topk: IntInput = 78,
        max_items: IntInput = 10,
        include_debug: BoolInput = False,
    ) -> dict[str, Any]:
        """执行完整 CG_RAG 流程：召回、重排、受约束生成。该流程不包含 domain gate。"""
        return await _call_service(
            lambda: service.rag(
                CgRagRequest(
                    query=query,
                    scope=scope,
                    topk=topk,
                    max_items=max_items,
                    include_debug=include_debug,
                )
            ),
            operation_ok_from_data=_cg_rag_operation_ok,
        )

    return mcp