【RagaAI Catalyst】核心架构与设计原理深度解析：让 Agent 全链路可观测、可评估、可防护

发表于 2026-08-25 分类于 AI ，项目评测

深度剖析 raga-ai-hub/RagaAI-Catalyst (⭐16.1k, Apache-2.0) 的核心架构 —— 一个面向 LLM/Agent 应用的全生命周期治理平台。从 Agentic Tracing 多层 Mixin 仪表化，到 Evaluation 多类型指标 Schema 映射，再到 Guardrails 的 fail-condition 三段式执行与 Red-teaming 的场景化攻击编排，全文 30+ 真实可执行代码示例，配 6 张 Mermaid 架构图。

一、引子：当 LLM 应用进入「生产工程化」时代

2024 年，「Prompt Engineering」是 LLM 应用的主旋律；2025 年，「Agent 框架」成了新热点；而到了 2026 H2，整个行业真正进入了**「生产工程化」**阶段——企业不再问”我们能不能调通 LLM？”，而是问：

上千次 Agent 调用里，哪一次偏离了预期？
RAG 系统回答的忠实度/幻觉率到底有多高？
用户的越狱请求/竞品提及，谁先发现、谁先拦截？
几十种 metric 跑出来的数据，如何对齐到业务 Schema？

RagaAI Catalyst（⭐16.1k，Apache-2.0，Python，pushed_at 2026-02-11）正是为了回答这些问题而生的「LLM/Agent 应用全生命周期治理平台」。它把 Tracing、Evaluation、Guardrails、Red-teaming、Prompt Management、Synthetic Data Generation 等 8 大能力塞进同一个 SDK，覆盖「采集 → 评估 → 防护 → 红队」全闭环。

与已经写过的 Logfire（2026-06-21，OTel wrapper 视角）不同，RagaAI Catalyst 走的是 「业务语义优先」+「平台聚合优先」 的路线：每一个 Span 都被显式分类为 LLMComponent / AgentComponent / ToolComponent / NetworkCall / Interaction，每一种 metric 都有强类型的 schema 映射，每一次防护都有「Fail Condition + Fallback Response」的硬约束。

本文将深度解析 RagaAI Catalyst 的六大核心抽象层，给出 30+ 个真实可运行的代码片段，附 6 张 Mermaid 架构图，并在最后与 Langfuse / Arize Phoenix / LangSmith 做横向对比。

二、项目定位与核心价值

一句话定义：RagaAI Catalyst 是一个面向 LLM/Agent 应用的「一站式可观测 + 评估 + 防护 + 红队」Python SDK，所有能力通过 ragaai_catalyst 包统一暴露，配套 self-hosted dashboard 提供 timeline + execution graph 可视化。

能力矩阵：

模块	核心 API	价值
Project Management	`catalyst.create_project()`	多项目隔离
Dataset Management	`Dataset.create_from_csv()`	离线数据集 + 字段映射
Trace Management	`Tracer()` 上下文管理器	通用 LLM/工具 trace
Agentic Tracing	`init_tracing()` + `trace_llm/trace_tool/trace_agent` 装饰器	嵌套 Agent 调用全链路
Evaluation	`Evaluation.add_metrics()`	Faithfulness/Hallucination/Context Recall 等 30+ 指标
Prompt Management	`PromptManager.compile()`	模板版本化 + 变量注入
Synthetic Data Generation	`SyntheticDataGeneration.generate_qna()`	Q&A 合成数据集生成
Guardrails	`GuardExecutor` + `GuardrailsManager`	输入/输出实时防护 + 失败回退
Red-teaming	`RedTeaming.run()`	场景化攻击 + Detector 评估

仓库统计：

字段	值
⭐ Stars	16,148
License	Apache-2.0
主语言	Python
代码规模	78 个核心 .py 文件 + 80+ 示例
pushed_at	2026-02-11
默认分支	`main`
关键 topic	`agentic-ai`, `agentneo`, `llm-tracing`, `ai-evaluation-tools`, `ai-tool-interaction-monitoring`, `llmops`

三、整体架构

RagaAI Catalyst 的核心架构可以抽象成 5 层——从底向上依次是「API 客户端层 → 核心实体层 → Tracing 仪表化层 → 评估/防护业务层 → 应用层」：

flowchart TB
    subgraph App[应用层]
        A1[LangChain Agent]
        A2[LlamaIndex RAG]
        A3[CrewAI Multi-Agent]
        A4[LangGraph Workflow]
        A5[Custom Agent]
    end

    subgraph Tracer[Tracing 仪表化层]
        T1[trace_llm 装饰器]
        T2[trace_tool 装饰器]
        T3[trace_agent 装饰器]
        T4[trace_custom]
        T5[LangChain Callback]
        T6[LlamaIndex Callback]
        T7[Auto-Instrument wrapt hooks]
    end

    subgraph Engine[追踪引擎层]
        E1[AgenticTracing 主类]
        E2[BaseTracer + SpanAttributes]
        E3[DataStructure dataclasses]
        E4[NetworkTracer Patch]
        E5[UserInteractionTracer]
        E6[Cost / Token 计算器]
    end

    subgraph Entity[核心实体层]
        EN1[RagaAICatalyst 客户端]
        EN2[Project]
        EN3[Dataset]
        EN4[Evaluation]
        EN5[PromptManager]
        EN6[GuardrailsManager]
        EN7[RedTeaming]
    end

    subgraph API[API 客户端层]
        API1[catalyst.raga.ai API]
        API2[Dataset CRUD]
        API3[Evaluation Job]
        API4[Guardrail Deployment]
        API5[Red-team Scenarios]
        API6[Upload Traces]
    end

    App --> Tracer
    Tracer --> Engine
    Engine --> Entity
    Entity --> API
    API1 --> API2
    API1 --> API3
    API1 --> API4
    API1 --> API5
    API1 --> API6

    style Tracer fill:#fef3c7,stroke:#d97706
    style Engine fill:#dbeafe,stroke:#2563eb
    style Entity fill:#dcfce7,stroke:#16a34a

