使用 AdventureWorks 数据集的 Fabric 数据代理程序示例（预览版）

本文介绍如何使用 lakehouse 作为示例数据源在 Microsoft Fabric 中设置数据代理。我们首先创建并填充一个 Lakehouse，然后创建一个 Fabric 数据代理并将该 lakehouse 添加到其中。如果已有Power BI语义模型，请确保具有通过数据代理与之交互的读取权限（仅需要写入权限才能修改语义模型或使用 AI 准备等功能）。对于仓库、KQL 数据库或本体，请遵循相同的步骤，改为选择该源。虽然本演练使用的是 Lakehouse，但其余数据来源的模式相同，仅在于数据源的选择不同。

重要

此功能目前为预览版。

先决条件

付费的 F2 或更高版本的 Fabric 容量，或启用了 Microsoft Fabric 的 Power BI 每容量（P1 或更高）容量。
已启用Fabric数据代理租户设置，包括 容量可指定为Fabric Copilot容量设置。
启用了 AI 的跨地理位置处理。
已启用适用于 AI 的跨地理位置存储。
至少包括一个带有数据的资源：仓库、湖仓、一个或多个 Power BI 语义模型、KQL 数据库或本体论。
通过 XMLA 终结点租户交换机启用的 Power BI 语义模型功能适用于 Power BI 语义模型数据源。
对于与数据代理一起使用的Power BI语义模型，请确保通过代理交互的用户对语义模型具有读取权限。进行交互不需要工作区成员身份或构建权限。

重要

确保在 Power BI 管理门户中启用独立 Copilot 体验（租户设置> Copilot > 独立 Copilot 体验）。如果未启用，即使其他 Copilot 租户开关已打开，也无法在 Copilot 场景中使用数据代理。有关详细信息，请参阅Power BI租户设置中的 Copilot。

使用 AdventureWorksLH 创建湖屋

首先，创建湖屋并使用必要的数据进行填充。

如果在湖仓（或仓库）中已有 AdventureWorksLH 实例，则可以跳过此步骤。如果没有，可以使用 Fabric 笔记本中的以下步骤来导入 lakehouse 中的数据。

在要在其中创建Fabric数据代理的工作区中创建新笔记本。
在资源管理器窗格的左侧，选择“+ 数据源”。使用此选项，可以添加现有的湖屋或创建新湖屋。为清楚起见，请创建新湖屋并为其分配名称。

在顶部单元格中，添加以下代码片段：

import pandas as pd
from tqdm.auto import tqdm
base = "https://synapseaisolutionsa.z13.web.core.windows.net/data/AdventureWorks"

# load list of tables
df_tables = pd.read_csv(f"{base}/adventureworks.csv", names=["table"])

for table in (pbar := tqdm(df_tables['table'].values)):
    pbar.set_description(f"Uploading {table} to lakehouse")

    # download
    df = pd.read_parquet(f"{base}/{table}.parquet")

    # save as lakehouse table
    spark.createDataFrame(df).write.mode('overwrite').saveAsTable(table)

选择“全部运行”。

几分钟后，系统将对湖屋填充必要的数据。

注意

继续运行的笔记本（例如，由于意外的死循环或持续轮询）可能会无限期地消耗Fabric的容量。数据加载完成后，请停止任何活动单元格并结束笔记本会话（笔记本工具栏 > 停止会话），如果不再需要它。避免在没有超时的情况下添加长时间运行的循环。

创建Fabric数据代理

若要创建新的Fabric数据代理，请导航到工作区，然后选择“+ 新建项”按钮，如以下屏幕截图所示：

在“所有项”选项卡中，搜索Fabric 数据代理，以找到适当的选项。选择后，系统会提示你提供Fabric数据代理的名称，如以下屏幕截图所示：

截图显示Fabric数据代理名称提供的位置。

输入名称后，请继续执行以下步骤，使Fabric数据代理符合特定要求。

选择数据

选择在上一步中创建的湖屋，然后选择“添加”，如以下屏幕截图所示：

将 lakehouse 添加为数据源后，在 Fabric 数据代理页面左侧的 Explorer 窗格中会显示 lakehouse 的名称。选择该湖屋以查看所有可用的表。使用复选框选择要提供给 AI 的表。对于此方案，请选择以下表：

dimcustomer
dimdate
dimgeography
dimproduct
dimproductcategory
dimpromotion
dimreseller
dimsalesterritory
factinternetsales
factresellersales

数据代理中语义模型的权限

用户只需对Power BI语义模型具有读取权限，才能将其添加到数据代理，并通过代理提问。通过数据代理进行交互不需要工作区访问权限（成员角色）和“构建”权限。只有在需要修改语义模型或使用诸如“Prep for AI”之类的功能时，才需要写入权限。

