将 BigQuery 与 Workers AI 配合使用

开始使用 Workers AI 的最简单方式是在 Multi-modal Playground ↗ 和 LLM playground ↗ 中试用。如果决定让代码与 Workers AI 集成，可以使用其 REST API 端点或 Worker 绑定（binding）。

但数据呢？如果想让这些模型摄取存储在 Cloudflare 外部的数据怎么办？

在本教程中，你将学习如何将 Google BigQuery 的数据引入 Cloudflare Worker，以便用作 Workers AI 模型的输入。

前提条件

你需要：

运行 Hello World 脚本的 Cloudflare Worker 项目。
具有关联密钥 ↗ 文件（已下载）且对 BigQuery 有读取访问权限的 Google Cloud Platform 服务账户 ↗。
访问包含一些测试数据的 BigQuery 表，允许你创建 BigQuery Job Query ↗。建议为本教程创建自己的表，因为采样表 ↗除非克隆到你自己的 GCP namespace，否则无法对其运行 job query。本示例在其 MIT 许可下使用了 Hacker News Corpus ↗。

1. 设置 Cloudflare Worker

要将数据引入 Cloudflare 并输入 Workers AI，你将使用 Cloudflare Worker。如果尚未创建，请查看我们的快速入门教程。

按照创建 Worker 的步骤操作后，新的 Worker 项目中应有以下代码：

export default {
	async fetch(request, env, ctx) {
		return new Response("Hello World!");
	},
};

如果 Worker 项目已成功创建，还应能在控制台运行 npx wrangler dev 在本地运行 Worker：

[wrangler:inf] Ready on http://localhost:8787

在 http://localhost:8787/ 打开浏览器标签页查看已部署的 Worker。请注意，你的情况下端口 8787 可能不同。

浏览器中应显示 Hello World!：

Hello World!

如果此步骤遇到问题，请查看 Worker 快速入门指南。

2. 将 GCP 服务密钥作为 Secrets 导入 Worker

既然已确认 Worker 创建成功，你需要引用本教程前提条件部分创建的 Google Cloud Platform 服务密钥。

从 Google Cloud Platform 下载的密钥 JSON 文件应具有以下格式：

{
	"type": "service_account",
	"project_id": "<your_project_id>",
	"private_key_id": "<your_private_key_id>",
	"private_key": "<your_private_key>",
	"client_email": "<your_service_account_id>@<your_project_id>.iam.gserviceaccount.com",
	"client_id": "<your_oauth2_client_id>",
	"auth_uri": "https://accounts.google.com/o/oauth2/auth",
	"token_uri": "https://oauth2.googleapis.com/token",
	"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
	"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/<your_service_account_id>%40<your_project_id>.iam.gserviceaccount.com",
	"universe_domain": "googleapis.com"
}

在本教程中，你只需以下字段的值：client_email、private_key、private_key_id 和 project_id。

你不会在 Worker 中以明文存储此信息，而是使用 Secrets 确保未加密内容仅可通过 Worker 本身访问。

将 JSON 文件中的这些值导入 Secrets，从 JSON 密钥文件中名为 client_email 的字段开始，我们将其称为 BQ_CLIENT_EMAIL（你可以使用其他变量名）：

npx wrangler secret put BQ_CLIENT_EMAIL

系统会要求你输入 secret 值，即 JSON 密钥文件中 client_email 字段的值。

如果 secret 上传成功，将显示以下消息：

✨ Success! Uploaded secret BQ_CLIENT_EMAIL

现在导入剩余三个字段的 secret；分别为 private_key、private_key_id 和 project_id，命名为 BQ_PRIVATE_KEY、BQ_PRIVATE_KEY_ID 和 BQ_PROJECT_ID：

npx wrangler secret put BQ_PRIVATE_KEY

npx wrangler secret put BQ_PRIVATE_KEY_ID

npx wrangler secret put BQ_PROJECT_ID

此时，你已成功将从 Google Cloud Platform 下载的 JSON 密钥文件中的字段导入 Cloudflare Secrets，供 Worker 使用。

Secrets 仅在部署后才对 Worker 可用。要在开发期间使用，请创建 .dev.vars 文件在本地存储这些凭据并将其引用为环境变量。

你的 dev.vars 文件应如下所示：