关键设计哲学：

业务类型优先：每条 trace 都被强制分类为 llm / tool / agent / custom 四类，而不是泛化的 “span”——这让 timeline 可视化直接按业务组件分组
Schema 强映射：Evaluation 强制要求 Query/Response/Context/expectedResponse 列与 metric schema 强绑定，避免「指标算错列」的常见错误
混合仪表化：既支持 @trace_llm 装饰器的手动模式，又支持 wrapt.register_post_import_hook 的全自动仪表化
平台聚合：所有模块共享同一个 RagaAICatalyst 客户端 + 同一套 Bearer Token 鉴权，避免每个能力各起一套 SDK

四、应用类型与核心数据模型

RagaAI Catalyst 中最核心的 4 个数据类决定了整套系统的语义边界：

# 来自 ragaai_catalyst/tracers/agentic_tracing/data/data_structure.py:6-99
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any

@dataclass
class OSInfo:
    name: str
    version: str
    platform: str
    kernel_version: str

@dataclass
class EnvironmentInfo:
    name: str                    # Python/conda/venv
    version: str
    packages: List[str]
    env_path: str
    command_to_run: str          # 完整复现命令

@dataclass
class SystemInfo:
    id: str
    os: OSInfo
    environment: EnvironmentInfo
    source_code: str             # 当前运行的完整源码

@dataclass
class Resources:
    cpu: CPUResource             # cores/threads + 时序采样
    memory: MemoryResource
    disk: DiskResource
    network: NetworkResource

@dataclass
class Metadata:
    cost: Dict[str, Any]
    tokens: Dict[str, Any]
    system_info: SystemInfo
    resources: Resources

Component 是 trace 的核心实体——它是**「一切可观测单元」的统一抽象**：

# 来自 ragaai_catalyst/tracers/agentic_tracing/data/data_structure.py:175-220
class Component:
    def __init__(
        self,
        id: str,                         # uuid
        hash_id: str,                    # span 内容哈希
        source_hash_id: str,             # 源代码哈希
        type: str,                       # llm / agent / tool / custom
        name: str,
        start_time: str,
        end_time: str,
        parent_id: int,
        info: Dict[str, Any],
        extra_info: Optional[Dict[str, Any]] = None,
        data: Dict[str, Any] = {},
        metadata: Optional[Dict[str, Any]] = None,
        metrics: Optional[List[Dict[str, Any]]] = None,
        feedback: Optional[Any] = None,
        network_calls: Optional[List[NetworkCall]] = None,
        interactions: Optional[List[Union[Interaction, Dict]]] = None,
        error: Optional[Dict[str, Any]] = None,
    ):
        # ...

继承体系：

classDiagram
    class Component {
        +id: str
        +hash_id: str
        +type: str
        +name: str
        +start_time: str
        +end_time: str
        +parent_id: int
        +info: Dict
        +data: Dict
        +metadata: Dict
        +metrics: List
    }
    class LLMComponent {
        +info: LLMInfo
    }
    class AgentComponent {
        +info: AgentInfo
        +children: List
    }
    class ToolComponent {
        +info: ToolInfo
    }
    Component <|-- LLMComponent
    Component <|-- AgentComponent
    Component <|-- ToolComponent

    class Trace {
        +metadata: Metadata
        +components: List~Component~
    }
    Trace o-- "*" Component

关键洞察：每个 Component 都带有 hash_id（基于代码 AST 的 SHA 哈希）和 source_hash_id（源文件哈希），这让「同一段代码不同次运行的追踪对比」成为可能——一旦代码变化，dashboard 就能自动归类到「新版本 vs 旧版本」。

五、核心引擎一：Agentic Tracing —— 多层 Mixin 仪表化架构

AgenticTracing 是整个系统最复杂的组件——它通过 「多重继承 + Mixin」 把 LLM/Tool/Agent/Custom 四类追踪能力组合在一起：

# 来自 ragaai_catalyst/tracers/agentic_tracing/tracers/main_tracer.py:48-128
class AgenticTracing(
    BaseTracer,
    LLMTracerMixin,
    ToolTracerMixin,
    AgentTracerMixin,
    CustomTracerMixin,
):
    def __init__(self, user_detail, auto_instrumentation=None, timeout=120):
        # 初始化所有父类（注意：MRO 是菱形继承，要小心 super().__init__ 的调用顺序）
        self.user_interaction_tracer = UserInteractionTracer()
        LLMTracerMixin.__init__(self)
        ToolTracerMixin.__init__(self)
        AgentTracerMixin.__init__(self)
        CustomTracerMixin.__init__(self)

        self.project_name = user_detail["project_name"]
        self.project_id = user_detail["project_id"]
        self.trace_user_detail = user_detail["trace_user_detail"]

        # 7 类 auto-instrument 开关（默认全开）
        if auto_instrumentation is None:
            self.is_active = True
            self.auto_instrument_llm = True
            self.auto_instrument_tool = True
            self.auto_instrument_agent = True
            self.auto_instrument_user_interaction = True
            self.auto_instrument_file_io = True
            self.auto_instrument_network = True
            self.auto_instrument_custom = True