此权限更改仅适用于通过数据代理的交互。其他访问模式（例如，在 Excel 中分析或直接报告创建）遵循标准 Power BI 权限。

提供指示

若要添加说明，请选择 “数据代理说明 ”按钮以打开右侧的说明窗格。可以添加以下说明。

AdventureWorksLH 数据源包含来自三个表的信息：

dimcustomer，用于获取详细的客户人口统计数据和联系信息
dimdate，用于与日期相关的数据 - 例如日历和会计信息
dimgeography，用于地理详细信息，包括城市名称和国家/地区代码。

使用此数据源进行涉及客户详细信息、基于时间的事件和地理位置的查询和分析。

提供示例

若要添加示例查询，请选择 示例查询 按钮打开右侧的示例查询窗格。此窗格提供用于为所有受支持的数据源添加或编辑示例查询的选项。对于每个数据源，可以选择 添加或编辑示例查询 输入相关示例，如以下屏幕截图所示：

在这里，应为创建的 Lakehouse 数据源添加示例查询。

Question: Calculate the average percentage increase in sales amount for repeat purchases for every zipcode. Repeat purchase is a purchase subsequent to the first purchase (the average should always be computed relative to the first purchase)

SELECT AVG((s.SalesAmount - first_purchase.SalesAmount) / first_purchase.SalesAmount * 100) AS AvgPercentageIncrease
FROM factinternetsales s
INNER JOIN dimcustomer c ON s.CustomerKey = c.CustomerKey
INNER JOIN dimgeography g ON c.GeographyKey = g.GeographyKey
INNER JOIN (
    SELECT *
    FROM (
        SELECT
            CustomerKey,
            SalesAmount,
            OrderDate,
            ROW_NUMBER() OVER (PARTITION BY CustomerKey ORDER BY OrderDate) AS RowNumber
        FROM factinternetsales
    ) AS t
    WHERE RowNumber = 1
) first_purchase ON s.CustomerKey = first_purchase.CustomerKey
WHERE s.OrderDate > first_purchase.OrderDate
GROUP BY g.PostalCode;

Question: Show the monthly total and year-to-date total sales. Order by year and month.