BQ_CLIENT_EMAIL="<your_service_account_id>@<your_project_id>.iam.gserviceaccount.com"
BQ_CLIENT_KEY="-----BEGIN PRIVATE KEY-----<content_of_your_private_key>-----END PRIVATE KEY-----\n"
BQ_PRIVATE_KEY_ID="<your_private_key_id>"
BQ_PROJECT_ID="<your_project_id>"

确保将 .dev.vars 包含在项目 .gitignore 文件中，以防止使用版本控制时将凭据上传到仓库。

通过将 secret 值记录到控制台输出，检查 src/index.js 中是否正确加载 secret，如下所示：

export default {
	async fetch(request, env, ctx) {
		console.log("BQ_CLIENT_EMAIL: ", env.BQ_CLIENT_EMAIL);
		console.log("BQ_PRIVATE_KEY: ", env.BQ_PRIVATE_KEY);
		console.log("BQ_PRIVATE_KEY_ID: ", env.BQ_PRIVATE_KEY_ID);
		console.log("BQ_PROJECT_ID: ", env.BQ_PROJECT_ID);
		return new Response("Hello World!");
	},
};

重启 Worker 并运行 npx wrangler dev。应看到服务器现在提及新添加的变量：

Using vars defined in .dev.vars
Your worker has access to the following bindings:
- Vars:
  - BQ_CLIENT_EMAIL: "(hidden)"
  - BQ_PRIVATE_KEY: "(hidden)"
  - BQ_PRIVATE_KEY_ID: "(hidden)"
  - BQ_PROJECT_ID: "(hidden)"
[wrangler:inf] Ready on http://localhost:8787

如果在浏览器中打开 http://localhost:8787，应看到变量值显示在运行 npx wrangler dev 命令的控制台中，而浏览器窗口中仍只显示 Hello World! 文本。

你现在可以从 Worker 访问 GCP 凭据。接下来，将安装库以帮助创建与 GCP API 交互所需的 JSON Web Token。

3. 安装处理 JWT 操作的库

要与 BigQuery REST API 交互，需要使用上一步加载到 Worker secrets 中的凭据生成 JSON Web Token ↗ 来验证请求。

在本教程中，你将使用 jose ↗ 库进行 JWT 相关操作。在控制台运行以下命令安装：

npm i jose

要验证安装是否成功，可以运行 npm list 列出所有已安装包，检查是否添加了 jose 依赖：

<project_name>@0.0.0
/<path_to_your_project>/<project_name>
├── @cloudflare/vitest-pool-workers@0.4.29
├── jose@5.9.2
├── vitest@1.5.0
└── wrangler@3.75.0

4. 生成 JSON web token

既然已安装 jose 库，现在将其导入并在代码中添加生成签名 JSON Web Token（JWT）的函数：

import * as jose from 'jose';
...
const generateBQJWT = async (aCryptoKey, env) => {
const algorithm = "RS256";
const audience = "https://bigquery.googleapis.com/";
const expiryAt = (new Date().valueOf() / 1000);
	const privateKey = await jose.importPKCS8(env.BQ_PRIVATE_KEY, algorithm);

	// Generate signed JSON Web Token (JWT)
	return new jose.SignJWT()
    	.setProtectedHeader({
        	typ: 'JWT',
        	alg: algorithm,
        	kid: env.BQ_PRIVATE_KEY_ID
    	})
    	.setIssuer(env.BQ_CLIENT_EMAIL)
    	.setSubject(env.BQ_CLIENT_EMAIL)
    	.setAudience(audience)
    	.setExpirationTime(expiryAt)
    	.setIssuedAt()
    	.sign(privateKey)
}

export default {
	async fetch(request, env, ctx) {
       ...
// Create JWT to authenticate the BigQuery API call
    	let bqJWT;
    	try {
        	bqJWT = await generateBQJWT(env);
    	} catch (e) {
        	return new Response('An error has occurred while generating the JWT', { status: 500 })
    	}
	},
       ...
};

既然已创建 JWT，现在是时候调用 BigQuery API 获取一些数据。

5. 向 Google BigQuery 发起经过身份验证的请求

使用上一步创建的 JWT token，向 BigQuery API 发起 API 请求以从表检索数据。

你现在将查询本教程早些时候在 BigQuery 中创建的表。此示例使用在其 MIT 许可下使用并上传到 BigQuery 的 Hacker News Corpus ↗ 采样版本。