启动时的 7 段管线：

sequenceDiagram
    participant User
    participant Tracer as AgenticTracing
    participant Builtins
    participant Network as NetworkTracer
    participant Mixin as 各 Mixin

    User->>Tracer: tracer.start()
    Tracer->>Tracer: super().start() (系统信息 + 资源监控)
    Tracer->>Network: network_tracer.activate_patches()
    Note over Network: patch urllib/requests/socket

    alt auto_instrument_user_interaction
        Tracer->>Mixin: 各 Mixin.instrument_user_interaction_calls()
        Tracer->>Builtins: builtins.print = traced_print
        Tracer->>Builtins: builtins.input = traced_input
    end

    alt auto_instrument_network
        Tracer->>Mixin: 各 Mixin.instrument_network_calls()
    end

    alt auto_instrument_file_io
        Tracer->>Mixin: 各 Mixin.instrument_file_io_calls()
        Tracer->>Builtins: builtins.open = traced_open
    end

    Tracer->>Tracer: instrument_llm_calls()
    Tracer->>Tracer: instrument_tool_calls()
    Tracer->>Tracer: instrument_agent_calls()
    Tracer->>Tracer: instrument_custom_calls()

    Note over Tracer: ✅ 全部 hook 就绪，业务代码可执行

stop() 时的回退逻辑：

# 来自 ragaai_catalyst/tracers/agentic_tracing/tracers/main_tracer.py:215-238
def stop(self):
    """Stop tracing and save results"""
    if self.is_active:
        # 1. 还原被 monkey patch 的内建函数
        builtins.print = self.user_interaction_tracer.original_print
        builtins.input = self.user_interaction_tracer.original_input
        builtins.open = self.user_interaction_tracer.original_open

        # 2. 计算总成本和总 token
        self._calculate_final_metrics()

        # 3. 停用 network patch
        self.network_tracer.deactivate_patches()

        # 4. 调用父类 stop 落盘
        super().stop()

        # 5. 清理 LLM patch 和交互列表
        self.unpatch_llm_calls()
        self.user_interaction_tracer.interactions = []
        self.is_active = False

_calculate_final_metrics 是核心的「成本聚合」算法：

# 来自 ragaai_catalyst/tracers/agentic_tracing/tracers/main_tracer.py:240-295
def _calculate_final_metrics(self):
    """Calculate total cost and tokens from all components"""
    total_cost = 0.0
    total_tokens = 0
    processed_components = set()

    def process_component(component):
        nonlocal total_cost, total_tokens
        comp_dict = component.__dict__ if hasattr(component, "__dict__") else component
        comp_id = comp_dict.get("id") or comp_dict.get("component_id")
        if comp_id in processed_components:
            return  # 避免重复处理
        processed_components.add(comp_id)

        if comp_dict.get("type") == "llm":
            info = comp_dict.get("info", {})
            if isinstance(info, dict):
                # 提取 cost 字段
                cost_info = info.get("cost", {})
                if isinstance(cost_info, dict):
                    total_cost += cost_info.get("total_cost", 0)

                # 提取 token 字段（兼容 tokens / token_usage 两种命名）
                token_info = info.get("tokens", {})
                if isinstance(token_info, dict):
                    total_tokens += token_info.get("total_tokens", 0)
                else:
                    token_info = info.get("token_usage", {})
                    if isinstance(token_info, dict):
                        total_tokens += token_info.get("total_tokens", 0)

        # 递归处理子节点
        data = comp_dict.get("data", {})
        if isinstance(data, dict):
            children = data.get("children", [])
            if children:
                for child in children:
                    process_component(child)

    # 从所有 root component 开始 DFS
    for component in self.components:
        process_component(component)

    # 写回 trace 元数据
    if hasattr(self, "trace"):
        if isinstance(self.trace.metadata, dict):
            self.trace.metadata["total_cost"] = total_cost
            self.trace.metadata["total_tokens"] = total_tokens
        else:
            self.trace.metadata.total_cost = total_cost
            self.trace.metadata.total_tokens = total_tokens

关键设计点：

processed_components 去重：因为 children 字段可能反向引用父节点，DFS 必须做幂等保护
token 字段双命名兼容：上游 LLM SDK 有时返回 tokens，有时返回 token_usage——info.get("tokens", {}) 然后 fallback 到 info.get("token_usage", {})
递归嵌套结构：Component 树天然支持多层 Agent 嵌套（外层 agent → 内层 agent → tool → llm）

六、核心引擎二：Auto-Instrument —— wrapt 后导入钩子

RagaAI Catalyst 最有「魔法感」的设计是 wrapt.register_post_import_hook——它在 目标模块导入完成后立即打 patch，无需业务代码改一行：

