RocketRide Python SDK Chat 实战：用 Question / Answer 构建对话式 AI Pipeline

本指南以 RocketRide Python SDK 的 Chat 功能为核心，讲解如何基于 Question 构建对话请求、通过 client.chat() 发送到 chat / webhook / dropper 三类 pipeline 源，并用 Answer 解析 AI 返回的结构化结果。读完本文你将掌握一套可直接运行的对话式 AI 集成方案，包括多轮上下文、指令与示例引导、JSON 结构化输出、SSE 流式回调与 token 用量追踪。

Chat 是什么：对话式管道（Conversational lane）

在 RocketRide 中，Chat 是独立于普通数据发送的"对话通道"。它的定位非常明确：面向 chat、webhook、dropper 三种 pipeline 源。如果你的 pipeline 源是 chat，应该使用 client.chat()；只有当源是 webhook 或 dropper 时，才使用 send() / send_files()（见 README 说明）。

从底层实现看（chat.py 混入实现），一次 chat() 调用实际经历以下四步：

1. 打开一条 MIME 类型为 application/rocketride-question 的数据管道；
2. 将 Question 序列化为 JSON（question.model_dump_json()）写入管道；
3. 关闭管道，等待服务器返回结果；
4. 返回标准 PIPELINE_RESULT 字典。

这条 MIME 类型并非随意设定。在引擎侧，data_conn.py 的管道监听判定逻辑中，application/rocketride-question 会优先路由到名为 questions 的 lane（'questions' in listeners 时才命中，否则退回 raw 数据通道）；对应地，chat_pipeline.py 中的 LLM 组件正是以 'lane': 'questions' 作为输入监听。SDK 的 Chat 混入在源码注释中也明确说明："No provider filter — chat() works with chat, webhook, and dropper sources"，即该通道不依赖特定模型提供方，实际使用哪个 LLM 由 pipeline 配置决定。

构建 Question：对话请求的核心载体

Question 是发送给 AI 的完整请求封装，位于 rocketride.schema。最基本的用法如下：

from rocketride.schema import Question

question = Question(expectJson=True)
question.addInstruction('Format', 'Return a JSON object with keys: summary, keywords.')
question.addExample('Summarize X', {'summary': '...', 'keywords': ['a', 'b']})
question.addQuestion('Summarize the main points and list keywords.')

构造参数与字段

根据 question.py 的模型定义，Question 的完整构造签名可归纳为：

字段	类型	默认值	说明
type	QuestionType	QuestionType.QUESTION	问题类型：QUESTION（基础问答）、SEMANTIC（语义理解）、KEYWORD（关键词匹配）、GET（信息检索）、PROMPT（原始提示词，不附加处理）
filter	DocFilter	DocFilter()	控制检索哪些文档及结果处理方式
expectJson	bool	False	为 True 时要求并解析结构化 JSON 响应
role	str	''	设定 AI 角色/人设，例如 "You are a financial analyst"
instructions	List[QuestionInstruction]	[]	自定义指令列表
history	List[QuestionHistory]	[]	多轮对话历史
examples	List[QuestionExample]	[]	输入-输出示例对
context	List[str]	[]	补充背景上下文
goals	List[str]	[]	高层目标，渲染为独立的 ### Ultimate Goal 段
documents	List[Doc]	[]	指定 AI 引用的文档
questions	List[QuestionText]	[]	要问的问题文本（通常一个）

指令方法速查

Question 提供了一系列链式方法，逐一说明其语义与底层行为：

• addInstruction(title, instruction) — 追加一条指令。在 getPrompt() 中会被编号渲染进 ### System Instructions: 区块（question.py）。
• addExample(given, result) — 追加示例对。result 为 dict/list 时会被 json.dumps 序列化；渲染为 ### Examples: 区块，若 expectJson=True 还会包上 ```json 代码围栏（question.py）。
• addContext(context) — 追加背景信息，支持字符串、字典、列表（非字符串项会 str() 化），渲染进 ### Context: 区块。
• addHistory(item) — 追加一条 QuestionHistory(role, content) 历史消息，role 取值 user / system / assistant，渲染进 ### Conversation History: 区块。
• addGoal(goal) — 追加高层目标，渲染为独立的 ### Ultimate Goal: 区块，与步骤级指令分开。
• addDocuments(documents) — 指定 Doc 对象或字符串作为参考文档（字符串会包成 Doc(chunkId=0, objectId='', page_content=...)）。
• addQuestion(question) — 追加问题文本，渲染进 ### Current Task: 区块，是"主问题"所在。

