Haystack 集成 Together AI 生成组件全指南：TogetherAIChatGenerator 与 TogetherAIGenerator 实战-迪斯科星球

Haystack 集成 Together AI 生成组件全指南：TogetherAIChatGenerator 与 TogetherAIGenerator 实战

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack

本文围绕 Haystack 仓库中 Together AI 官方集成（togetherai-haystack）的两个核心生成组件展开：TogetherAIChatGenerator（对话补全）与TogetherAIGenerator（文本生成）。你将掌握 API Key 配置、generation_kwargs全参数清单、流式输出、工具调用（Tool/Toolset）以及将两者接入 Haystack Pipeline 完成 RAG 问答的完整方法，并结合仓库源码理解其底层实现原理。

集成概览：两个组件如何分工

Together AI 集成位于haystack_integrations.components.generators.togetherai模块，提供两个生成器组件：

TogetherAIChatGenerator：基于OpenAIChatGenerator，面向对话补全（chat completion）场景，输入输出均为ChatMessage结构，支持工具调用与多轮对话。
TogetherAIGenerator：继承自TogetherAIChatGenerator，提供更轻量的文本生成接口，直接接收字符串prompt，适合与PromptBuilder搭配的经典生成流水线。

两个组件的默认模型均为meta-llama/Llama-3.3-70B-Instruct-Turbo，默认 API 基地址为 Together AI 的 OpenAI 兼容端点https://api.together.xyz/v1。支持的模型全集请以 Together AI 官方模型列表为准，组件层面不做硬编码白名单。

环境准备：安装与 API Key 配置

安装集成包：

pip install togetherai-haystack

使用前提：需要有效的 Together AI 账号、足够的额度以及 API Key。API Key 有两种提供方式（官方推荐环境变量）：

设置TOGETHER_API_KEY环境变量（推荐）；
在__init__中通过api_key参数传入 Haystack Secret 对象，例如Secret.from_token("your-api-key-here")。

使用Secret而非明文传参是 Haystack 的安全最佳实践：密钥在组件序列化往返（to_dict/from_dict）过程中不会泄漏，详见 Secret 管理指南。

关于超时与重试的环境变量：若未显式传入timeout与max_retries，组件会回退读取OPENAI_TIMEOUT（默认 30 秒）与OPENAI_MAX_RETRIES（默认 5 次）。这一回退逻辑继承自基类 openai.py 中的_client_kwargs实现，可结合pip show togetherai-haystack查看实际安装版本对应的源码。

TogetherAIChatGenerator：对话补全组件

核心特性与兼容性

主兼容目标：无缝对接 Together AI chat completion 端点；
流式输出：支持从端点流式接收响应 token；
高可定制性：支持 Together AI chat completion 端点支持的全部参数（通过generation_kwargs透传）；
ChatMessage 结构化：输入输出均使用ChatMessage格式，保证多轮对话上下文连贯。

ChatMessage是 Haystack 的数据类，包含消息内容、角色（user、assistant、system等）与可选元数据，定义于 chat_message.py，常用工厂方法包括ChatMessage.from_user(...)、ChatMessage.from_system(...)。

初始化参数（init）

__init__( *, api_key: Secret = Secret.from_env_var("TOGETHER_API_KEY"), model: str = "meta-llama/Llama-3.3-70B-Instruct-Turbo", streaming_callback: StreamingCallbackT | None = None, api_base_url: str | None = "https://api.together.xyz/v1", generation_kwargs: dict[str, Any] | None = None, tools: ToolsType | None = None, timeout: float | None = None, max_retries: int | None = None, http_client_kwargs: dict[str, Any] | None = None ) -> None

参数	类型	说明
`api_key`	`Secret`	Together API Key，默认读取`TOGETHER_API_KEY`环境变量
`model`	`str`	Together AI chat completion 模型名称
`streaming_callback`	`StreamingCallbackT \\| None`	流式回调，每收到一个新 token 时调用，回调参数为`StreamingChunk`
`api_base_url`	`str \\| None`	API 基地址，默认`https://api.together.xyz/v1`（Together AI 提供 OpenAI 兼容端点），可覆盖
`generation_kwargs`	`dict[str, Any] \\| None`	透传给 Together AI 端点的生成参数，详见下方清单
`tools`	`ToolsType \\| None`	供模型调用准备的 Tool 与/或 Toolset 列表，或单个 Toolset，每个工具名称必须唯一
`timeout`	`float \\| None`	API 调用超时时间
`max_retries`	`int \\| None`	内部错误后重试次数上限，未设置时回退`OPENAI_MAX_RETRIES`环境变量，默认 5
`http_client_kwargs`	`dict[str, Any] \\| None`	自定义`httpx.Client`/`httpx.AsyncClient`的参数字典

generation_kwargs 支持参数清单

所有参数均直接透传给 Together AI chat completion 端点，以下是官方文档列出的常用项：