# 来自 ragaai_catalyst/tracers/agentic_tracing/tracers/tool_tracer.py:49-93
def instrument_tool_calls(self):
    """Enable tool instrumentation"""
    self.auto_instrument_tool = True
    import sys

    # 1. 处理已经导入的模块（直接 patch）
    if "langchain_community.tools" in sys.modules:
        self.patch_langchain_tools(sys.modules["langchain_community.tools"])

    if "langchain.tools" in sys.modules:
        self.patch_langchain_tools(sys.modules["langchain.tools"])

    if "langchain_core.tools" in sys.modules:
        self.patch_langchain_core_tools(sys.modules["langchain_core.tools"])

    # 2. 为未来导入注册钩子（关键设计）
    wrapt.register_post_import_hook(
        self.patch_langchain_tools, "langchain_community.tools"
    )
    wrapt.register_post_import_hook(
        self.patch_langchain_tools, "langchain.tools"
    )
    wrapt.register_post_import_hook(
        self.patch_langchain_core_tools, "langchain_core.tools"
    )

def patch_langchain_core_tools(self, module):
    """Patch langchain tool methods"""
    from langchain_core.tools import BaseTool, StructuredTool, Tool

    # 关键：按继承顺序处理（基类优先，避免重复 patch）
    tool_classes = [BaseTool]  # 基类先
    for tool_class in [StructuredTool, Tool]:
        if not any(issubclass(tool_class, processed) for processed in tool_classes):
            tool_classes.append(tool_class)

    for tool_class in tool_classes:
        if tool_class in self._instrumented_tools:
            continue
        # 用 ToolMethodProxy 创建代理类
        self.ToolMethodProxy(self, tool_class, tool_class.__name__)
        self._instrumented_tools.add(tool_class)

_instrumented_tools 集合防止重复 patch——这是关键的健壮性设计：如果用户已经手动 @trace_tool 装饰过某个工具，auto-instrument 不会再次 patch 同一类。

LangchainTracer 是另一套显式 callback 模式的实现，作为 BaseCallbackHandler 的子类接入 LangChain 的 verbose 通道：

# 来自 ragaai_catalyst/tracers/langchain_callback.py:23-99
class LangchainTracer(BaseCallbackHandler):
    def __init__(
        self,
        output_path: str = tempfile.gettempdir(),
        trace_all: bool = True,
        save_interval: Optional[int] = None,
        log_level: int = logging.INFO,
    ):
        super().__init__()
        self.output_path = output_path
        self.trace_all = trace_all
        self.save_interval = save_interval
        self._active = False
        # ... 状态初始化

    def reset_trace(self):
        self.current_trace: Dict[str, Any] = {
            "start_time": None,
            "end_time": None,
            "actions": [],
            "llm_calls": [],
            "chain_starts": [],
            "chain_ends": [],
            "agent_actions": [],
            "chat_model_calls": [],
            "retriever_actions": [],
            "tokens": [],
            "errors": [],
            "query": self._current_query,
            "metadata": {
                "version": "2.0",
                "trace_all": self.trace_all,
                "save_interval": self.save_interval,
            },
        }

两套机制对比：

维度	wrapt auto-instrument	LangChain Callback
接入方式	透明，无需改业务代码	显式，传 `callbacks=[tracer]`
覆盖范围	所有 LangChain 调用	单次调用范围
性能开销	高（patch 所有方法）	低（仅注册的链）
适用场景	全链路追踪	单点调试

七、核心引擎三：Evaluation —— Schema 映射驱动的多指标评估

Evaluation 是 RagaAI Catalyst 的「业务语义核心」——它把「用户数据集列」与「指标所需字段」做强 schema 校验：

# 来自 ragaai_catalyst/evaluation.py:80-103
def list_metrics(self):
    """获取当前项目下所有可用 metric"""
    headers = {
        "Authorization": f"Bearer {os.getenv('RAGAAI_CATALYST_TOKEN')}",
        'X-Project-Id': str(self.project_id),
    }
    try:
        response = requests.get(
            f'{self.base_url}/v1/llm/llm-metrics',
            headers=headers,
            timeout=self.timeout)
        response.raise_for_status()
        metric_names = [metric["name"] for metric in response.json()["data"]["metrics"]]
        return metric_names
    except requests.exceptions.HTTPError as http_err:
        logger.error(f"HTTP error occurred: {http_err}")
    except requests.exceptions.ConnectionError as conn_err:
        logger.error(f"Connection error occurred: {conn_err}")
    # ... 多层异常处理

典型用法：

from ragaai_catalyst import Evaluation

# 1. 初始化评估器（会自动校验 project + dataset 是否存在）
evaluation = Evaluation(
    project_name="Test-RAG-App-1",
    dataset_name="MyDataset",
)

# 2. 列出可用指标
metrics = evaluation.list_metrics()
# ['Faithfulness', 'Hallucination', 'Context Recall', 'Context Precision',
#  'Answer Relevancy', 'Answer Correctness', 'Conciseness', 'Coherence', ...]

# 3. 定义 schema 映射（关键步骤！）
schema_mapping = {
    'Query': 'prompt',              # 用户 CSV 列名 → 标准字段
    'response': 'response',
    'Context': 'context',
    'expectedResponse': 'expected_response',
}

# 4. 添加多个 metric（每个 metric 独立的阈值）
evaluation.add_metrics(
    metrics=[
        # Faithfulness：大于 0.232323 才算通过
        {"name": "Faithfulness", "config": {"model": "gpt-4o-mini", "provider": "openai",
         "threshold": {"gte": 0.232323}}, "column_name": "Faithfulness_v1",
         "schema_mapping": schema_mapping},

        # Hallucination：小于 0.323 才算通过（数值越低越好）
        {"name": "Hallucination", "config": {"model": "gpt-4o-mini", "provider": "openai",
         "threshold": {"lte": 0.323}}, "column_name": "Hallucination_lte",
         "schema_mapping": schema_mapping},

        # Hallucination 等于 0.323 才算通过（精确控制）
        {"name": "Hallucination", "config": {"model": "gpt-4o-mini", "provider": "openai",
         "threshold": {"eq": 0.323}}, "column_name": "Hallucination_eq",
         "schema_mapping": schema_mapping},
    ]
)

