Pathway LLM xpack 详解：LLM Chats 封装（OpenAIChat / LiteLLMChat / HFPipelineChat / CohereChat）

本文基于 Pathway 官方开发者文档 docs/2.developers/4.user-guide/50.llm-xpack/.chats/llm-chats.md 展开，系统讲解 Pathway Live Data Framework 的 LLM xpack 中四款 Chat 封装类的使用方式：消息格式、模型参数覆盖、异步并发与重试/缓存策略配置。读完本文后，你可以把 Pathway 表中的提示词（prompt）列直接接入 OpenAI、Gemini（经由 LiteLLM）、本地 HuggingFace 模型或 Cohere RAG 服务，并理解这些封装类在源码层面如何作为 UDF 融入流式计算图。

一、LLM xpack 的 Chat 封装总览

Pathway 的 LLM xpack 开箱提供两类 UDF：文本生成（chat）与向量嵌入（embedding）。对于文本生成，官方文档给出的封装路线是：

• 原生封装：OpenAI Chat 模型（OpenAIChat）与本地运行的 HuggingFace 模型（HFPipelineChat）；
• 统一网关封装：Azure OpenAI、HuggingFace API、Gemini 等大量其他提供商通过 LiteLLMChat 接入；
• Cohere 封装：CohereChat 额外支持把上下文文档随查询一起发送，并在结果中返回引用文档（citations）。

这四个封装类全部定义在 python/pathway/xpacks/llm/llms.py 中：

• BaseChat（L43）：所有 Chat 封装的基类；
• OpenAIChat（L95）；
• LiteLLMChat（L331）；
• HFPipelineChat（L466）；
• CohereChat（L631）。

此外，从源码结构看，该文件还额外实现了 BedrockChat（L771），用于调用 AWS Bedrock Converse API 上的 Claude、Llama、Titan 等模型，可作为上述四类之外的补充选项。

二、封装的设计：每个 Chat 封装都是一个 UDF

官方文档在 "UDFs" 一节明确指出：每个 wrapper 都是一个 UDF（User Defined Function），它接收输入、处理并返回输出；在 Pathway 的语境中，UDF 让自定义逻辑（比如调用 LLM）能无缝嵌入表格与列的转换中。

从源码可以进一步看清这一设计。以 llms.py 中的 BaseChat 为例：

class BaseChat(pw.UDF):
    """Base class for the LLM chat instances.
    Constructor arguments are passed to the :py:func:`~pathway.UDF` constructor.
    """

它直接继承自 pw.UDF，因此每个封装实例本质上是一个可被 select 应用到列上的算子。典型的接入模式是两步：

1. 构造封装实例（此时传入模型、密钥以及并发/重试/缓存等执行参数）；
2. 将封装应用到包含 prompt 的列（此时可覆盖部分参数）。

官方文档给出的示例先构造一张调试表：

import pathway as pw
queries = pw.debug.table_from_markdown(
    """
questions | max_tokens
How many 'r' there are in 'strawberry'? | 400
""",
    split_on_whitespace=False,
)

所有 Chat 封装都实现了相同的调用约定：model(prompt_column, **参数) 返回一个 ColumnExpression，其中 prompt_column 是 list[dict] 或 pw.Json 类型的消息列。这个统一签名使不同提供商的模型可以互相替换。

三、OpenAIChat：原生 OpenAI 封装

3.1 基本用法

对 OpenAI 使用 OpenAIChat 类创建封装，再把 queries 表中 questions 列的问题发给模型：

from pathway.xpacks.llm import llms

model = llms.OpenAIChat(
    model="gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],  # Read OpenAI API key from environmental variables
)
# Send queries from column `question` in table `queries` to OpenAI
responses = queries.select(result=model(llms.prompt_chat_single_qa(pw.this.questions)))
# Run the computations (including sending requests to OpenAI) and print the output table
pw.debug.compute_and_print(responses)

在 pyproject.toml 中可以看到，xpack-llm 可选依赖固定了 openai >= 2.20, < 3.0，因此上述代码以该 SDK 版本区间为适用前提。

从源码看，OpenAIChat.__init__（L242-L274）有几个值得注意的细节：

