企业官网建设流程全解析

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两个生成器组件的设计原理、完整参数体系、流式输出、工具调用（Tool/Toolset）与 Pipeline 编排方式。读完本文，你将能够在自己的 RAG 或 Agent 应用中直接接入 Together AI 托管的开源大模型（如 Llama-3.3-70B-Instruct），并通过generation_kwargs精细控制生成行为。本文以 version-2.20 的集成 API 参考 为骨架，并结合 核心源码 与配套使用文档进行纵深展开。

集成定位：OpenAI 兼容层之上的 Together AI 生成器

Together AI 提供云托管的开源模型推理服务，其 chat completion 端点与 OpenAI API 高度兼容。Haystack 正是利用这一兼容性设计了本集成：

• TogetherAIChatGenerator直接继承自 Haystack 核心组件OpenAIChatGenerator（基类源码位于 haystack/components/generators/chat/openai.py），复用其 OpenAI Python SDK 客户端、消息归一化、流式分块处理与序列化逻辑；
• TogetherAIGenerator又继承自TogetherAIChatGenerator，在聊天生成能力之上封装出面向单条文本提示（prompt string）的简单接口。

从源码结构可以推断，这一继承关系意味着两个组件天然具备与 OpenAI 生态一致的请求/响应处理链路：默认走https://api.together.xyz/v1这一 OpenAI 兼容 base URL，因此可以直接复用OPENAI_TIMEOUT（默认 30 秒）与OPENAI_MAX_RETRIES（默认 5 次）环境变量来控制客户端行为（对应实现见 openai.py 的_client_kwargs方法）。

安装与 API Key 配置

使用前需要先安装集成包：

pip install togetherai-haystack

运行需要满足两个前提：

1. 拥有一个有效的 Together AI 账号，且账户内有足够额度；
2. 具备 API Key，按以下任一方式提供：
   • 推荐方式：设置环境变量TOGETHER_API_KEY，这也是组件的默认行为——api_key参数的默认值即为Secret.from_env_var("TOGETHER_API_KEY")；
   • 显式传参：使用 Haystack 的 Secret 管理机制 传入，如api_key=Secret.from_token("your-api-key-here")。

组件默认模型为meta-llama/Llama-3.3-70B-Instruct-Turbo，默认 API 基址为https://api.together.xyz/v1。若你的网络环境需要代理或服务商提供其他端点，可通过api_base_url覆盖。

TogetherAIChatGenerator：面向多轮聊天的生成器

构造函数签名与参数详解

__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

各参数含义如下：

generation_kwargs 支持参数速查（聊天端点）

generation_kwargs中的所有键值对都会被原样透传到 Together AI chat completion 端点，因而支持端点定义的全部参数。API 参考中明确列出的常用项包括：

工具调用：Tool 与 Toolset 的灵活组合

TogetherAIChatGenerator通过tools参数支持函数调用（function calling）。tools的类型为ToolsType，其灵活性体现在三种组织方式：

• 单个Tool列表：将独立工具作为列表传入；
• 单个Toolset：直接把整个工具集传入；
• 混合模式：同一列表中同时混入多个Toolset与独立Tool。

参考 使用文档 的示例：

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] # Toolset 与 Tool 的混合 )

底层实现中，OpenAIChatGenerator在初始化时会调用_check_duplicate_tool_names校验工具名唯一性，并在warm_up()阶段调用warm_up_tools预热工具（见 openai.py 初始化与预热逻辑）。更完整的工具定义方法可参考 Tool 文档 与 Toolset 文档。

序列化

TogetherAIChatGenerator提供to_dict()方法将组件序列化为字典（返回dict[str, Any]），便于在 Pipeline 的 YAML 或 JSON 编排中持久化与复用。序列化时工具与回调函数会以可反序列化的形式存储。

快速上手：TogetherAIChatGenerator 使用示例

独立使用

API 参考给出的最小示例：

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)

输出示例（ChatMessage携带模型名、索引、结束原因与 token 用量）：