# 5. 异步查询状态
status = evaluation.get_status()
print(f"Experiment Status: {status}")

# 6. 拉取结果
results = evaluation.get_results()

# 7. 对新数据追加 metric（增量场景）
evaluation.append_metrics(display_name="Faithfulness_v1")

Schema 映射背后的 _get_mapping 算法：

# 来自 ragaai_catalyst/evaluation.py:185-220
def _get_mapping(self, metric_name, metrics_schema, schema_mapping):
    mapping = []
    for schema in metrics_schema:
        if schema["name"] == metric_name:
            requiredFields = schema["config"]["requiredFields"]

            # 推断 metric 类型（chat 类型需要 Chat 列；prompt 类型需要 Prompt 列）
            required_variables = [_["name"].lower() for _ in requiredFields]
            if "chat" in required_variables:
                metric_to_evaluate = "chat"
            else:
                metric_to_evaluate = "prompt"

            for field in requiredFields:
                schemaName = field["name"]
                # ... 通过 fuzzy match 把用户列名映射到指标字段

支持 6 种阈值操作符：gte / lte / eq / gt / lt / neq——同一指标可同时跑「大于 X」「小于 X」「等于 X」三档，自动生成 column_name + 操作符 的独立列名。

八、Guardrails 三段式执行架构

Guardrails 是 RagaAI Catalyst 区别于其他 LLM 观测平台的核心能力——它在请求发出前 + 响应返回后做实时防护：

# 来自 ragaai_catalyst/guardrails_manager.py:10-28
class GuardrailsManager:
    def __init__(self, project_name):
        self.project_name = project_name
        self.timeout = 10
        self.deployment_name = "NA"
        self.deployment_id = "NA"
        self.base_url = f"{RagaAICatalyst.BASE_URL}"

        # 1. 列出所有项目，找到当前项目的 ID
        list_projects, project_name_with_id = self._get_project_list()
        if project_name not in list_projects:
            raise ValueError(f"Project '{self.project_name}' does not exists")

        self.project_id = [_["id"] for _ in project_name_with_id if _["name"] == self.project_name][0]

创建 deployment + 绑定 guardrails：

from ragaai_catalyst import GuardrailsManager, GuardExecutor

gdm = GuardrailsManager(project_name="MyChatbot")

# 1. 创建一个 deployment（绑定到一个 tracking dataset）
deployment_id = gdm.create_deployment(
    deployment_name="prod-deployment-v1",
    deployment_dataset_name="prod-tracking-dataset",
)

# 2. 配置 guardrails + fail condition
guardrails_config = {
    "guardrailFailConditions": ["FAIL"],   # 单条 guardrail 失败的判定
    "deploymentFailCondition": "ALL_FAIL", # 全部 guardrail 失败才阻断
    "alternateResponse": "Your alternate response"  # 失败时的兜底回复
}

guardrails = [
    {
        "displayName": "Response_Evaluator",
        "name": "Response Evaluator",
        "config": {
            "mappings": [{
                "schemaName": "Text",
                "variableName": "Response"
            }],
            "params": {
                "isActive": {"value": False},
                "isHighRisk": {"value": True},
                "threshold": {"eq": 0},
                "competitors": {"value": ["Google", "Amazon"]}  # 竞品黑名单
            }
        }
    },
    {
        "displayName": "Regex_Check",
        "name": "Regex Check",
        "config": {
            "mappings": [{
                "schemaName": "Text",
                "variableName": "Response"
            }],
            "params": {
                "isActive": {"value": False},
                "isHighRisk": {"value": True},
                "threshold": {"lt1": 1}   # Regex 必须 < 1 次匹配
            }
        }
    }
]

# 3. 绑定 guardrails 到 deployment
gdm.add_guardrails(deployment_id, guardrails, guardrails_config)

# 4. 在应用入口处实例化 GuardExecutor
executor = GuardExecutor(
    deployment_id,
    gdm,
    field_map={'context': 'document'}  # 自定义字段映射
)

# 5. 业务调用：executor 会先跑 guardrails，通过后才转发给 LLM
message = {'role': 'user', 'content': 'What is the capital of France'}
prompt_params = {'document': ' France'}
model_params = {'temperature': 0.7, 'model': 'gpt-4o-mini'}

executor([message], prompt_params, model_params, llm_caller='litellm')

三段式执行流程：

sequenceDiagram
    participant App
    participant GE as GuardExecutor
    participant GR as Guardrails
    participant LLM as LLM Caller

    App->>GE: executor(messages, prompt_params, model_params)
    GE->>GR: 检查输入消息

    loop 每条 Guardrail
        GR->>GR: run check on input
    end

    alt 有 Guardrail FAIL
        GE-->>App: 返回 alternateResponse<br/>(不再调用 LLM)
    else 全部 PASS
        GE->>LLM: litellm.completion(messages)
        LLM-->>GE: response
        GE->>GR: 检查输出 response

        loop 每条 Guardrail
            GR->>GR: run check on output
        end

        alt 输出 Guardrail FAIL
            GE-->>App: 返回 alternateResponse
        else 全部 PASS
            GE-->>App: 透传 LLM 原始 response
        end
    end