const queryBQ = async (bqJWT, path) => {
	const bqEndpoint = `https://bigquery.googleapis.com${path}`
	// In this example, text is a field in the BigQuery table that is being queried (hn.news_sampled)
	const query = 'SELECT text FROM hn.news_sampled LIMIT 3';
	const response = await fetch(bqEndpoint, {
    	method: "POST",
    	body: JSON.stringify({
        	"query": query
    	}),
    	headers: {
        	Authorization: `Bearer ${bqJWT}`
    	}
	})
	return response.json()
}
...
export default {
	async fetch(request, env, ctx) {
		...
    		let ticketInfo;
    		try {
    		ticketInfo = await queryBQ(bqJWT);
    	} catch (e) {
        	return new Response('An error has occurred while querying BQ', { status: 500 });
    	}
	...
	},
};

拥有 BigQuery 的原始行数据意味着你现在可以将其格式化为类 JSON 样式。

6. 格式化查询结果

既然已从 BigQuery 检索数据，BigQuery API 响应应类似如下：

{
	...
	"schema": {
    	"fields": [
        	{
            	"name": "title",
            	"type": "STRING",
            	"mode": "NULLABLE"
        	},
        	{
            	"name": "text",
            	"type": "STRING",
            	"mode": "NULLABLE"
        	}
    	]
	},
	...
	"rows": [
    	{
        	"f": [
            	{
                	"v": "<some_value>"
            	},
            	{
                	"v": "<some_value>"
            	}
        	]
    	},
    	{
        	"f": [
            	{
                	"v": "<some_value>"
            	},
            	{
                	"v": "<some_value>"
            	}
        	]
    	},
    	{
        	"f": [
            	{
                	"v": "<some_value>"
            	},
            	{
                	"v": "<some_value>"
            	}
        	]
    	}
	],
	...
}

迭代结果时，此格式可能难以阅读和处理。因此你现在将实现一个函数，将 schema 映射到每个单独的值， resulting 输出将更易于阅读，如下所示。每行对应数组中的一个对象。

[
	{
		title: "<some_value>",
		text: "<some_value>",
	},
	{
		title: "<some_value>",
		text: "<some_value>",
	},
	{
		title: "<some_value>",
		text: "<some_value>",
	},
];

创建 formatRows 函数，接收 BigQuery 响应体返回的行和字段，并返回具有命名字段的对象结果数组。

const formatRows = (rowsWithoutFieldNames, fields) => {
	// Index to fieldName
	const fieldsByIndex = new Map();

	// Load all fields by name and have their index in the array result as their key
	fields.forEach((field, index) => {
    	fieldsByIndex.set(index, field.name)
	})

	// Iterate through rows
	const rowsWithFieldNames = rowsWithoutFieldNames.map(row => {
    	// Per each row represented by an array f, iterate through the unnamed values and find their field names by searching them in the fieldsByIndex.
    	let newRow = {}
    	row.f.forEach((field, index) => {
        	const fieldName = fieldsByIndex.get(index);
        	if (fieldName) {
		// For every field in a row, add them to newRow
            	newRow = ({ ...newRow, [fieldName]: field.v });
        	}
    	})
    	return newRow
	})

	return rowsWithFieldNames
}

export default {
	async fetch(request, env, ctx) {
		...
    	// Transform output format into array of objects with named fields
    	let formattedResults;

    	if ('rows' in ticketInfo) {
        	formattedResults = formatRows(ticketInfo.rows, ticketInfo.schema.fields);
        	console.log(formattedResults)
    	} else if ('error' in ticketInfo) {
        	return new Response(ticketInfo.error.message, { status: 500 })
    	}
	...
	},
};

7. 将数据输入 Workers AI

既然已将 BigQuery API 的响应转换为结果数组，通过 Workers AI 使用 LLM 生成一些 tag 并附加关联的情感分数：

const generateTags = (data, env) => {
	return env.AI.run("@cf/meta/llama-3.1-8b-instruct", {
    	prompt: `Create three one-word tags for the following text. return only these three tags separated by a comma. don't return text that is not a category.Lowercase only. ${JSON.stringify(data)}`,
	});
}

const generateSentimentScore = (data, env) => {
	return env.AI.run("@cf/meta/llama-3.1-8b-instruct", {
    	prompt: `return a float number between 0 and 1 measuring the sentiment of the following text. 0 being negative and 1 positive. return only the number, no text. ${JSON.stringify(data)}`,
	});
}