一个值得注意的实现细节：当 expectJson=True 时，getPrompt() 会自动注入一段"JSON Response Format"系统指令，要求模型"只返回围栏包裹的合法 JSON、正确转义引号、不加额外文本"，并在传入 has_previous_json_failed=True 时追加一条 CRITICAL 指令要求修复上次的非法 JSON（question.py）。也就是说，结构化输出不是只靠 expectJson 一个标志位，而是由 SDK 自动拼装提示词约束模型完成的，这也是"请求侧约束 + 响应侧校验"双保险的体现。

用 DocFilter 控制检索范围

Question 的 filter 参数是 DocFilter（doc_filter.py），常用字段包括：

字段	默认值	说明
offset / limit	0 / 25	分页控制
fullDocuments / fullTables	False	是否把同一文档/表格的 chunk 合并为完整结果
isTable	None	只检索表格数据（True）或排除表格（False）
useQuickRank	False	用 AI 重排提升相关性
useGroupRank	False	对相关文档分组重排
followUpQuestions	5	为对话生成后续问题数量
objectIds / chunkIds / tableIds	None	按 ID 精确圈定检索范围

发送 Chat 请求

chat() 是关键字参数（keyword-only）方法，签名如下（chat.py 混入）：

async def chat(self, *, token: str, question: Question, on_sse=None) -> PIPELINE_RESULT

用法示例：

response = await client.chat(token=token, question=question)

参数说明：

• token：client.use() 返回的 pipeline token，标识本次会话的资源与鉴权；
• question：上一步构建好的 Question 对象，不允许为空，否则抛出 RuntimeError('Question cannot be empty')；
• on_sse：可选回调，用于接收服务器推送的 server-sent events（典型场景是逐 token 流式输出）。它会被透传给底层 pipe() 调用，最终答案仍可从结果 body 中取得。

chat() 的返回值是 PIPELINE_RESULT 字典，包含 answers（AI 答案字符串列表）、result_types、objectId、name、path 等字段。发送前需要注意：Question 内部会为每次调用生成自增序号，用于 objinfo['name']，例如 "Question 1"、"Question 2"，方便在运行日志中定位。

用 Answer 解析响应

Answer 负责从 AI 文本中提取结构。AI 的原始输出经常包裹在 Markdown 或代码围栏中，因此 SDK 明确不会自动把 Answer 附加到结果上——你需要自己取出 body 再喂给它：

from rocketride.schema import Answer

answer_text = (response.get('answers') or [None])[0]
answer = Answer(expectJson=True)
answer.setAnswer(answer_text or '')
if answer.isJson():
    structured = answer.getJson()
else:
    structured = answer.getText()

关键语义（务必牢记）

