循序渐进 · 教学 · 函数（二十）
流式函数：分块返回结果
Python 与 TypeScript v2 函数可以把结果分块流式返回，数据边生成边处理，不用等全部算完。
本文来源 · Source
内容整理自 Palantir Foundry 官方文档：
https://www.palantir.com/docs/foundry/functions/streaming-functions/
原始标题：Language-agnostic features > Streaming functions
https://www.palantir.com/docs/foundry/functions/streaming-functions/
原始标题：Language-agnostic features > Streaming functions
★
先记住这几条
① 流式 = 边算边给
结果是分块（chunk）返回的，调用方可以立即开始处理。
② 适合耗时生成场景
数据随时间产生时（如大模型逐字输出）最有用。
③ 只有部分语言支持
Python 与 TypeScript v2 支持，TypeScript v1 不支持。
④ 通过 OSDK 调用
消费端要用支持流式的方式接收。
0
写在前面
Python 和 TypeScript v2 函数可以把一次执行的结果分块流式返回。当数据是随时间陆续产生的时候，这个能力很有用 —— 调用方不必等结果全部生成完，就能开始处理。
1
写流式函数
写流式函数
要点：函数侧怎么写才能分块产出结果。
要流式返回，Python 函数必须把返回类型标注为 Iterable[T]，并用 yield 关键字在数据可用时逐个产出。在 TypeScript 里，你需要在 function 关键字后加一个 * 来声明异步生成器函数，返回类型写 AsyncIterable<T>，同样用 yield 逐个产出。
下面的例子演示了一个函数如何返回整数流，每个整数之间间隔一秒：
from functions.api import function
from typing import Iterable
import time
@function
def my_lazy_number_generator(n: int) -> Iterable[int]:
for i in range(n):
time.sleep(1)
yield iimport { Integer } from "@osdk/functions";
export default async function* myLazyNumberGenerator(n: Integer): AsyncIterable<Integer> {
for (let i = 0; i < n; i++) {
await new Promise(resolve => setTimeout(resolve, 1_000));
yield i;
}
}流式返回在处理语言模型时特别有用 —— 模型生成完整输出往往要花不少时间。通过把模型产出的每一块内容立刻 yield 出去，你就能提供实时体验，而不必阻塞等待整个响应完成。关于在函数里调用语言模型，参见TypeScript v2 与 Python 函数中的语言模型。
下面的例子用 openai SDK 调用语言模型，并通过在请求里打开 stream 开关把响应流式传回来：
from openai import OpenAI
from functions.api import function
from functions.aliases import model
from foundry_sdk.v2.language_models import (
get_openai_base_url,
get_foundry_token,
get_http_client,
)
from typing import Iterable
@function
def create_chat_completion(prompt: str) -> Iterable[str]:
client = OpenAI(
api_key=get_foundry_token(preview=True),
base_url=get_openai_base_url(preview=True),
http_client=get_http_client(preview=True),
)
stream = client.chat.completions.create(
model=model("gpt55").rid,
messages=[
{
"role": "user",
"content": prompt,
},
],
stream=True
)
for event in stream:
if event.choices:
content = event.choices[0].delta.content
if content:
yield contentimport { PlatformClient } from "@osdk/client";
import OpenAI from "openai";
import { Aliases } from "@osdk/functions";
import { getFoundryToken, getOpenAiBaseUrl, createFetch } from "@osdk/language-models";
export default async function* createChatCompletion(client: PlatformClient, prompt: string): AsyncIterable<string> {
const oaiClient = new OpenAI({
apiKey: await getFoundryToken(client),
baseURL: getOpenAiBaseUrl(client),
fetch: createFetch(client),
});
const stream = await oaiClient.chat.completions.create({
model: Aliases.model("gpt55").rid,
messages: [
{ role: 'user', content: prompt },
],
stream: true
});
for await (const event of stream) {
const content = event.choices[0]?.delta?.content;
if (content) {
yield content;
}
}
}2
通过 OSDK 调用流式函数
通过 OSDK 调用流式函数
要点：消费端如何逐块接收并处理。
流式函数打好标签并发布之后，就可以通过本体 SDK来调用它 —— 在 React 应用里、在 Workshop 的自定义组件里，或者在另一个函数里。
:::callout{theme="neutral" title="Beta"}
通过本体 SDK 执行带流式响应的函数目前处于 beta 阶段，开发过程中功能可能发生变化。
from foundry_sdk_runtime import AllowBetaFeatures
with AllowBetaFeatures():
with client.ontology.queries.create_chat_completion_streaming(prompt="珠穆朗玛峰在哪里？") as stream:
for text in stream:
# ...import { __EXPERIMENTAL__NOT_SUPPORTED_YET__executeStreamingFunction } from "@osdk/api/unstable";
const stream = client(__EXPERIMENTAL__NOT_SUPPORTED_YET__executeStreamingFunction).executeStreamingFunction(
createChatCompletion,
{
prompt: "珠穆朗玛峰在哪里？",
}
);
for await (const text of stream) {
// ...
}延伸阅读 · 相关页面
按主题横向跳转，不必顺着目录一篇篇读。
常见问题速答 · FAQ
关于「流式函数：分块返回结果」，读者最常问的几个问题。
写流式函数是什么？
函数侧怎么写才能分块产出结果。要流式返回，Python 函数必须把返回类型标注为 Iterable[T]，并用 yield 关键字在数据可用时逐个产出。
通过 OSDK 调用流式函数是什么？
消费端如何逐块接收并处理。流式函数打好标签并发布之后，就可以通过本体 SDK来调用它 —— 在 React 应用里、在 Workshop 的自定义组件里，或者在另一个函数里。