Pathway LLM 聊天封装详解：在 RAG 与实时数据管道中统一调用 OpenAI、LiteLLM、HuggingFace 与 Cohere

Pathway Live Data Framework 的 LLM xpack 提供了一组开箱即用的「聊天模型封装」（LLM Chat Wrappers），让你把 OpenAI、LiteLLM（Gemini、Azure OpenAI、HuggingFace API 等数十种模型）、本地 HuggingFace 模型以及 Cohere 服务，统一成可作用于表格列的 UDF。读完本文，你将掌握每个封装类的构造参数与调用方式、消息格式（messages）的两种传法、「初始化设默认、应用时可覆盖」的参数机制，以及 capacity、retry_strategy、cache_strategy 三个异步控制项的用法，并能在 Python 代码或模板 YAML 中复制运行这些示例。

封装即 UDF：聊天模型在 Pathway 中的定位

每个聊天封装类都是 UDF（User Defined Function）。UDF 在 Pathway 语境下是任意「接收输入、处理、返回输出」的自定义函数，用于把自定义逻辑（如调用 LLM）无缝集成进表格计算。

聊天封装正是 UDF 的典型用法：把它当做一个列表达式级别的函数，传入 prompt 列，得到回复列，从而自然融入表格/列式的数据流。所有封装类都继承自 BaseChat（定义于 python/pathway/xpacks/llm/llms.py），其构造函数参数会透传给 pathway.UDF 的构造器，因此容量、重试、缓存等行为对所有封装一致。

当前 xpack 提供的聊天封装包括：

• OpenAI（OpenAIChat）
• LiteLLM（LiteLLMChat，覆盖 Gemini、Azure OpenAI、HuggingFace API 等众多供应商，完整列表以 LiteLLM 官方文档为准）
• Hugging Face Pipeline（HFPipelineChat，本地运行）
• Cohere（CohereChat）

从源码结构看，仓库中还存在 BedrockChat（AWS Bedrock Converse API）封装，同样实现 BaseChat 接口，可用于 Claude、Llama、Titan 等 Bedrock 模型；本文以文档化的四个封装为主线展开。

准备工作：构造一张查询表

下面所有示例共用一张由 Markdown 构造的查询表：

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,
)

OpenAIChat：原生 OpenAI 封装

对 OpenAI，使用 OpenAIChat 类创建封装实例：

from pathway.xpacks.llm import llms
import os

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)

在模板（模板 YAML 中通过标签 !pw.xpacks.llm.llms.OpenAIChat 声明）场景下对应写法为：

chat: !pw.xpacks.llm.llms.OpenAIChat
  model: "gpt-4o-mini
  api_key: $OPENAI_API_KEY

消息格式（Message format）

OpenAIChat 期望的消息格式与 OpenAI API 一致——一个字典列表，每个字典是对话中到当前为止的一条消息。只问单个问题时，用 pw.xpacks.llm.llms.prompt_chat_single_qa 把字符串包装成 OpenAI 期望的格式，即上面示例的用法。

prompt_chat_single_qa 本身是一个 @pw.udf：把问题字符串转换为单元素列表 [{"role": "user", "content": question}]，见 llms.py 中的定义。

如果想更精细地控制发给 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)

从源码看，OpenAIChat.__wrapped__ 会先通过 _prepare_messages（llms.py L30-L40）把 pw.Json 或 list[dict] 统一解码为 list[dict]，因此两种列类型（普通 list[dict] 列与 pw.Json 列）都能接受。

模型参数：默认值与运行时覆盖

OpenAI API 接受大量参数（如 model、api_key、max_tokens、temperature 等）。OpenAIChat 允许在初始化时设置默认值，也可以在实际调用（应用）时覆盖：

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 set when initializing OpenAIChat
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)

这一「默认值 + 覆盖」机制的底层逻辑在源码中一目了然：构造函数把除 api_key、base_url 以外的所有关键字参数都收进 self.kwargs（作为默认值），而 __wrapped__ 里执行 kwargs = {**self.kwargs, **kwargs} 合并——调用时传入的参数优先于初始化默认值（llms.py L242-L316）。因此像 max_tokens 这样的参数既可以写死在构造函数里，也可以像示例那样从表格列中逐行取值，天然适配「每行不同参数」的流式场景。