SELECT
    Year,
    Month,
    MonthlySales,
    SUM(MonthlySales) OVER (PARTITION BY Year ORDER BY Year, Month ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS CumulativeTotal
FROM (
    SELECT
       YEAR(OrderDate) AS Year,
       MONTH(OrderDate) AS Month,
       SUM(SalesAmount) AS MonthlySales
    FROM factinternetsales
    GROUP BY YEAR(OrderDate), MONTH(OrderDate)
) AS t

显示添加 SQL 示例的

注释

目前不支持为 Power BI 语义模型数据源添加示例查询与问题配对。

测试和修改Fabric数据代理

配置Fabric数据代理后，添加了Fabric数据代理说明，并为 Lakehouse 提供了示例查询，可以通过提问和接收答案来与之交互。继续测试时，可以添加更多示例并优化说明，进一步提高Fabric数据代理的性能。与同事协作，收集反馈，并根据他们的输入，确保提供的示例查询和说明符合他们想要提问的问题类型。

发布Fabric数据代理程序

在验证Fabric数据代理的性能后，您可以决定发布它，然后与想要针对数据进行问答的同事共享。在这种情况下，请选择“发布”，如以下屏幕截图所示：

“发布数据代理”框将会打开，如以下屏幕截图所示：

在此框中，选择Publish发布Fabric数据代理。 Fabric数据代理的已发布 URL 将显示，如以下屏幕截图所示：

在 Power BI Copilot 中使用Fabric数据代理

可以在发布Power BI中使用Copilot与Fabric数据代理进行交互。使用Power BI中的Copilot，可以直接使用数据代理和其他项（例如报表或语义模型），而无需在它们之间切换。

选择左侧导航窗格中的 Copilot 按钮，在Power BI中打开Copilot。接下来，选择 在底部的文本框中添加项目，以便获得更好的结果，来添加数据代理。在打开的窗口中选择 数据代理 。您只能看到自己有权访问的数据管理员。选择所需的数据代理，然后选择“ 确认”。此示例演示如何使用单个数据代理，但可以添加更多项，例如其他数据代理、报表或语义模型。以下屏幕截图演示了单个数据代理的步骤：

当数据代理包含Power BI语义模型时，用户只需对该语义模型具有读取权限才能通过 Copilot 与之交互;不需要工作区访问。语义模型更改和 AI 准备仍需要写入权限。

在将数据代理程序添加到 Power BI 的 Copilot 后，你可以询问与 Fabric 数据代理程序相关的任何问题，如以下屏幕截图所示：

以编程方式使用Fabric数据代理

可以在Fabric笔记本中以编程方式使用Fabric数据代理。若要确定Fabric数据代理是否具有已发布的 URL 值，请选择 Settings，如以下屏幕截图所示：

发布Fabric数据代理之前，它没有已发布的 URL 值，如以下屏幕截图所示：

如果以前尚未发布Fabric数据代理，可以按照前面的步骤中的说明发布它。然后，可以复制已发布的 URL 并将其用于Fabric笔记本。这样，可以通过调用Fabric笔记本中的Fabric数据代理 API 来查询Fabric数据代理。将复制的 URL 粘贴到此代码片段中。然后，将问题替换为与Fabric数据代理相关的任何查询。此示例使用 \<generic published URL value\> 作为 URL。

重要

以编程方式调用数据代理时，实现：

轮询超时（请参阅以下示例），以避免无限期循环。
最小轮询频率（从 2-5 秒开始;仅在需要时增加）。
完成后清理已创建的线程或资源。
完成后关闭笔记本会话以释放 Fabric 容量。

注释

如果这些确切版本过时，请将版本引脚（openai、synapseml、pandas、tqdm）调整为Fabric运行时的最新验证版本。

%pip install "openai==1.70.0"
%pip install "synapseml==1.0.5"  # Required for synapse.ml.mlflow (update version as needed)
%pip install pandas tqdm  # Skip if already available in the Fabric runtime

import typing as t
import time
import uuid

# OpenAI SDK internals
from openai import OpenAI
from openai._models import FinalRequestOptions
from openai._types import Omit
from openai._utils import is_given

# SynapseML helper for env config
from synapse.ml.mlflow import get_mlflow_env_config

# Removed unused imports: requests, json, pprint, APIStatusError, SynapseTokenProvider
 
base_url = "https://<generic published base URL value>"
question = "What data sources do you have access to?"

configs = get_mlflow_env_config()

# Create OpenAI Client
class FabricOpenAI(OpenAI):
    def __init__(
        self,
        api_version: str ="2024-05-01-preview",
        **kwargs: t.Any,
    ) -> None:
        self.api_version = api_version
        default_query = kwargs.pop("default_query", {})
        default_query["api-version"] = self.api_version
        super().__init__(
            api_key="",
            base_url=base_url,
            default_query=default_query,
            **kwargs,
        )
    
    def _prepare_options(self, options: FinalRequestOptions) -> None:
        headers: dict[str, str | Omit] = (
            {**options.headers} if is_given(options.headers) else {}
        )
        options.headers = headers
        headers["Authorization"] = f"Bearer {configs.driver_aad_token}"
        if "Accept" not in headers:
            headers["Accept"] = "application/json"
        if "ActivityId" not in headers:
            correlation_id = str(uuid.uuid4())
            headers["ActivityId"] = correlation_id

        return super()._prepare_options(options)

# Pretty printing helper
def pretty_print(messages):
    print("---Conversation---")
    for m in messages:
        print(f"{m.role}: {m.content[0].text.value}")
    print()

fabric_client = FabricOpenAI()
# Create assistant
assistant = fabric_client.beta.assistants.create(model="not used")
# Create thread
thread = fabric_client.beta.threads.create()
# Create message on thread
message = fabric_client.beta.threads.messages.create(thread_id=thread.id, role="user", content=question)
# Create run
run = fabric_client.beta.threads.runs.create(thread_id=thread.id, assistant_id=assistant.id)

# Wait for run to complete (avoid indefinite loop)
terminal_states = {"completed", "failed", "cancelled", "requires_action"}
poll_interval = 2
timeout_seconds = 300  # Adjust based on expected workload
start_time = time.time()

while run.status not in terminal_states:
    if time.time() - start_time > timeout_seconds:
        raise TimeoutError(f"Run polling exceeded {timeout_seconds} seconds (last status={run.status})")
    run = fabric_client.beta.threads.runs.retrieve(
        thread_id=thread.id,
        run_id=run.id,
    )
    print(run.status)
    time.sleep(poll_interval)

if run.status != "completed":
    print(f"Run finished with status: {run.status}")

# Print messages
response = fabric_client.beta.threads.messages.list(thread_id=thread.id, order="asc")
pretty_print(response)

# Delete thread
fabric_client.beta.threads.delete(thread_id=thread.id)

反馈

此页面是否有帮助？

Last updated on 2026-04-10