• model 的默认值是 "gpt-3.5-turbo"（注意：若你打算在 UDF 调用时逐行指定 model，需在构造时将其设为 None）；
• retry_strategy 默认为 pw.udfs.ExponentialBackoffRetryStrategy()；
• API Key 通过 api_key 参数传入，或回退到 OPENAI_API_KEY 环境变量（由 openai.AsyncOpenAI 客户端处理）；
• 内部客户端显式设置 max_retries=0，重试逻辑统一在 __wrapped__ 内通过 self.retry_strategy.invoke(...) 执行（见 L292-L294），源码注释说明这样设计是为了让直接调用 __wrapped__ 的场景也被重试覆盖；
• 每次请求/响应都会以 JSON 事件形式写入日志（openai_chat_request / openai_chat_response），非 verbose 模式下内容会被截断或脱敏（图片 base64 会被 _prep_message_log 打码）。

3.2 消息格式（Message format）

OpenAIChat 期望的消息格式是 OpenAI Chat API 要求的字典列表——每个字典代表对话至今的一条消息。对单轮问答，文档建议使用 pw.xpacks.llm.llms.prompt_chat_single_qa 把字符串包装成合规格式。该函数在源码 L1058-L1080 中实现，本质是把 question 转成 pw.Json([{"role": "user", "content": question}])：

@pw.udf
def prompt_chat_single_qa(question: str) -> pw.Json:
    return pw.Json([dict(role="user", content=question)])

如果你希望更精细地控制发往 OpenAI 的消息（例如加入 system 提示词），可以自己构造消息表：

messages = pw.debug.table_from_rows(
    pw.schema_from_types(questions=list[dict]),
    rows=[
        (
            [
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": "How many 'r' there are in 'strawberry'?"},
            ],
        )
    ],
)
responses = messages.select(result=model(pw.this.questions))
pw.debug.compute_and_print(responses)

源码中的 _prepare_messages（L30-L40）负责把 pw.Json 列或 list[dict] 列统一解码为纯 Python 字典列表，再交给 SDK。

3.3 模型参数：构造时设默认，调用时可覆盖

OpenAI API 接受 model、max_tokens、temperature 等大量参数。文档指出：OpenAIChat 允许在初始化时设置默认值，但也可以在应用（调用）时覆盖。这与源码构造函数签名一致——**openai_kwargs 会被收集进 self.kwargs，在 __wrapped__ 中通过 kwargs = {**self.kwargs, **kwargs} 完成覆盖（L278）。

model = llms.OpenAIChat(
    model="gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],  # Read OpenAI API key from environmental variables
    max_tokens=200,  # Set default value of max_tokens to be 200
)
# As max_tokens is not set, value 200 will be used
responses = queries.select(result=model(llms.prompt_chat_single_qa(pw.this.questions)))
# Now value of max_tokens is taken from column `max_tokens`, overriding default value
responses = queries.select(result=model(llms.prompt_chat_single_qa(pw.this.questions), max_tokens(pw.this.max_tokens)))
pw.debug.compute_and_print(responses)

这意味着 max_tokens 这类参数可以逐行来自列表达式——这正是流式场景下同一图里对不同问题使用不同生成上限的关键能力。

还有一个源码层面的防呆机制：OpenAIChat._accepts_call_arg（L318-L328）借助 LiteLLM 的 get_supported_openai_params 校验某个参数是否被当前 model 支持（见 python/pathway/xpacks/llm/_utils.py 中的 _check_model_accepts_arg）。如果构造时 model 为 None，则一律返回 False。单元测试 python/pathway/xpacks/llm/tests/test_llms.py 验证了 top_p、temperature、max_tokens 被接受、made_up_arg 被拒绝的行为。

四、LiteLLMChat：一个网关接入多家模型

Pathway 为 LiteLLM 提供了 LiteLLMChat 封装。以 Gemini 为例：

from pathway.xpacks.llm import llms

model = llms.LiteLLMChat(
    model="gemini/gemini-pro",  # Choose the model you want
    api_key=os.environ["GEMINI_API_KEY"],  # Read GEMINI API key from environmental variables
)
# Ask Gemini questions from `prompt` column
responses = queries.select(result=model(llms.prompt_chat_single_qa(pw.this.questions)))
pw.debug.compute_and_print(responses)

借助 LiteLLM，同一封装即可覆盖 Azure OpenAI、HuggingFace API、Anthropic、本地 Ollama 等众多提供商（完整提供商清单以 LiteLLM 官方文档为准）。