两个值得注意的实现细节：

• model 的特殊地位：OpenAIChat 构造函数中 model 默认为 "gpt-3.5-turbo"；若你希望每次调用都动态指定模型，应在构造函数中显式传 model=None，然后在 UDF 调用时传入 model=...（类文档字符串与 源码 均有说明）。
• 参数合法性校验：调用期参数并非无限制透传。_accepts_call_arg 会借助 LiteLLM 的 get_supported_openai_params 查询目标模型实际支持的参数集（_utils.py L15-L23）。单测 test_llms.py 验证了 top_p、temperature、max_tokens 合法而 made_up_arg 被拒绝，避免把模型不认识的参数误发给 API。

LiteLLM：一个封装接入众多供应商

框架为 LiteLLM 提供了 LiteLLMChat 封装。以 Gemini 为例，创建实例后同样作用于消息列：

from pathway.xpacks.llm import llms
import os

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)

模板场景对应：

llm: !pw.xpacks.llm.llms.LiteLLMChat
  model: "gemini/gemini-pro", # Choose the model you want

借助 LiteLLM 封装，Pathway 可以调用其支持列表中的大量流行 LLM（Azure OpenAI、HuggingFace API、Gemini 等），供应商前缀即写在 model 字符串里（如 gemini/gemini-pro、anthropic/claude-3-5-sonnet-20240620）。

与 OpenAIChat 的关键差异在源码中可见：

• LiteLLMChat 的 model 无默认值，必须在构造函数或每次调用时提供（llms.py L386）；
• _accepts_call_arg 会解析 model 中的 provider/model 前缀（例如 replicate/meta/meta-llama-3-8b 会被拆成 provider replicate 与模型 meta/meta-llama-3-8b），再查该供应商模型支持的参数（llms.py L444-L463），因此不同供应商（Anthropic 支持 response_format 而 Cohere 不支持）的参数校验是精确到模型的。

Hugging Face Pipeline：本地模型封装

对希望本地运行的 Hugging Face 模型，框架提供了独立封装 HFPipelineChat（走 HuggingFace API 则用 LiteLLM 封装）。创建实例时会初始化一个 HuggingFace pipeline，因此 pipeline 的任何参数（包括模型名）都必须在 HFPipelineChat 初始化时设置；而 pipeline.__call__ 的参数（call_kwargs）则同前，可在初始化时设置或在应用时覆盖。

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)

模板场景：

llm: !pw.xpacks.llm.llms.HFPipelineChat
  model: "TinyLlama/TinyLlama-1.1B-Chat-v1.0", # Choose the model you want

注意：HF pipeline 接受的问题格式取决于模型。像 gpt2 这类模型期望 prompt 字符串；会话类模型则接受消息字典列表，此时会应用模型自身的 prompt 模板。模板（AI pipelines）场景则期望会话模型，因此 gpt2 这类纯生成模型无法使用。

以会话模型 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)

源码层面，HFPipelineChat 有两个其他封装没有的构造参数：device（默认 "cpu"，指定运行设备）与 batch_size（默认 32，作为 UDF 的最大批大小，更大的批在 GPU 上通常能降低生成耗时），见 llms.py L500-L516。其 __wrapped__ 还实现了「能否批处理」的运行时判断：当各行覆盖参数不一致（per-row kwargs），或 tokenizer 无 pad_token_id 时退化为逐行推理，否则整批送入 pipeline（llms.py L518-L590）——这与异步 API 封装的并发模型不同，是本地批推理的体现。此外它还附带一个实用工具方法 crop_to_max_length，用 tokenizer 把输入截断到指定长度（默认 500 token），适合在长文档问答前做 prompt 裁剪。

Cohere：带引用（citations）的 RAG 友好封装