// Iterates through values, sends them to an AI handler and encapsulates all responses into a single Promise
const getAIGeneratedContent = (data, env, aiHandler) => {
	let results = data?.map(dataPoint => {
    	return aiHandler(dataPoint, env)
	})
	return Promise.all(results)
}
...
export default {
	async fetch(request, env, ctx) {
		...
let summaries, sentimentScores;
    	try {
        	summaries = await getAIGeneratedContent(formattedResults, env, generateTags);
        	sentimentScores = await getAIGeneratedContent(formattedResults, env, generateSentimentScore)
    	} catch {
        	return new Response('There was an error while generating the text summaries or sentiment scores')
    	}
},

formattedResults = formattedResults?.map((formattedResult, i) => {
        	if (sentimentScores[i].response && summaries[i].response) {
            	return {
                	...formattedResult,
                	'sentiment': parseFloat(sentimentScores[i].response).toFixed(2),
                	'tags': summaries[i].response.split(',').map((result) => result.trim())
            	}
        	}
    	}
};

取消项目中 Wrangler 文件以下行的注释：

{
	"ai": {
		"binding": "AI"
	}
}

[ai]
binding = "AI"

重启本地运行的 Worker，然后访问应用端点：

curl http://localhost:8787

使用 Worker AI 时，可能会要求你登录 Cloudflare 账户并授予 Wrangler（Cloudflare CLI）临时访问权限以使用你的账户。

访问 http://localhost:8787 后，应看到类似以下的输出：

{
  "data": [
	{
  	"text": "You can see a clear spike in submissions right around US Thanksgiving.",
  	"sentiment": "0.61",
  	"tags": [
    	"trends",
    	"submissions",
    	"thanksgiving"
  	]
	},
	{
  	"text": "I didn't test the changes before I published them.  I basically did development on the running server. In fact for about 30 seconds the comments page was broken due to a bug.",
  	"sentiment": "0.35",
  	"tags": [
    	"software",
    	"deployment",
    	"error"
  	]
	},
	{
  	"text": "I second that. As I recall, it's a very enjoyable 700-page brain dump by someone who's really into his subject. The writing has a personal voice; there are lots of asides, dry wit, and typos that suggest restrained editing. The discussion is intelligent and often theoretical (and Bartle is not scared to use mathematical metaphors), but the tone is not academic.",
  	"sentiment": "0.86",
  	"tags": [
    	"review",
    	"game",
    	"design"
  	]
	}
  ]
}

实际值和字段主要取决于第 5 步中执行的查询，然后输入 LLM。

最终结果

各步骤中显示的所有代码合并为 src/index.js 中的以下代码：

import * as jose from "jose";

const generateBQJWT = async (env) => {
	const algorithm = "RS256";
	const audience = "https://bigquery.googleapis.com/";
	const expiryAt = new Date().valueOf() / 1000;
	const privateKey = await jose.importPKCS8(env.BQ_PRIVATE_KEY, algorithm);

	// Generate signed JSON Web Token (JWT)
	return new jose.SignJWT()
		.setProtectedHeader({
			typ: "JWT",
			alg: algorithm,
			kid: env.BQ_PRIVATE_KEY_ID,
		})
		.setIssuer(env.BQ_CLIENT_EMAIL)
		.setSubject(env.BQ_CLIENT_EMAIL)
		.setAudience(audience)
		.setExpirationTime(expiryAt)
		.setIssuedAt()
		.sign(privateKey);
};

const queryBQ = async (bgJWT, path) => {
	const bqEndpoint = `https://bigquery.googleapis.com${path}`;
	const query = "SELECT text FROM hn.news_sampled LIMIT 3";
	const response = await fetch(bqEndpoint, {
		method: "POST",
		body: JSON.stringify({
			query: query,
		}),
		headers: {
			Authorization: `Bearer ${bgJWT}`,
		},
	});
	return response.json();
};

const formatRows = (rowsWithoutFieldNames, fields) => {
	// Index to fieldName
	const fieldsByIndex = new Map();

	fields.forEach((field, index) => {
		fieldsByIndex.set(index, field.name);
	});

	const rowsWithFieldNames = rowsWithoutFieldNames.map((row) => {
		// Map rows into an array of objects with field names
		let newRow = {};
		row.f.forEach((field, index) => {
			const fieldName = fieldsByIndex.get(index);
			if (fieldName) {
				newRow = { ...newRow, [fieldName]: field.v };
			}
		});
		return newRow;
	});

	return rowsWithFieldNames;
};

const generateTags = (data, env) => {
	return env.AI.run("@cf/meta/llama-3.1-8b-instruct", {
		prompt: `Create three one-word tags for the following text. return only these three tags separated by a comma. don't return text that is not a category.Lowercase only. ${JSON.stringify(data)}`,
	});
};

const generateSentimentScore = (data, env) => {
	return env.AI.run("@cf/meta/llama-3.1-8b-instruct", {
		prompt: `return a float number between 0 and 1 measuring the sentiment of the following text. 0 being negative and 1 positive. return only the number, no text. ${JSON.stringify(data)}`,
	});
};

const getAIGeneratedContent = (data, env, aiHandler) => {
	let results = data?.map((dataPoint) => {
		return aiHandler(dataPoint, env);
	});
	return Promise.all(results);
};

export default {
	async fetch(request, env, ctx) {
		// Create JWT to authenticate the BigQuery API call
		let bqJWT;
		try {
			bqJWT = await generateBQJWT(env);
		} catch (error) {
			console.log(error);
			return new Response("An error has occurred while generating the JWT", {
				status: 500,
			});
		}

		// Fetch results from BigQuery
		let ticketInfo;
		try {
			ticketInfo = await queryBQ(
				bqJWT,
				`/bigquery/v2/projects/${env.BQ_PROJECT_ID}/queries`,
			);
		} catch (error) {
			console.log(error);
			return new Response("An error has occurred while querying BQ", {
				status: 500,
			});
		}

		// Transform output format into array of objects with named fields
		let formattedResults;
		if ("rows" in ticketInfo) {
			formattedResults = formatRows(ticketInfo.rows, ticketInfo.schema.fields);
		} else if ("error" in ticketInfo) {
			return new Response(ticketInfo.error.message, { status: 500 });
		}

		// Generate AI summaries and sentiment scores
		let summaries, sentimentScores;
		try {
			summaries = await getAIGeneratedContent(
				formattedResults,
				env,
				generateTags,
			);
			sentimentScores = await getAIGeneratedContent(
				formattedResults,
				env,
				generateSentimentScore,
			);
		} catch {
			return new Response(
				"There was an error while generating the text summaries or sentiment scores",
			);
		}

		// Add AI summaries and sentiment scores to previous results
		formattedResults = formattedResults?.map((formattedResult, i) => {
			if (sentimentScores[i].response && summaries[i].response) {
				return {
					...formattedResult,
					sentiment: parseFloat(sentimentScores[i].response).toFixed(2),
					tags: summaries[i].response.split(",").map((result) => result.trim()),
				};
			}
		});

		const response = { data: formattedResults };

		return new Response(JSON.stringify(response), {
			headers: { "Content-Type": "application/json" },
		});
	},
};

如果要部署此 Worker，可以运行 npx wrangler deploy：

Total Upload: <size_of_your_worker> KiB / gzip: <compressed_size_of_your_worker> KiB
Uploaded <name_of_your_worker> (x sec)
Deployed <name_of_your_worker> triggers (x sec)
  https://<your_public_worker_endpoint>
Current Version ID: <worker_script_version_id>

这将创建一个公共端点，供你全球访问 Worker。使用生产数据时请牢记这一点，并确保包含额外的访问控制。

结论

在本教程中，你学习了如何通过创建 GCP 服务账户密钥并将其部分存储为 Worker secrets 来集成 Google BigQuery 和 Cloudflare Workers。随后在代码中导入，并使用 jose npm 库创建 JSON Web Token 以验证对 BigQuery 的 API 查询。

获得结果后，你将其格式化以通过 Workers AI 传递给生成式 AI 模型，生成 tag 并对提取的数据执行情感分析。

后续步骤

如果你的工作流需要定期获取和存储数据（例如在 R2 或 D1 中），而不是在浏览器中显示将数据输入 AI 模型的结果，可以考虑为此 Worker 添加 scheduled handler。这使你能够通过 Cron Trigger 以预定义节奏触发 Worker。请考虑查看将 BigQuery 数据导入 Workers AI的参考架构图。

从其他来源（如本教程中所做）摄取数据的用例是创建 RAG 系统。如果这与你相关，请查看构建检索增强生成（RAG）AI 教程。

要了解更多可在 Cloudflare 使用的 AI 模型，请访问文档的 Workers AI 部分。