从源码实现看（L379-L422）有两个差异点：

1. model 没有默认值，必须在构造或每次调用时指定；
2. 参数合法性校验时会解析 provider/model 前缀（例如 anthropic/claude-3-5-sonnet 被拆成 provider anthropic 与 model claude-3-5-sonnet，并支持 replicate/meta/meta-llama-3-8b 这类多级命名，见 L444-L463）。

测试 test_mixed_call_args 展示了不同提供商参数差异的实际影响：stream_options、response_format 在 Claude 上被接受，而在 Cohere command-r 上被拒绝。

五、HFPipelineChat：本地运行 HuggingFace 模型

对希望在本地推理的 HuggingFace 模型，Pathway 提供单独的封装 HFPipelineChat（调用 HuggingFace 的在线 API 则应使用 LiteLLM 封装）。构造该封装时会立即初始化一个 HuggingFace pipeline，因此所有 pipeline 的参数——包括模型名——必须在初始化 HFPipelineChat 时确定；而 pipeline.__call__ 的参数可以像前面一样在初始化时设置、在调用时覆盖。

最简示例（生成式模型 gpt2，输入是纯 prompt 字符串列）：

from pathway.xpacks.llm import llms

model = llms.HFPipelineChat(
    model="gpt2",  # Choose the model you want
)
responses = queries.select(result=model(pw.this.questions))
pw.debug.compute_and_print(responses)

文档特别提醒：HuggingFace pipeline 中问题的格式取决于模型——gpt2 这类模型期望 prompt 字符串，而对话类模型也接受消息字典列表（此时会套用模型自带的 prompt 模板）。例如对话模型 TinyLlama/TinyLlama-1.1B-Chat-v1.0 就应该用 prompt_chat_single_qa 包装：

from pathway.xpacks.llm import llms

model = llms.HFPipelineChat(
    model="TinyLlama/TinyLlama-1.1B-Chat-v1.0",
)
responses = queries.select(result=model(llms.prompt_chat_single_qa(pw.this.questions)))
pw.debug.compute_and_print(responses)

从源码（L500-L516）可以补充出文档未展开的构造参数：

参数	默认值	说明
model	"gpt2"	传给 transformers.pipeline 的模型 ID
call_kwargs	{}	每次调用 pipeline.__call__ 时传入的参数，可在应用时覆盖
device	"cpu"	推理设备（如 "cpu" / "cuda"）
batch_size	32	单批最大样本数，对应 UDF 的 max_batch_size；文档说明更大的 batch 在 GPU 上可能降低生成耗时
**pipeline_kwargs	—	透传给 transformers.pipeline 初始化

批量推理逻辑在 __wrapped__（L518-L590）中：如果所有行共享同一组调用参数、且 tokenizer 的 pad_token_id 可用，则整批一起送入 pipeline；只要存在逐行不同的参数（per-row kwargs）或缺少 pad token，就退化为逐条推理。另外该类还提供一个便捷的 crop_to_max_length 方法（L592-L602），可按 tokenizer 把列中字符串裁剪到 max_prompt_length（默认 500）个 token。

六、CohereChat：带引用文档的 RAG 问答

Pathway 同时提供 Cohere Chat Services 的封装。与前三者的不同点在于：它允许把文档列与查询一起发送，返回结果是 (response, cited_documents) 元组——响应文本加上引用文档列表（无引用时为空列表），非常适合 RAG 场景。

from pathway.xpacks.llm import llms

model = llms.CohereChat()
queries_with_docs = pw.debug.table_from_rows(
    schema=pw.schema_from_types(questions=str, docs=list[dict]),
    rows=[
        (
            "What is RAG?",
            [
                {
                    "text": "Pathway is a high-throughput, low-latency data processing framework that handles live data & streaming for you."
                },
                {"text": "RAG stands for Retrieval Augmented Generation."},
            ],
        )
    ],
)

r = queries_with_docs.select(
    ret=model(llms.prompt_chat_single_qa(pw.this.questions), documents=pw.this.docs)
)
parsed_table = r.select(response=pw.this.ret[0], citations=pw.this.ret[1])
pw.debug.compute_and_print(parsed_table)