max_tokens：输出文本的最大 token 数；
temperature：采样温度，越高越冒险。创意类应用建议 0.9，答案明确的任务建议 0（argmax 采样）；
top_p：核采样（nucleus sampling），模型只考虑累计概率质量达top_p的 token。0.1 表示只考虑概率质量前 10% 的 token；
stream：是否流式返回部分进度。开启后 token 以>from haystack.tools import Tool, Toolset from haystack_integrations.components.generators.togetherai import TogetherAIChatGenerator # 创建独立工具 weather_tool = Tool(name="weather", description="Get weather info", ...) news_tool = Tool(name="news", description="Get latest news", ...) # 将相关工具归组为 toolset math_toolset = Toolset([add_tool, subtract_tool, multiply_tool]) # 混合传入 Toolset 与 Tool generator = TogetherAIChatGenerator( tools=[math_toolset, weather_tool, news_tool] )
Tool 与 Toolset 的详细用法可参考 Tool 文档与 Toolset 文档。
独立使用示例
```
from haystack_integrations.components.generators.togetherai import TogetherAIChatGenerator from haystack.dataclasses import ChatMessage messages = [ChatMessage.from_user("What's Natural Language Processing?")] client = TogetherAIChatGenerator() response = client.run(messages) print(response) >>{'replies': [ChatMessage(_content='Natural Language Processing (NLP) is a branch of artificial intelligence >>that focuses on enabling computers to understand, interpret, and generate human language in a way that is >>meaningful and useful.', _role=<ChatRole.ASSISTANT: 'assistant'>, _name=None, >>_meta={'model': 'meta-llama/Llama-3.3-70B-Instruct-Turbo', 'index': 0, 'finish_reason': 'stop', >>'usage': {'prompt_tokens': 15, 'completion_tokens': 36, 'total_tokens': 51}})]}
```
可见返回的replies中_meta携带模型名、完成原因（finish_reason）与 token 用量统计（usage），便于追踪成本与生成质量。
流式输出示例
```
from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.togetherai import TogetherAIChatGenerator client = TogetherAIChatGenerator( model="meta-llama/Llama-3.3-70B-Instruct-Turbo", streaming_callback=lambda chunk: print(chunk.content, end="", flush=True), ) response = client.run([ChatMessage.from_user("What are Agentic Pipelines? Be brief.")]) # 查看响应用到的模型 print("\n\nModel used:", response["replies"][0].meta.get("model"))
```
流式回调在 Pipeline 与独立使用场景中均生效，关于StreamingChunk的类型说明可参考选择生成器指南。
序列化
```
to_dict() -> dict[str, Any]
```
将组件序列化为字典，用于 Pipeline 的 YAML 导出、断点调试与远程执行等场景。
TogetherAIGenerator：文本生成组件
TogetherAIGenerator继承自TogetherAIChatGenerator，提供面向纯文本补全的简化接口。
初始化参数（init）
```
__init__( api_key: Secret = Secret.from_env_var("TOGETHER_API_KEY"), model: str = "meta-llama/Llama-3.3-70B-Instruct-Turbo", api_base_url: str | None = "https://api.together.xyz/v1", streaming_callback: StreamingCallbackT | None = None, system_prompt: str | None = None, generation_kwargs: dict[str, Any] | None = None, timeout: float | None = None, max_retries: int | None = None, ) -> None
```
与 Chat 版本相比，多出system_prompt（可选系统提示词；未提供时省略，使用模型默认系统提示词），且不暴露tools与http_client_kwargs。
generation_kwargs在文本生成场景的常用参数除温度、top_p 外还包括：n（每个 prompt 的补全数，如 3 个 prompt、n=2 时共生成 6 个补全）、stop、presence_penalty、frequency_penalty、logit_bias等（清单见前文）。
timeout未设置时回退OPENAI_TIMEOUT（默认 30），max_retries未设置时回退OPENAI_MAX_RETRIES（默认 5）。
run / run_async
```
run( *, prompt: str, system_prompt: str | None = None, streaming_callback: StreamingCallbackT | None = None, generation_kwargs: dict[str, Any] | None = None ) -> dict[str, Any]
```
- prompt：文本生成的输入提示词（必填）；
- system_prompt：可选，未提供时使用__init__中设置的系统提示词；
- streaming_callback：可选，提供时覆盖__init__中的回调；
- generation_kwargs：可选，会覆盖__init__中同名参数（支持temperature、max_new_tokens、top_p等）。
返回字典包含两个键：
- replies：生成的文本补全字符串列表；
- meta：每个补全的元数据字典列表（模型名、完成原因、token 用量统计）。
run_async签名与run完全一致，用于异步生成文本补全，可无缝接入 Haystack 的异步 Pipeline 执行环境。
使用示例
```
from haystack_integrations.components.generators.togetherai import TogetherAIGenerator generator = TogetherAIGenerator(model="deepseek-ai/DeepSeek-R1", generation_kwargs={ "temperature": 0.9, }) print(generator.run("Who is the best Italian actor?"))
```
带系统提示词：
```
from haystack_integrations.components.generators.togetherai import TogetherAIGenerator client = TogetherAIGenerator( model="meta-llama/Llama-3.3-70B-Instruct-Turbo", system_prompt="You are a helpful assistant that provides concise answers.", ) response = client.run("What's Natural Language Processing?") print(response["replies"][0])
```
带流式输出：
```
from haystack_integrations.components.generators.togetherai import TogetherAIGenerator client = TogetherAIGenerator( model="meta-llama/Llama-3.3-70B-Instruct-Turbo", streaming_callback=lambda chunk: print(chunk.content, end="", flush=True), ) response = client.run("What's Natural Language Processing? Be brief.") print(response)
```
反序列化
```
from_dict(data: dict[str, Any]) -> TogetherAIGenerator
```
从字典恢复组件实例，与to_dict配对使用，保证组件可在序列化-反序列化往返中无损重建。
底层实现原理：基于 OpenAIChatGenerator 的继承设计
两个组件均建立在 Haystack 核心的OpenAIChatGenerator之上（源码见 openai.py），继承链为：
```
TogetherAIChatGenerator → OpenAIChatGenerator TogetherAIGenerator → TogetherAIChatGenerator
```
这正是"OpenAI 兼容端点"设计的体现：Together AI 提供与 OpenAI 协议兼容的 REST 端点，因此集成只需在基类基础上固定默认api_base_url为https://api.together.xyz/v1并将默认api_key环境变量改为TOGETHER_API_KEY，即可复用基类完整的客户端管理、流式处理、工具预热与序列化逻辑。
从源码结构看，基类在_client_kwargs（openai.py）中统一解析timeout、max_retries与base_url，并在warm_up/warm_up_async（openai.py）中延迟初始化同步/异步 OpenAI 客户端；工具重名检查（_check_duplicate_tool_names）与工具预热（warm_up_tools）也由基类统一完成，TogetherAIChatGenerator直接继承这些能力。
在 Pipeline 中实战：RAG 与聊天流水线
文本生成 + PromptBuilder 的 RAG 流水线
将TogetherAIGenerator置于PromptBuilder之后，配合检索器实现检索增强生成：
```
from haystack import Pipeline, Document from haystack.components.retrievers.in_memory import InMemoryBM25Retriever from haystack.components.builders.prompt_builder import PromptBuilder from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack_integrations.components.generators.togetherai import TogetherAIGenerator docstore = InMemoryDocumentStore() docstore.write_documents([ Document(content="Rome is the capital of Italy"), Document(content="Paris is the capital of France") ]) query = "What is the capital of France?" template = """ Given the following information, answer the question. Context: {% for document in documents %} {{ document.content }} {% endfor %} Question: {{ query }}? """ pipe = Pipeline() pipe.add_component("retriever", InMemoryBM25Retriever(document_store=docstore)) pipe.add_component("prompt_builder", PromptBuilder(template=template)) pipe.add_component("llm", TogetherAIGenerator(model="meta-llama/Llama-3.3-70B-Instruct-Turbo")) pipe.connect("retriever", "prompt_builder.documents") pipe.connect("prompt_builder", "llm") result = pipe.run({ "prompt_builder": {"query": query}, "retriever": {"query": query} }) print(result) >> {'llm': {'replies': ['The capital of France is Paris.'], >> 'meta': [{'model': 'meta-llama/Llama-3.3-70B-Instruct-Turbo', ...}]}}
```
聊天生成 + ChatPromptBuilder 流水线
```
from haystack import Pipeline from haystack.components.builders import ChatPromptBuilder from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.togetherai import ( TogetherAIChatGenerator, ) prompt_builder = ChatPromptBuilder() llm = TogetherAIChatGenerator(model="meta-llama/Llama-3.3-70B-Instruct-Turbo") pipe = Pipeline() pipe.add_component("builder", prompt_builder) pipe.add_component("llm", llm) pipe.connect("builder.prompt", "llm.messages") messages = [ ChatMessage.from_system("Give brief answers."), ChatMessage.from_user("Tell me about {{city}}"), ] response = pipe.run( data={"builder": {"template": messages, "template_variables": {"city": "Berlin"}}}, ) print(response)
```
ChatPromptBuilder负责按模板变量渲染ChatMessage列表，TogetherAIChatGenerator消费messages输入，二者通过builder.prompt → llm.messages连接，构成可复用的多轮对话流水线。
总结
Together AI 集成以TogetherAIChatGenerator与TogetherAIGenerator两个组件覆盖了对话补全与文本生成两大场景：
- 面向多轮对话与工具调用，使用TogetherAIChatGenerator（输入ChatMessage，支持tools与response_format结构化输出）；
- 面向PromptBuilder驱动的经典 RAG 流水线，使用TogetherAIGenerator（输入字符串prompt，可附加system_prompt）；
- 两者共享 OpenAI 兼容端点设计、generation_kwargs透传机制、流式回调与序列化能力，默认模型为meta-llama/Llama-3.3-70B-Instruct-Turbo，均通过TOGETHER_API_KEY环境变量完成认证。
建议读者继续阅读仓库内的 TogetherAIGenerator 组件文档、TogetherAIChatGenerator 组件文档以及基类实现 openai.py，以便在真实项目中按需定制参数与扩展行为。
【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析