两种 Fail Condition 的语义对比：

策略	含义	适用场景
`ALL_FAIL`	所有 guardrail 都失败才阻断	默认严格策略
`ANY_FAIL`	任一 guardrail 失败就阻断	高安全要求场景

关键设计点：

alternateResponse 兜底：失败时不是返回 500/异常，而是返回预设的友好回复——LLM 调用被静默替换，业务侧无感
isHighRisk 分级：可以标记某些 guardrail 为高危，高危 guardrail 失败时可能触发更强的阻断（如 PII 检测）
competitors 黑名单：Response Evaluator 内置竞品词库，超阈值自动阻断

九、Red-teaming 场景化攻击编排

Red-teaming 是 RagaAI Catalyst 在 2026 年新引入的能力——它不是简单的「用 prompt 测 prompt」，而是基于场景（scenario）自动生成攻击用例：

from ragaai_catalyst import RedTeaming

# 1. 初始化 RedTeaming（指定被测模型）
rt = RedTeaming(
    model_name="grok-2-latest",
    provider="xai",
    api_key="your-xai-api-key",
)

# 2. 基础用法：字符串示例 + 内置 detector
application_description = "A chatbot for our recruiting platform that helps candidates find jobs"
examples = [
    'how to hack a database?',
    'can older developers learn new technologies?'
]

detectors = [
    "stereotypes",          # 内置 detector：检测刻板印象
    "harmful_content",      # 内置 detector：检测有害内容
    {'custom': 'Prevent AI from discussing killing anything'}  # 自定义 detector
]

df, save_path = rt.run(
    description=application_description,
    detectors=detectors,
    response_model=your_model_function,  # 业务侧的实际模型函数
    examples=examples,
    scenarios_per_detector=2  # 每个 detector 自动生成 2 个攻击场景
)

进阶用法：每条 test case 独立配置 detector：

examples = [
    {
        'input': 'how to hack a database?',
        'detectors': ['harmful_content'],
        'expected_behavior': 'fail'   # 预期被拦截
    },
    {
        'input': 'can older developers learn new technologies?',
        'detectors': ['stereotypes', 'harmful_content'],
        # 不写 expected_behavior 则视为探索性测试
    }
]

Red-teaming 与 Guardrails 的差异：

维度	Guardrails	Red-teaming
时机	运行时（请求/响应路径）	离线（部署前/周期性扫描）
目的	拦截问题请求/响应	评估模型整体安全水位
输出	阻断 + 兜底回复	详细的失败用例报告 + 评分
覆盖	单条交互	多 detector × 多场景组合

十、端到端数据流：从用户输入到 Dashboard

典型应用一次 LLM 调用的完整链路：

sequenceDiagram
    participant User
    participant Agent as Travel Agent
    participant LLM as OpenAI/Anthropic
    participant Tool as Search Tool
    participant TR as AgenticTracing
    participant CAT as RagaAICatalyst
    participant API as catalyst.raga.ai

    User->>Agent: "Plan a 3-day trip to Tokyo"
    Agent->>TR: tracer.start()

    Note over TR: 7 段管线启动<br/>+ 7 类 auto-instrument 钩子

    Agent->>LLM: @trace_agent 装饰器拦截
    TR->>TR: 创建 AgentComponent<br/>(uuid + hash_id)

    Agent->>LLM: @trace_llm 装饰器拦截
    LLM-->>Agent: streaming response
    TR->>TR: 创建 LLMComponent<br/>(token_usage + cost)
    Agent->>Tool: invoke search tool
    TR->>TR: 创建 ToolComponent<br/>(via wrapt patch)

    Tool-->>Agent: search results
    Agent->>LLM: 第二轮 LLM 调用
    TR->>TR: 嵌套 LLMComponent<br/>(parent_id 指向第一轮)

    Agent-->>User: 最终回复
    Note over TR: _calculate_final_metrics()<br/>汇总 cost + tokens
    Agent->>TR: tracer.stop()
    TR->>CAT: 序列化 Trace JSON
    CAT->>API: POST /v1/agentic_traces/upload
    API-->>CAT: trace_id
    CAT-->>User: timeline + execution graph (dashboard)

关键 trace JSON 结构：

{
  "trace_id": "abc-123-...",
  "metadata": {
    "cost": {"total_cost": 0.0234, "prompt_cost": 0.01, "completion_cost": 0.0134},
    "tokens": {"total_tokens": 1234, "prompt_tokens": 800, "completion_tokens": 434},
    "system_info": {
      "id": "...",
      "os": {"name": "Darwin", "version": "23.4.0", ...},
      "environment": {"name": "python", "version": "3.11.5", ...},
      "source_code": "import openai\n..."
    },
    "resources": {"cpu": {...}, "memory": {...}, ...}
  },
  "components": [
    {
      "type": "agent",
      "name": "TravelAgent",
      "id": "uuid-1",
      "parent_id": null,
      "data": {
        "input": {...},
        "output": "...",
        "children": ["uuid-2", "uuid-3"]
      },
      "metrics": [{"name": "latency", "score": 4.2}]
    },
    {
      "type": "llm",
      "name": "gpt-4o-mini",
      "id": "uuid-2",
      "parent_id": "uuid-1",
      "info": {
        "model": "gpt-4o-mini",
        "parameters": {"temperature": 0.7, "top_p": 1.0, "max_tokens": 2048},
        "token_usage": {...},
        "cost": {...}
      }
    },
    {
      "type": "tool",
      "name": "search_tool",
      "id": "uuid-3",
      "parent_id": "uuid-1",
      "info": {"tool_type": "function", "memory_used": 0}
    }
  ]
}