>>{'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}})]}

输入输出均采用 Haystack 的ChatMessage格式。ChatMessage数据类（源码见 haystack/dataclasses/chat_message.py）封装了消息内容、角色（user/system/assistant/tool，对应ChatRole枚举）以及可选元数据，保证多轮对话上下文的连贯性。

启用流式输出

流式模式下，token 生成即被回调消费，显著降低首 token 延迟感知：

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

聊天场景下最典型的位置是接在ChatPromptBuilder之后，由 builder 生成消息序列，再喂给生成器：

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的完整用法可参考 ChatPromptBuilder 文档。

TogetherAIGenerator：面向纯文本提示的生成器

TogetherAIGenerator继承自TogetherAIChatGenerator，在内部把字符串 prompt 包装为ChatMessage后再走聊天端点，因此对外表现为"输入字符串、输出字符串列表"的简洁接口。

构造函数签名与参数

__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

除与聊天版本相同的参数外，本组件独有的关键参数是system_prompt：

• system_prompt（str | None）：用于设定生成任务的上下文或指令。未提供时 system prompt 会被省略，此时采用模型自带的默认系统提示词。

此外，两个版本在超时与重试的默认取值策略上保持一致：timeout未设置时取OPENAI_TIMEOUT环境变量，否则为 30 秒；max_retries未设置时取OPENAI_MAX_RETRIES环境变量，否则为 5。

generation_kwargs 支持参数速查（文本生成）

API 参考在文本生成版本中列出的参数覆盖更广，除前文提到的max_tokens、temperature、top_p外还包括：

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__中设置的 system prompt；
• streaming_callback：可选，运行时传入会覆盖初始化时的回调；
• generation_kwargs：运行时的额外生成参数，会逐键合并并优先于初始化时传入的同名参数；
• 返回值：字典包含两个键——
  • replies：生成的文本补全字符串列表；
  • meta：与每条生成结果对应的元数据字典列表，包含模型名、结束原因（finish reason）与 token 用量统计。

异步版本run_async拥有完全相同的参数与返回值，可直接在asyncio环境中await调用。

序列化

TogetherAIGenerator同时提供to_dict()（序列化）与from_dict(data)（反序列化）两个方法。from_dict接收组件的字典表示并返回TogetherAIGenerator实例，这使得基于 YAML/JSON 的 Pipeline 配置能够无损还原组件状态。

快速上手：TogetherAIGenerator 使用示例

独立使用（含 generation_kwargs）

API 参考中的示例同时演示了generation_kwargs的传法：

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?"))

带 system prompt 的调用

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

组装 RAG Pipeline

在 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', ...}]}}

PromptBuilder的模板语法与更多用法可参考 PromptBuilder 文档。

源码级原理：继承链与客户端行为

要真正理解这两个组件，值得回到基类源码确认几个关键行为：

1. 客户端初始化延迟到warm_up：OpenAIChatGenerator在构造时仅保存参数，真正的同步/异步 OpenAI 客户端在warm_up()/warm_up_async()时才按需创建（见 openai.py）。这意味着组件可以在不触发网络连接的情况下安全地实例化、序列化与反序列化。

2. 超时与重试的默认值注入：_client_kwargs()方法（openai.py）展示了默认值的最终来源——OPENAI_TIMEOUT（默认 30.0）与OPENAI_MAX_RETRIES（默认 5），随后连同base_url、api_key一起传给 OpenAI SDK。这正是 API 参考中"未设置时取环境变量"说法的实现依据。

3. 消息归一化与流式分块：run内部先调用_normalize_messages处理输入，再根据是否设置回调决定走流式（_handle_stream_response，逐 chunk 触发streaming_callback）还是非流式路径，最终统一转换为ChatMessage列表返回。StreamingChunk与FinishReason等类型定义在 haystack/dataclasses/streaming_chunk.py。

4. 工具调用链路：工具列表在初始化时做重名校验（_check_duplicate_tool_names），在warm_up时统一预热（warm_up_tools），调用时扁平化后随请求发送，模型返回的工具调用被解析进ChatMessage的ToolCall字段，从而支撑 Agent 式多轮工具编排。

选型建议与注意事项

• 聊天场景选TogetherAIChatGenerator，纯文本补全选TogetherAIGenerator：前者接受ChatMessage列表、天然适合多轮对话与工具调用；后者接受字符串、适合一次性的文本补全。若对话历史或工具调用不是刚需，二者皆可。另可参考 如何选择合适的生成器 的对比说明。
• 从最新文档看，TogetherAIGenerator已被标记为弃用，官方推荐改用TogetherAIChatGenerator——后者同样接受纯字符串输入（run会将字符串归一化为单条 user 消息），且功能完全覆盖前者。新项目建议直接以TogetherAIChatGenerator为入口。
• response_format与流式不可混用 Pydantic 模型：如需同时启用结构化输出与流式，response_format必须传 JSON Schema 而非 Pydantic 模型。
• 默认端点与模型：组件默认指向https://api.together.xyz/v1与meta-llama/Llama-3.3-70B-Instruct-Turbo，实际可用模型清单以 Together AI 官方文档为准。
• 密钥管理：优先通过TOGETHER_API_KEY环境变量注入密钥，避免密钥硬编码进代码或 Pipeline 配置；序列化（to_dict）时 API Key 以Secret形式保留，配合 Haystack 的 Secret 管理 机制使用。

延伸阅读

• 本集成 API 参考：docs-website/reference_versioned_docs/version-2.20/integrations-api/togetherai.md
• TogetherAIChatGenerator 使用文档
• TogetherAIGenerator 使用文档
• 基类实现：OpenAIChatGenerator 源码
• 消息与流式数据类：ChatMessage 源码、StreamingChunk 源码
• 工具体系：Tool 文档、Toolset 文档

【免费下载链接】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