• setAnswer(value)：存储响应值；当 expectJson=True 时，若传入 dict/list 直接存储，若传入字符串则尝试 json.loads 解析，解析失败会抛出 ValueError('Expected a JSON-compatible answer (dict or list).')（question.py）。
• isJson()：返回的是 expectJson 标志位本身，并不检查内容。所以判断依据是"你是否要求了 JSON"，而不是"内容是不是 JSON"。
• getJson()：返回已解析的 JSON 结构；仅当从未 setAnswer 时才返回 None，一旦存入了非法 JSON 内容则抛出 ValueError('Answer is not in JSON format.')。
• getText()：以纯文本返回答案；若内部存的是 dict/list，会自动 json.dumps 序列化。
• parsePython(value)：从含 ```python 围栏的 AI 输出中剥离围栏标记，提取干净 Python 代码。
• answer.tokens：携带服务器上报的本轮 LLM token 用量（含 input/output/cache/model/calls 及逐次调用的 breakdown 列表），可用于用量监控与成本分析；注意计费走独立的 metrics.counter 路径，不受该字段影响（question.py 字段注释）。

此外 Answer 在创建时即校验：answer 字段不允许在构造时直接赋值，必须通过 setAnswer() 写入，防止绕过校验逻辑（question.py）。

完整可运行示例

将上面的片段组合成一个端到端程序（与 examples.md 中的示例 6 一致）：

import asyncio
from rocketride import RocketRideClient
from rocketride.schema import Question, Answer


async def main():
    async with RocketRideClient(uri='https://api.rocketride.ai', auth='my-key') as client:
        result = await client.use(filepath='chat_pipeline.pipe')
        token = result['token']

        question = Question(expectJson=True)
        question.addInstruction('Format', 'Return a JSON object with keys: summary, keywords.')
        question.addExample('Summarize X', {'summary': '...', 'keywords': ['a', 'b']})
        question.addQuestion('Summarize the main points and list keywords.')

        response = await client.chat(token=token, question=question)
        answer_text = (response.get('answers') or [None])[0]

        answer = Answer(expectJson=True)
        answer.setAnswer(answer_text or '')
        structured = answer.getJson() if answer.isJson() else answer.getText()
        print(structured)

        await client.terminate(token)


asyncio.run(main())

多轮对话：结合 history 与 role

addHistory 与 role 组合即可实现有状态的连续对话：

from rocketride.schema import Question, QuestionHistory

question = Question(role='You are a patient technical tutor.')
question.addHistory(QuestionHistory(role='user', content='What is a pipeline?'))
question.addHistory(QuestionHistory(role='assistant', content='A pipeline is a graph of nodes...'))
question.addQuestion('Now explain what a node is.')

历史消息会按 role: content 的顺序渲染进提示词的 ### Conversation History: 区块，让模型感知前文语境。

流式输出：on_sse 回调

当需要逐 token 展示输出时，传入 on_sse：

async def on_sse(event):
    # event 结构由服务器 SSE 推送决定，通常是 {event: ..., body: ...}
    print(event.get('event'), event.get('body'))

response = await client.chat(token=token, question=question, on_sse=on_sse)

回调以异步方式接收服务器事件，最终完整答案依然会出现在返回结果的 answers 字段中，流式回调与整体结果二者并存。

配套的 Chat Pipeline 是怎么配置的

client.chat() 之所以能工作，前提是目标 pipeline 的源是 chat 组件。仓库测试目录下的 chat_pipeline.py 给出了完整的动态配置范式，其结构为：

{
    'components': [
        {'id': 'chat_1', 'provider': 'chat',
         'config': {'hideForm': True, 'mode': 'Source', 'type': 'chat'}},
        llm_component,   # 动态选择 OpenAI / Anthropic / Gemini / Ollama
        {'id': 'response_1', 'provider': 'response',
         'config': {'lanes': []},
         'input': [{'lane': 'answers', 'from': llm_component['id']}]},
    ],
    'source': 'chat_1',
    'project_id': '8b866c3b-6c76-42d7-8091-301be3dce0f2',
}

要点：

1. chat 源组件：provider: 'chat'，config.type: 'chat'，作为 pipeline 的 source；
2. LLM 组件：以 input: [{'lane': 'questions', 'from': 'chat_1'}] 监听 questions lane——这正是 application/rocketride-question MIME 类型被路由到的 lane（引擎侧判定见 data_conn.py）；模型提供方按环境变量 ROCKETRIDE_OPENAI_KEY / ROCKETRIDE_ANTHROPIC_KEY / ROCKETRIDE_GEMINI_KEY / ROCKETRIDE_OLLAMA_HOST 优先级动态选取；
3. response 组件：监听 LLM 的 answers lane，负责把答案装配回 answers 结果字段。

另外，仓库的 TypeScript 客户端与 n8n 节点也实现了同一条 application/rocketride-question 通道（分别见 client.ts 与 helpers.ts），说明该 MIME 约定是多语言客户端共用的协议标准。

常见问题与最佳实践

• 我的 pipeline 源是 webhook/dropper，能用 chat() 吗？ 可以。chat() 对 chat、webhook、dropper 三种源都适用，但不设 provider 过滤，具体效果取决于 pipeline 内如何消费 questions lane。
• 为什么 getJson() 会抛异常？ 因为 expectJson=True 时，setAnswer() 与 getJson() 都对 JSON 合法性做严格校验；拿到 isJson() 为 True 并不意味着内容合法，务必用 try/except ValueError 兜底。
• 如何拿到干净的 Python 代码？ 用 Answer.parsePython(raw)，它会自动剥离 ```python ... ``` 围栏。
• 结构化输出不稳定怎么办？ 充分利用 addExample 给出目标格式范本，并让 addInstruction 明确字段清单；SDK 本身已会在 expectJson=True 时自动注入 JSON 格式约束与重试修复指令。
• 多轮对话丢上下文？ 每次请求把之前的 QuestionHistory 通过 addHistory 带回，配合 role 人设稳定对话风格。

延伸阅读

• Chat 官方文档（本文对应的 SDK 指南原文）
• Python SDK 示例合集（含完整的示例 6）
• API 参考：Question / Answer
• RocketRide Python SDK 总览
• Question 模型实现 与 Chat 混入实现