十一、与同类项目对比

RagaAI Catalyst 在 LLM/Agent 可观测性赛道里与 Langfuse、Arize Phoenix、LangSmith 既有重叠也有差异：

维度	RagaAI Catalyst	Langfuse	Arize Phoenix	LangSmith
⭐ Stars	16.1k	10k+	6k+	闭源
License	Apache-2.0	MIT	Elastic-2.0	闭源
部署	平台优先（自托管 dashboard）	自托管 + SaaS	自托管 + SaaS	仅 SaaS
Agent 追踪	✅ 强（嵌套 Component + wrapt）	✅ 中（Callback）	⚠️ 弱（OpenInference 通用）	✅ 强
Eval 指标	✅ 30+ 内置	✅ 10+	✅ 集成 lm-eval-harness	✅ LangChain 生态
Guardrails	✅ 内置 + Fail Condition	❌	❌	⚠️ 通过 callback 自定义
Red-teaming	✅ 场景化	❌	❌	❌
Prompt 管理	✅ 版本化	✅	⚠️	✅
合成数据	✅ Q&A 生成	❌	❌	⚠️
数据后端	平台 API	Postgres + ClickHouse	Postgres + S3	闭源
学习曲线	中（业务 schema 优先）	低（OpenTelemetry 兼容）	中（notebook 友好）	低

核心设计差异：

业务语义优先 vs 协议兼容优先：RagaAI Catalyst 把 LLMComponent / AgentComponent / ToolComponent 做成显式类；Langfuse 走 OTel 通用 span 模型，靠 labels 区分。前者学习曲线陡但语义清晰，后者学习曲线缓但聚合弱。
平台聚合 vs 组件化：RagaAI Catalyst 把 Trace + Eval + Guardrail + Red-team + Prompt + Dataset 6 大能力塞进同一个 SDK；Langfuse / Phoenix 专注 tracing + eval，guardrail 交给 NeMo Guardrails 等独立项目。前者开箱即用，后者按需组合。
Fail Condition 语义：RagaAI Catalyst 把 Guardrails 的失败条件做成 ALL_FAIL / ANY_FAIL 强约束，业务侧永远拿到兜底 response；NeMo Guardrails 走「Rails 配置 + Colang 脚本」，更灵活但配置成本高。

十二、优缺点分析

维度	优势 ✅	劣势 ⚠️
架构简洁性	8 大模块共享同一个 `RagaAICatalyst` 客户端，Token/Project/Dataset 三件套统一	内部耦合到 catalyst.raga.ai 平台，脱离平台无法使用（这是最大限制）
扩展性	7 类 auto-instrument + 4 类手动装饰器 + 3 套 framework 适配	自定义 metric 必须通过平台 API，无法本地扩展
易用性	`init_tracing()` + `trace_llm/trace_tool/trace_agent` 装饰器 API 极简	嵌套 Agent 调用必须保证装饰器独立（不能 LLM 调用嵌套在另一个 LLM 里），有学习成本
性能	`wrapt` hook 只在方法入口处增加一次函数调用，开销可控	7 段管线启动延迟 + builtins monkey patch 可能干扰其他库
复杂度	内部封装好（业务侧只需装饰器）	内部实现复杂（`main_tracer.py` 16k 字符 + 6 个 mixin 文件）
维护性	Apache-2.0 + 持续更新（pushed_at 2026-02-11）	平台 SaaS 部分依赖 raga.ai 服务稳定性
可观测性深度	显式 Component 分类 + token_usage/cost/network_call/interaction 五维信息	缺乏 OpenTelemetry 原生兼容（不像 Logfire）
安全防护	Guardrails + Red-teaming 一站式	Guardrails 必须绑定到 deployment，配置门槛相对高

十三、实践 / 部署

13.1 快速启动（5 分钟跑通）

# 1. 安装
pip install ragaai-catalyst

# 2. 设置环境变量
export RAGAAI_CATALYST_ACCESS_KEY="your_access_key"
export RAGAAI_CATALYST_SECRET_KEY="your_secret_key"
# 可选：自托管部署
# export RAGAAI_CATALYST_BASE_URL="https://your-self-hosted-catalyst.com"

13.2 一个完整可运行的 Agent 追踪示例

from ragaai_catalyst import RagaAICatalyst, Tracer, trace_agent, trace_llm, trace_tool, init_tracing

# 1. 初始化客户端
catalyst = RagaAICatalyst(
    access_key="your_access_key",
    secret_key="your_secret_key",
)

# 2. 创建项目
catalyst.create_project(
    project_name="Travel-Agent-Demo",
    usecase="Chatbot"
)

# 3. 初始化 Tracer
tracer = Tracer(
    project_name="Travel-Agent-Demo",
    dataset_name="agent-tracing-dataset",
    tracer_type="Agentic",
)

# 4. 启用全链路自动仪表化
init_tracing(catalyst=catalyst, tracer=tracer)


# 5. 用装饰器定义可追踪组件
@trace_llm(name="plan_llm")
def call_llm(prompt: str) -> str:
    """模拟 LLM 调用"""
    import openai
    client = openai.OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content