框架还提供 CohereChat 封装（对应 Cohere Chat Services），特色是支持把文档作为上下文直接传给模型，并在结果中返回被引用的文档——对 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)

从 源码 可以看到 CohereChat 的签名比其他封装多了一个 documents 位置参数：内部把消息列表的最后一条作为 message、其余作为 chat_history 传给 cohere.Client.chat，documents 以 [{"text": ...}] 形式附送，返回值即 (ret.text, ret.citations)。其模型默认为 command。documents 列可以是 list[dict]、pw.Json，甚至由多个文档列 pw.reducers.tuple 聚成的元组，封装内部都会归一化。

封装是异步的：capacity、retry_strategy 与 cache_strategy

OpenAI 与 LiteLLM 封装（无论聊天还是嵌入）都是异步的，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)

模板（YAML）对应写法：

chat: !pw.xpacks.llm.llms.OpenAIChat
  model: "gpt-4o-mini
  capacity: 10
  retry_strategy: !pw.udfs.ExponentialBackoffRetryStrategy
    max_retries: 5
    initial_delay: 1000
    backoff_factor: 2

这三个参数更完整的语义参见 UDFs 指南中的 AsyncExecutor 章节（UDF 文档），其要点是：

• capacity 决定 UDF 最多同时跑多少个在途 API 请求，是控制下游限流的第一道阀门；
• retry_strategy 支持 ExponentialBackoffRetryStrategy（指数退避，max_retries/initial_delay/backoff_factor 可控）、FixedDelayRetryStrategy（固定延迟）等，默认即指数退避策略——见各封装构造函数签名中的默认值 pw.udfs.ExponentialBackoffRetryStrategy()；
• cache_strategy 默认 None（不缓存）；传入如 DefaultCache 等策略后，若设置了 PATHWAY_PERSISTENT_STORAGE，重复相同的调用会命中持久化缓存，避免重复付费调用 LLM。

源码纵深：重试发生在哪一层？

阅读 llms.py 会发现一个刻意的设计：OpenAIChat 创建的 openai.AsyncOpenAI 客户端显式设置了 max_retries=0（禁用 SDK 自带重试），重试改由 self.retry_strategy.invoke(self.client.chat.completions.create, ...) 在 __wrapped__ 内部执行（llms.py L256-L294）。这样即使绕过 executor 直接调用 __wrapped__，重试语义依然成立。Bedrock 封装更进一步：每次重试都会重建 AWS client，因为失败后旧 client 可能处于损坏状态（llms.py L996-L1004）。

这些行为都有测试印证：test_llms.py 中 test_openai_chat_wrapped_retries_transient_errors 用 mock 让第一次请求抛出运行时错误、第二次成功，断言最终拿到响应且 create 恰好被调用两次；Bedrock 侧亦有同构测试（L235-L262），还验证了 top_k 这类模型专属参数会被正确路由到 additionalModelRequestFields（L265-L294）。

另一个从源码结构可推断的细节：OpenAIChat/LiteLLMChat/BedrockChat 构造函数还接受关键字参数 async_mode，取值 "batch_async"（默认）或 "fully_async"，由 _utils.py 中的 _prepare_executor 分派到 pw.udfs.async_executor 或 pw.udfs.fully_async_executor。原文档未展开此参数，实际使用时默认值即可满足大多数场景；选择 fully_async 与否应结合你的管道并发特征评估。

小结

Pathway 的 LLM 聊天封装把「调用哪家模型」收敛为「实例化哪个封装类」，而「以什么并发、如何重试、是否缓存」则由统一的 capacity/retry_strategy/cache_strategy 三参数控制；参数默认值可初始化时设定、可在应用时按列覆盖，使 LLM 调用天然成为流式表中的一列。实现集中在 python/pathway/xpacks/llm/llms.py，行为由 python/pathway/xpacks/llm/tests/test_llms.py 覆盖；在模板体系（模板 YAML 中 !pw.xpacks.llm.llms.* 标签）中同样可声明式使用，相关配置示例另见 RAG 配置示例文档 与 完整管道示例。