文件agent_mcp/tools/rag_tools.py是本项目中的 RAG（检索增强生成）工具模块，主要功能是提供一个自然语言查询接口，让已认证的代理可以向项目RAG系统提问。它通过验证代理身份、接收查询文本，调用核心RAG系统逻辑处理查询并返回结果，实现了基于项目索引文档、上下文和元数据的智能问答功能，同时记录审计日志以确保操作可追溯。
async def ask_project_rag_tool_impl(arguments: Dict[str, Any]) -> List[mcp_types.TextContent]: agent_auth_token = arguments.get("token") query_text = arguments.get("query") requesting_agent_id = get_agent_id(agent_auth_token) if not requesting_agent_id: return [mcp_types.TextContent(type="text", text="未授权：需要有效的代理令牌")] if not query_text or not isinstance(query_text, str): return [mcp_types.TextContent(type="text", text="错误：查询文本是必需的，且必须是字符串。")] # 记录审计日志（main.py:1578） log_audit(requesting_agent_id, "ask_project_rag", {"query": query_text}) logger.info(f"代理 '{requesting_agent_id}' 正在向项目RAG提问：'{query_text[:100]}...'") try: # 调用来自features/rag/query.py的核心RAG系统函数 # 该函数（query_rag_system）处理所有复杂的RAG逻辑 answer_text = await query_rag_system(query_text) # query_rag_system已内部处理错误并返回字符串 return [mcp_types.TextContent(type="text", text=answer_text)] except Exception as e: # 此捕获块专门用于此tool_impl包装器中的意外错误， # 不包括query_rag_system内部的错误，因为这些错误已由其内部处理 logger.error(f"代理 '{requesting_agent_id}' 的ask_project_rag_tool_impl中发生意外错误：{e}", exc_info=True) return [mcp_types.TextContent(type="text", text=f"处理您的RAG查询时发生意外错误：{str(e)}")] # --- 注册RAG工具 --- def register_rag_tools(): register_tool( name="ask_project_rag", description="询问有关项目的自然语言问题。系统使用RAG（检索增强生成）从索引文档、上下文和元数据中查找相关信息，以合成答案。", input_schema={ "type": "object", "properties": { "token": {"type": "string", "description": "发出查询的代理的认证令牌。"}, "query": {"type": "string", "description": "要询问的有关项目的自然语言问题。"} }, "required": ["token", "query"], "additionalProperties": False }, implementation=ask_project_rag_tool_impl ) # 导入此模块时调用注册函数 register_rag_tools()