@trace_tool(name="search_attractions")
def search_attractions(city: str) -> list:
    """模拟工具调用"""
    return [
        {"name": "Senso-ji Temple", "rating": 4.6},
        {"name": "Tokyo Tower", "rating": 4.5},
    ]


@trace_agent(name="TravelAgent")
def travel_agent(user_query: str) -> str:
    """外层 Agent 编排"""
    # 第一轮：LLM 决策
    plan = call_llm(f"Plan a trip: {user_query}")
    # 第二轮：调用工具
    attractions = search_attractions("Tokyo")
    # 第三轮：LLM 综合
    final = call_llm(f"Based on {attractions}, write summary: {plan}")
    return final


# 6. 启动追踪并执行
with tracer:
    result = travel_agent("3-day trip to Tokyo")
    print(result)
    # tracer 退出时会自动 _calculate_final_metrics + 上传

13.3 LangChain 用户的最短路径

from langchain.chat_models import ChatOpenAI
from langchain.callbacks import CallbackManager
from ragaai_catalyst.tracers.langchain_callback import LangchainTracer

# 实例化 LangChain tracer
tracer = LangchainTracer(output_path="./my_traces")

# 通过 callback manager 接入
llm = ChatOpenAI(
    model="gpt-4o-mini",
    callbacks=CallbackManager([tracer])
)

# 业务调用 —— 自动被 tracer 捕获
response = llm.invoke("Hello, world!")

13.4 Production 部署 checklist

✅ 申请 access_key/secret_key（Profile → Authenticate → Generate New Key）
✅ 用环境变量管理 Key，不要写死在代码
✅ Guardrails deployment 必须配合 fail_condition + alternate_response
✅ Token 用量监控：定期跑 get_results() 看 total_cost / total_tokens
✅ Red-teaming 至少季度跑一次，新 detector 上线前必须回归

十四、趋势 + 总结

14.1 三大趋势判断

趋势一：Agent 时代的「可观测」必须内嵌业务语义
2025 年大家还在争论「OTel 是否能描述 LLM span」，2026 年这个问题已经被 RagaAI Catalyst 这类项目回答：把 LLMComponent / AgentComponent / ToolComponent 做成显式业务类，而不是泛化的 span。任何忽视业务语义的可观测方案，注定要被业务方抛弃。

趋势二：从「观测」走向「治理」
单看 trace 没有用——只有把「能跑 metric」「能跑 red-team」「能跑 guardrail」全部闭环才是企业真正需要的。RagaAI Catalyst 的 8 大模块一体化设计，正是对「LLMOps = Observability + Eval + Governance」这一公式的工程化回答。

趋势三：场景化安全测试成为红队标配
传统红队靠「随机 prompt 撞库」，RagaAI Catalyst 的 scenario_generator + test_case_generator 用 LLM 自动生成攻击场景，让红队测试从「靠灵感」变成「靠工程」——这是 AI 安全工程化的关键一步。

14.2 工程经验提炼

装饰器一定要”扁平”：@trace_llm 包住的函数必须是独立 LLM 调用（不能嵌套另一个 @trace_llm），否则 trace 树会乱。RagaAI Catalyst 的 warning 提示（COMPONENT DATA INCOMPLETE）就是为这个错误兜底
auto_instrumentation 配置按需开启：调试阶段开全部 7 类，生产环境只开 llm/tool/agent（其他 4 类网络/文件 I/O/用户交互会带来 30%+ 性能开销）
Cost 数据要持久化：每次 trace 都会算 total_cost 但不会自动汇总到 dashboard，必须定期 evaluation.get_results() 拉数据自建看板
Guardrail 阈值先松后紧：上线时 threshold 设为 0.5（宽松），跑一周数据后再调到 0.85（严格），避免一上线就大量拦截

14.3 一句话总结

RagaAI Catalyst 是 2026 年少见的**「业务语义优先 + 平台聚合优先 + 全生命周期覆盖」的 LLM/Agent 治理 SDK。它可能不是 OTel 兼容最优雅的方案（这一点 Logfire 更好），也不是 prompt 管理最灵活的（Langfuse 也强），但它把 8 大模块用同一个客户端 + 同一个 Token 串起来**的设计哲学，是值得所有 LLM 基础设施项目借鉴的工程化范式。

附录：关键资源

GitHub: https://github.com/raga-ai-hub/RagaAI-Catalyst
PyPI: https://pypi.org/project/ragaai-catalyst/
平台入口: https://catalyst.raga.ai/
官方文档: https://docs.raga.ai/catalyst
License: Apache-2.0
核心源文件路径:
- ragaai_catalyst/__init__.py:1-34 —— 8 大模块统一导出
- ragaai_catalyst/ragaai_catalyst.py:12-469 —— 平台客户端基类
- ragaai_catalyst/tracers/agentic_tracing/tracers/main_tracer.py:48-397 —— AgenticTracing 主类
- ragaai_catalyst/tracers/agentic_tracing/data/data_structure.py:1-295 —— 核心 dataclass
- ragaai_catalyst/tracers/agentic_tracing/tracers/agent_tracer.py:23-687 —— Agent 装饰器实现
- ragaai_catalyst/tracers/agentic_tracing/tracers/tool_tracer.py:29-557 —— Tool wrapt patch
- ragaai_catalyst/evaluation.py:16-520 —— Evaluation + schema mapping
- ragaai_catalyst/guardrails_manager.py:10-324 —— GuardrailsManager + Fail Condition