从源码（L702-L738）看，CohereChat 会把消息列表的最后一条作为 message、其余作为 chat_history 传给 cohere.Client.chat(..., documents=docs)，然后解包 ret.text 与 ret.citations 返回。默认模型为 command（构造参数 model: str | None = "command"，L677-L700），api_key 支持在调用时传入，否则走 Cohere 客户端的环境变量。文档中 parsed_table 的列切分 response=pw.this.ret[0]、citations=pw.this.ret[1] 正是对应这个二元组返回值。

七、异步执行：capacity、retry_strategy 与 cache_strategy

官方文档最后一节说明：OpenAI 与 LiteLLM 的封装（chat 与 embedding）都是异步的，Pathway 允许通过三个参数控制其行为，且这三个参数必须在构造封装时设置：

• capacity：允许的最大并发操作数；
• retry_strategy：失败时的重试策略；
• cache_strategy：缓存机制。

model = llms.OpenAIChat(
    # maximum concurrent operations is 10
    capacity=10,
    # in case of failure, retry 5 times, each time waiting twice as long before retrying
    retry_strategy=pw.udfs.ExponentialBackoffRetryStrategy(max_retries=5, initial_delay=1000, backoff_factor=2),
    # if PATHWAY_PERSISTENT_STORAGE is set, then it is used to cache the calls
    cache_strategy=pw.udfs.DefaultCache(),
    # select the model
    model="gpt-4o-mini",
    # read OpenAI API key from environmental variables
    api_key=os.environ["OPENAI_API_KEY"],
)
responses = queries.select(result=model(prompt_chat_single_qa(pw.this.questions)))
pw.debug.compute_and_print(responses)

这三个参数在源码中的落地位置是 python/pathway/xpacks/llm/_utils.py 的 _prepare_executor：async_mode（默认 "batch_async"，可选 "fully_async"）决定使用 pw.udfs.async_executor 还是 pw.udfs.fully_async_executor，capacity 与 retry_strategy 随 executor 传入，cache_strategy 则传给 pw.UDF 构造器。以 OpenAIChat 为例（L242-L271），capacity 缺省为 None（不限制并发）。

重试策略的实际效果有测试佐证：test_openai_chat_wrapped_retries_transient_errors 用 FixedDelayRetryStrategy(max_retries=4, delay_ms=1) 构造封装，mock 的 API 第一次抛出连接错误、第二次成功，断言最终返回 "mocked" 且 create 被调用了 2 次——即重试确实发生在 UDF 内部而非 SDK 层。

八、安装依赖与适用前提

上述封装均以可选依赖（extras）形式提供，从 pyproject.toml 可以确认安装方式与版本约束：

xpack-llm = [
    "openai >= 2.20, < 3.0",
    "litellm >= 1.84.0, < 1.92; python_version < '3.14'",
    "litellm == 1.83.0; python_version >= '3.14'",
    "cohere >= 5.1, < 8.0",
    ...
]
xpack-llm-local = [  # requirements that allow local ML inference
    "sentence_transformers",
    "transformers >= 4.50.2, < 5.0",
]

也就是说：OpenAIChat、LiteLLMChat、CohereChat 需要安装 pathway[xpack-llm]（源码中通过 optional_imports("xpack-llm") 强制检查 openai / litellm / cohere 可导入）；本地推理的 HFPipelineChat 则依赖 pathway[xpack-llm-local]（检查 transformers 可导入，见 L508-L509）。源码中对这些导入做了 optional import 保护，未安装对应 extras 时导入封装会给出明确提示而非静默失败。

小结与验证入口

• 四个 Chat 封装统一实现了「构造时设默认 + 调用时覆盖」的参数约定，且都以 model(prompt_column, **kwargs) -> ColumnExpression 的签名接入 select，可无缝替换；
• 参数合法性按模型逐校验（_accepts_call_arg），避免把某模型不支持的参数发给 API；
• 异步封装的并发、重试、缓存行为由 executor 与 strategy 对象统一承接，OpenAIChat 内部关闭 SDK 自带重试、改由 retry_strategy.invoke 统一重试；
• 想深入验证行为，可以直接阅读 python/pathway/xpacks/llm/tests/test_llms.py（构造参数、重试、参数校验的测试）与 python/pathway/xpacks/llm/llms.py 的实现。