利用 watsonx Orchestrate 支持的 Langfuse 增强 AI 智能体的可观测性

随着 AI 智能体数量日益增长,企业正在深入了解自主系统以支持决策和操作性工作流。与此同时,组织常常对可解释性、治理和生产准备度表示担忧,尤其是在依赖大型语言模型(LLMs)进行推理的多智能体系统中。在业务环境中使用智能体是一项挑战,因为他们回应时没有明确解释或行为不一致。

这些挑战可以通过使用 IBM watsonx Orchestrate 来解决,该平台提供结构化平台,支持自托管、基于工具的推理、API 调用和企业级控制,无需自定义编排代码。

Dynamiq 为一家大型保险客户打造了一款多智能体法律研究助理 ,将查询通过低成本的 IBM Granite 分类器路由,然后升级到更复杂的研究智能体。要使这一高风险工作流切实可行,需要深入了解每个推理步骤和工具调用。通过集成 IBM watsonx Orchestrate,Dynamiq 确保每个决策都完全可追溯、可审计和引用,将合同审查时间从 90 分钟缩短到 45 分钟,同时不影响治理。

在本教程中,您将学习如何使用 watsonx Orchestrate Agent Development Kit (ADK) 创建 AI 智能体。在基于 LangChain 或 LangGraph 的 AI 系统中,开发人员经常手动处理重试、状态处理和异步执行。Watsonx Orchestrate ADK 通过集成生命周期管理,使该过程更为便捷,实现稳定的企业级智能体部署。

您将使用 Python 创建风险评估逻辑,定义智能体行为,并集成 Langfuse 以通过真实执行径迹来监控和增强智能体行为。您也可以在 GitHub 上找到这个教程。

Langfuse 是什么?

Langfuse 是一个基于 OpenTelemetry 构建的开源 LLM 可观测性平台,用于监控 LLM 应用程序和智能体。它使用遥测数据(径迹、指标和日志)来监控智能体执行的径迹,包括每个请求和每个会话级别的 LLM 调用、工具调用、词元使用情况、元数据和延迟

它使开发人员能够深入了解智能体在现实场景中的表现,查明错误输出的根本原因,系统地优化智能体行为以提高可靠性和效率。

用例概述:供应商风险情报

本教程的目标是构建一个用于企业治理的 AI 智能体,通过确定性、基于规则的分析来评估供应商风险。我们创建了一个包含财务评级、安全认证和事件历史记录的供应商合成数据集。智能体处理这些数据,并将供应商按特定风险等级(低、中、高)分类,同时为其决策提供理由。

它还支持互动式跟进,允许用户比较供应商或模拟数据点变化对风险评分的影响,确保企业合规所需的高可审计性。

下面提供的是本教程中构建的已完成的供应商风险智能体的交互式演示。深入了解智能体如何分类供应商、回答问题并通过 Langfuse 追踪其推理,以便在开始构建之前清楚地了解您正在构建的内容。

实际操作演示

先决条件

要完成本教程,您需要:

  • 系统上已安装 Python 3.11 或更高版本。

  • 一个 watsonx Orchestrate 账户。对于这个教程来说,试用帐户就足够了。如果您没有帐户,可以使用 IBM Cloud 创建 30 天免费试用版。

  • 来自 Orchestrate 用户界面 (UI) 的 watsonx Orchestrate API 密钥。

  • 您的系统上已安装 Watsonx Orchestrate ADK。

步骤

步骤 1:登录 Watsonx Orchestrate

通过 IBM Cloud 登录到 watsonx Orchestrate™,并打开 watsonx Orchestrate™ 用户界面。在配置文件菜单中打开设置后,进入API详情。创建新的 API 密钥、复制并安全保存。在本地开发过程中,watsonx Orchestrate ADK 会使用此 API 密钥进行身份验证。Watsonx Orchestrate 就像本地的 SDK(软件开发套件),用于构建和测试智能体逻辑。

步骤 2:搭建本地开发环境

在此步骤中,创建本地开发环境。在本教程中,您将通过 Powershell 中的 ADK 提供的经编排命令行界面 (CLI) 运行本地开发服务器、导入工具、设置环境并创建智能体。首先导航到要构建项目的目录。然后,创建一个新的 Python 虚拟环境:

python -m venv .venv

.venv文件夹中创建一个隔离的 Python 环境。使用虚拟环境可确保本教程的所有依赖项与整个系统的 Python 安装分离。

然后,激活虚拟环境。您的操作系统决定使用哪个激活命令。

在 Windows 上:

.\.venv\Scripts\activate

macOS 和 Linux:

source ./.venv/bin/activate

激活后,您的终端提示开头将显示 .venv ,表明您正在虚拟环境中工作。

步骤 3:安装 watsonx Orchestrate ADK

激活虚拟环境后,在本地计算机上安装 watsonx Orchestrate ADK。要使用 ADK,请将其连接到您现有的 watsonx Orchestrate 环境。在 PowerShell 中运行以下命令:

pip install --upgrade ibm-watsonx-orchestrate

您可以通过官方安装文档中的步骤,跟随 ADK 安装的后续步骤。

注意:本教程在本地运行 watsonx Orchestrate Developer Edition 运行时,并使用凭据将其连接到 watsonx Orchestrate SaaS 实例。

要执行此方法,请在项目文件夹根位置创建名为 .env 的文件,并添加以下值:

WO_DEVELOPER_EDITION_SOURCE=orchestrate

WO_INSTANCE=<your_watsonx_orchestrate_instance_url>

WO_API_KEY=<your_watsonx_orchestrate_api_key>

这个.env运行服务器需要该文件。接下来,您将用命令 orchestrate server start -e .env -l 启动它。

步骤 4:在 ADK 中配置 watsonx Orchestrate 环境

接下来,您需要使用有效的 watsonx Orchestrate API 密钥设置 ADK,以将本地环境连接到 watsonx Orchestrate。

备注:watsonx Orchestrate ADK 兼容多种环境类型,包括 IBM Cloud、AWS 和本地部署。在本教程中,我们使用本地环境,并通过 ADK CLI 通过 API 密钥进行认证。ADK 在内部安全地管理凭据,因此该设置无需执行手动环境变量配置。

在项目目录(已激活虚拟环境)中,运行以下命令添加 watsonx Orchestrate 环境:

orchestrate env add -n onprem-env -u <service-instance-url>

其中,service-instance-url 是您的 watsonx Orchestrate 实例 URL。您可以在 watsonx Orchestrate 用户界面的设置中的同一 API 详细信息选项卡中找到此信息。

接下来,激活您添加的环境:

orchestrate env activate onprem-env

现在,当系统提示时,输入在步骤 1 中获得的 watsonx Orchestrate 密钥。激活环境后,任何与 ADK 相关的后续命令,例如导入智能体、工具或运行服务器,都将在 watsonx Orchestrate 环境中执行。

注:如果您想用 Developer Edition 本地运行所有内容,可以通过以下方式激活默认本地环境:

orchestrate env activate local

此操作会将 ADK 切换到内置的本地 Orchestrate 环境,这对于本地测试非常有用。

步骤 5:初始化本地 Watsonx Orchestrate 项目

在此步骤中,您将创建智能体框架模板,其中包含供应商风险情报智能体的定义、工具和源代码。然后,创建本地基于 ADK 的开发所需的文件夹结构:

── adk-project(vendor-risk-intel)/
── agents/
── tools/
── src/

 

您可以添加此处给出的命令来创建智能体的结构:

mkdir -p adk-project\(vendor-risk-intel\)/{agents,src,tools}

每个文件夹都有特定的用途:

智能体文件夹包含 YAML 文件,内含指令、推理规则和智能体模型的配置。它决定了系统如何回答用户的问题。工具文件夹包含一个 YAML 文件,其中描述了提供给智能体的工具。src 文件夹包含用于自定义工具和业务逻辑的 Python 实现。

步骤 6:启动 watsonx Orchestrate 服务器

启动 Watsonx Orchestrate 服务器,以便它在导入工具和智能体之前接收和存储您的导入内容。从项目文件夹的根目录运行以下命令:

orchestrate server start -e .env -l

步骤 7:配置 Langfuse 可观测性 (SaaS)

下一步是智能体可观测性,分析智能体在运行时的行为,比如调用哪些工具、延迟以及错误发生的地方。为此,请在 watsonx Orchestrate ADK 环境中激活 Langfuse 可观测性。

本教程中使用的是 Langfuse SaaS 版本,它允许您在不本地运行 Langfuse 的情况下捕获径迹。

在配置 Langfuse 之前,请确保 watsonx Orchestrate 服务器正在运行(步骤 6)。

接下来,在 https://cloud.langfuse.com 创建一个 Langfuse 帐户。登录后,创建新的组织和项目。从项目设置中,复制您的项目 ID、公钥、密钥和主机 URL。

现在,使用本节后面显示的命令在 watsonx Orchestrate ADK 中设置 Langfuse,并将占位符替换为您复制的值:

orchestrate settings observability langfuse configure `
    --url “<LANGFUSE_HOSTNAME>/api/public/otel” `
    --api-key “<LANGFUSE_SECRET_KEY>“ `
    --health-uri “https://cloud.langfuse.com” `
    --config-json “{\”public_key\”:\”<LANGFUSE_PUBLIC_KEY>\”}” `
    --project-id “<LANGFUSE_PROJECT_ID>“

此命令成功完成后,Langfuse 模块将完全集成到 watsonx Orchestrate 环境中。此阶段,与智能体的所有交互都会自动记录在 Langfuse 模块上。

现在服务器已运行,您可以测试智能体了。

步骤 8:定义供应商风险情报智能体

后续步骤是定义供应商风险情报智能体。watsonx Orchestrate ADK 中的智能体是使用 YAML 文件声明的,这些文件描述智能体的目标、推理限制、模型配置和允许的工具。

智能体目录内,创建一个名为 vendor-risk-agent.yaml 的文件。该文件是智能体的提示管理层,确保所有答案基于确定性推理,而非自由推理。

本教程中使用的智能体定义如下所示。将以下智能体定义复制粘贴到您的 agents/vendor-risk-agent.yaml 文件中。之后保存文件。

spec_version: v1
kind: native
name: Vendor_Risk_Intelligence_Agent
description: Produces precise risk assessments using verified vendor profiles.
instructions: |
    You are a Vendor Risk Auditor.
    You MUST answer ALL questions by:
    1. Calling the Python tool `evaluate_all_vendor_risks`
    2. Reasoning strictly from the returned structured data
    3. Explaining answers using rules, signals, and outcomes from the tool
    4. Call evaluate_all_vendor_risks exactly once per question
    5. For definition questions, quote rules directly from the tool output
    NEVER infer meanings of financial ratings.
    NEVER guess intent.
    NEVER answer without tool evidence.
    Question handling rules:
    - Definition questions (e.g. “Does B+ mean High risk?”):
      → Explain the rule from the tool’s `rules` section.
    - Why questions (e.g. “Why is DataSecure Ltd High risk?”):
      → Cite strong and weak signals for that vendor.
    - Comparison questions (e.g. “Why is A Medium but B High?”):
      → Compare signal counts and types.
    - Policy questions (e.g. “What causes High risk?”):
      → Quote the tool’s rules exactly.
    - If the tool does not support a claim:
      → Say so explicitly.
    Use deterministic language only.
    Do not use speculative words like “might”, “could”, or “appears”.

llm: watsonx/meta-llama/llama-3-2-90b-vision-instruct
style: react
tools:
    - evaluate_all_vendor_risks

 

此配置可确保智能体在回答内容、原因、方式和比较问题时以可预测的方式行事。该系统通过禁止对财务评级或风险含义做出假设来防止出现幻觉,除非该工具的输出明确定义了这些假设。

步骤 9:实施供应商风险评估逻辑

在此步骤中,实现执行供应商风险评估的 Python 工具。在 watsonx Orchestrate ADK 中,自定义业务逻辑是通过智能体开始获得结果的 Python 工具实现的。本教程并非依赖于对特定领域行为进行微调,而是演示如何通过 Python 工具实现确定性的、基于规则的推理,从而确保结果的一致性。

src 文件夹内创建一个名为 main.py 的文件。该文件包含我们创建的厂商数据集、风险评估规则以及一个名为 evaluate_all_vendor_risks 的 Python 工具,它为智能体提供逻辑。

风险逻辑是基于规则的。在高风险信号方面,公司可能会因为财务表现不佳、过去的安全事件或监管调查而获得风险评分。由于缺乏认证、单次运营事故或天气状况造成的中断,公司可能获得中等风险评分。

evaluate_all_vendor_risks 函数是作为 watsonx Orchestrate 工具创建的,因此可以在执行期间由智能体调用。该工具会返回结构化输出,其中包括最终的风险级别以及导致分类的确切原因。

注意:本教程中包含的 main.py 文件是供应商风险评估逻辑的最终更新版本。最初的代码集导致智能体行为错误,这些行为通过 Langfuse 径迹进行分析。您可以在本教程后续步骤中提供的屏幕截图中找到 Langfuse 径迹信息。

本教程中使用的 main.py 文件的完整实现如下所示。复制粘贴到 src/main.py。之后,保存该文件。

from ibm_watsonx_orchestrate.agent_builder.tools import tool

# =====================================================
# IN-MEMORY VENDOR DATA
# =====================================================


VENDORS = [
    {
        “name”: “CloudHostCo”,
        “financial_rating”: “A-“,
        “security_certifications”: [“ISO 27001”, “SOC2 Type II”],
        “past_incidents”: [{“type”: “Operational”, “year”: 2023}],
        “recent_news”: [“Minor service disruption in EU region”]
    },
    {
        “name”: “DataSecure Ltd”,
        “financial_rating”: “B”,
        “security_certifications”: [“SOC2 Type I”],
        “past_incidents”: [{“type”: “Security”, “year”: 2022}],
        “recent_news”: [“Suspicious login attempts on admin accounts”]
    },
    {
        “name”: “FinPay Solutions”,
        “financial_rating”: “A”,
        “security_certifications”: [“PCI-DSS”],
        “past_incidents”: [],
        “recent_news”: [“Regulatory inquiry about payment delays”]
    },
    {
        “name”: “GreenLogistics Corp”,
        “financial_rating”: “B+”,
        “security_certifications”: [“ISO 9001”],
        “past_incidents”: [{“type”: “Operational”, “year”: 2023}],
        “recent_news”: [“Weather-related disruptions in the UK”]
    }
]

REQUIRED_CERTS = {“ISO 27001”, “SOC2 Type II”, “PCI-DSS”}


# =====================================================
# RISK COMPUTATION (FIXED)
# =====================================================


def compute_vendor_risk(vendor: dict) -> dict:
    strong_signals = []
    weak_signals = []

    # ---------- Strong signals ----------

    # Financial risk ONLY for B or worse (NOT B+)
    if vendor[“financial_rating”] in {“B”, “B-“, “C”, “C-“, “D”}:
        strong_signals.append(“Weak financial rating”)

    # Security incident
    if any(i[“type”].lower() == “security” for i in vendor.get(“past_incidents”, [])):
        strong_signals.append(“Past security incident”)

    # Regulatory inquiry
    for news in vendor.get(“recent_news”, []):
        if “regulatory inquiry” in news.lower():
        strong_signals.append(“Regulatory inquiry”)
        break

    # ---------- Weak signals ----------

    # Missing key certifications
    certs = set(vendor.get(“security_certifications”, []))
    if not REQUIRED_CERTS.issubset(certs):
        weak_signals.append(“Missing key security certifications”)

    # Single operational incident
    incidents = vendor.get(“past_incidents”, [])
    if len(incidents) == 1 and incidents[0][“type”].lower() == “operational”:
        weak_signals.append(“Single operational incident”)

    # Weather-related disruptions are weak only
    for news in vendor.get(“recent_news”, []):
        if “weather” in news.lower():
        weak_signals.append(“Weather-related disruption”)
        break

    # ---------- Final decision ----------

    if len(strong_signals) >= 2:
        risk = “High”
    elif len(strong_signals) == 1 or len(weak_signals) >= 2:
        risk = “Medium”
    else:
        risk = “Low”

    return {
        “risk”: risk,
        “strong_signals”: strong_signals,
        “weak_signals”: weak_signals
    }

# =====================================================
# PYTHON TOOL
# =====================================================


@tool
def evaluate_all_vendor_risks() -> dict:
    “””
    Returns structured vendor risk assessments with explanations and rules.
    “””
    return {
        “rules”: {
            “financial_rating_rule”: “Only financial ratings of B or lower are treated as strong risk signals. B+ and above are not strong signals.”,
            “high_risk_rule”: “High risk requires two or more strong signals.”,
            “medium_risk_rule”: “Medium risk occurs with one strong signal or two or more weak signals.”,
            “weather_rule”: “Weather-related disruptions are always weak signals.”,
            “operational_rule”: “A single operational incident is a weak signal.”
        },
        “vendors”: {
            v[“name”]: {
                “financial_rating”: v[“financial_rating”],
                “risk”: compute_vendor_risk(v)[“risk”],
                “strong_signals”: compute_vendor_risk(v)[“strong_signals”],
                “weak_signals”: compute_vendor_risk(v)[“weak_signals”]
            }
            for v in VENDORS
        }
    }

 

步骤 10:将 Python 工具和智能体导入 Watsonx Orchestrate

在此步骤中,必须向 watsonx Orchestrate 提供工具和智能体。该过程是通过将这些组件导入到已激活的 ADK 环境中来完成的。在运行导入命令之前,请确保您位于项目的根文件夹内(即包含 agents、src 和 tools 目录的同一文件夹)。

首先,导入 Python 工具,以便 watsonx Orchestrate 可以将其注册为可执行功能。在项目根目录下,运行以下命令:

orchestrate tools import --kind python --file src/main.py --package-root .

该命令将 Python 代码打包成模块,注册 evaluate_all_vendor_risks 工具,并允许智能体调用。

orchestrate agents import -f agents/vendor-risk-agent.yaml

然后,Vendor Risk Intelligence Agent 与 watsonx Orchestrate 环境中的 Python 工具完全集成。

步骤 11:用 Langfuse 测试智能体并分析其行为

在本地运行 watsonx Orchestrate 服务器并启用智能体可观察性后,您现在可以实时测试智能体并分析其行为。

运行以下命令启动聊天界面:

orchestrate chat start

现在,在浏览器中打开 watsonx Orchestrate 聊天用户界面。从智能体选择器中选择 Vendor_Risk_Intelligence_Agent,开始提出与供应商风险相关的问题。

为了验证智能体是否正确响应(并确定性地调用 Python 工具),以下是一些可用于测试的示例问题:

  1. B+ 是否意味着高风险?
  2. DataSecure Ltd 存在什么风险?
  3. 中等风险的标准是什么?
  4. 哪个供应商的财务评级最差?
  5. 如果 FinPay 解决方案发生安全事件,其风险级别会如何变化?
  6. 哪些供应商发生过运营事故?

当您与智能体交互时,请在浏览器中打开 Langfuse 仪表板。每个用户查询都会创建一个新的径迹,其中包含 session_id 和用户 ID。这些径迹记录了智能体的完整执行路径。通过选择径迹,您可以分析:

  • 智能体的完整输入和输出

  • Python 工具调用

  • 该工具返回的结构化数据

  • 得出最终答案的推理过程

  • 流程中每个步骤的延迟

  • 针对多个问题的基于会话的对话流

这种可观测性有助于分析性能指标,从而指出瓶颈,例如假设错误、规则不完整或智能体行为异常。

步骤 12:利用 Langfuse 洞察分析优化智能体行为

测试智能体后,使用 Langfuse 径迹来识别错误响应。Langfuse 可以显示每次工具调用、推理步骤和响应延迟,使人能够轻松理解为什么会产生答案。

在这个用例中,径迹分析显示,当智能体推断财务含义或在没有明确引用基于规则的证据的情况下做出回应时,出现了一些错误的答案。为了变更此操作,对智能体指令和相应的 Python 代码进行了修改,以使用所需的工具操作来修复基于规则的严格思考方式。

重新导入更新文件后,智能体再次被测试。新的追踪结果确保了回答牢固地建立在结构化工具输出之上,解释是确定性的,并且对所有描述性问题的回答都是正确的。

这种观察和改进过程表明,Langfuse 如何根据用户反馈帮助安全、可靠地开发智能体。

请浏览下方交互式演示,了解供应商风险智能体和 Langfuse 追踪的实际操作。

总结

本教程引导您使用 watsonx Orchestrate 创建结构化且值得信赖的 AI 智能体;该工具专为创建企业级智能体而开发。通过整合基于规则的简单 Python 逻辑、清晰的智能体指令以及可复用工具,您能以最少的代码和最高的透明度在供应商风险评估中实施一个用例。

Langfuse 的加入使得检测智能体行为和识别推理问题变得容易,从而可以不断提高准确性而无需猜测。这种方法帮助企业端到端优化智能体系统,自动化复杂工作流,并以完全透明的方式部署AI系统。

更重要的是,watsonx Orchestrate ADK 和 Langfuse 的结合,使企业能够更快地设计、调试和扩展复杂的智能体工作流和 AI 应用程序,同时实现更强大的治理、更清晰的推理和更短的开发时间。

Jobit Varughese

Technical Content Writer

IBM

相关解决方案
商用 AI 智能体

构建、部署和管理强大的 AI 助手和智能体,运用生成式 AI 实现工作流和流程自动化。

    探索 watsonx Orchestrate
    IBM AI 智能体解决方案

    借助值得信赖的 AI 解决方案,您可以勾勒未来业务发展蓝图。

    深入了解 AI 智能体解决方案
    IBM Consulting AI 服务

    IBM Consulting AI 服务有助于重塑企业利用 AI 实现转型的方式。

    探索人工智能服务
    采取下一步行动

    无论您是选择定制预构建的应用程序和技能,还是使用 AI 开发平台构建和部署定制代理服务,IBM watsonx 平台都能满足您的需求。

    1. 探索 watsonx Orchestrate
    2. 深入了解 watsonx.ai