在 Haystack 中接入 Together AITogetherAIChatGenerator 与 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 官方参考文档 togetherai.md 为核心骨架全面讲解如何在 Haystack 中通过TogetherAIChatGenerator与TogetherAIGenerator两个生成器组件调用 Together AI 托管的开源大模型。你将掌握组件初始化参数、generation_kwargs生成参数、流式输出、Tool/Toolset 函数调用、结构化输出、Pipeline 集成以及序列化等完整能力并能在自己的 RAG 与对话应用中直接落地使用。一、集成概览两个组件两条生成路径Together AI 为 Haystack 提供了一个官方集成包togetherai-haystack其核心是haystack_integrations.components.generators.togetherai模块下的两个组件TogetherAIChatGenerator基于OpenAIChatGenerator实现见参考文档Bases: OpenAIChatGenerator面向**聊天补全chat completion**场景输入输出均为ChatMessage对象。TogetherAIGenerator继承自TogetherAIChatGenerator面向纯文本生成场景输入一个 prompt 字符串返回文本列表与元数据。之所以两个组件能共用一套底层是因为 Together AI 提供了OpenAI 兼容的 API 端点默认https://api.together.xyz/v1。Together AI 官方文档将这种兼容性作为核心特性参考文档中也明确指出该集成的三大关键能力主要兼容性与 Together AI chat completion 端点无缝协作流式支持支持从端点流式接收响应高度可定制支持 Together AI chat completion 端点支持的所有参数。两个组件都基于ChatMessage格式组织输入输出确保在聊天式文本生成场景中返回连贯、上下文相关的回复。ChatMessage的详细说明见 ChatMessage 数据类文档 及源码 chat_message.py。默认模型为meta-llama/Llama-3.3-70B-Instruct-Turbo完整支持模型列表以 Together AI 官方文档为准参考文档明确指引读者查阅 Together AI 文档确认可用模型。二、环境准备安装、API Key 与端点配置2.1 安装集成包两个组件都来自togetherai-haystack集成包使用前需安装pip install togetherai-haystack该包是独立的第三方集成维护于 deepset-ai/haystack-core-integrations 仓库安装后即可与当前仓库中的 Haystack 核心协同工作。2.2 提供 API Key使用该集成需要有效的 Together AI 订阅、充足额度与 API Key。参考文档与使用指南给出了两种方式方式一推荐设置环境变量export TOGETHER_API_KEYyour-together-api-keyTogetherAIChatGenerator和TogetherAIGenerator的api_key参数默认值都是Secret.from_env_var(TOGETHER_API_KEY)因此不传该参数时会自动从环境变量读取。方式二通过 Haystack Secret 机制显式传入from haystack.utils import Secret generator TogetherAIChatGenerator(api_keySecret.from_token(your-api-key-here))Haystack 的Secret抽象源码见 auth.py支持from_token()token 型不可序列化与from_env_var()环境变量型可接受单个或多个候选变量名strictTrue时若全部未设置会抛出异常两种构造方式。更多细节可参考 Secret 管理概念文档。2.3 端点Base URL配置两个组件的api_base_url默认均为https://api.together.xyz/v1即 Together AI 的 OpenAI 兼容端点。如需接入代理或自定义网关可通过该参数覆盖。三、TogetherAIChatGenerator聊天补全组件详解3.1 初始化参数全解TogetherAIChatGenerator的__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_keySecretTogether API Key默认读取TOGETHER_API_KEY环境变量modelstr使用的 Together AI chat completion 模型名默认meta-llama/Llama-3.3-70B-Instruct-Turbostreaming_callbackStreamingCallbackT \| None流式回调函数每当流中收到新 token 时被调用回调接收StreamingChunk作为参数api_base_urlstr \| NoneTogether AI API 基础地址默认https://api.together.xyz/v1generation_kwargsdict[str, Any] \| None直接透传给 Together AI 端点的其他生成参数toolsToolsType \| NoneTool 和/或 Toolset 对象列表或单个 Toolset供模型准备函数调用每个工具名须唯一timeoutfloat \| NoneTogether AI API 调用超时时间max_retriesint \| None遇到内部错误后重试次数上限未设置时取OPENAI_MAX_RETRIES环境变量否则默认 5http_client_kwargsdict[str, Any] \| None配置自定义httpx.Client/httpx.AsyncClient的关键字参数字典3.2 generation_kwargs透传 Together AI 生成参数参考文档强调任何 Together AI chat completion API 合法参数都可以通过__init__中的generation_kwargs或run方法中的generation_kwargs直接传入。参考文档列出的常用参数包括max_tokens输出文本的最大 token 数。temperature采样温度值越高模型越冒险。创造性任务可试0.9有明确答案的任务建议0argmax 采样。top_p核采样nucleus sampling替代方案只考虑概率质量累计到top_p的 token。如0.1表示只考虑概率质量前 10% 的 token。stream是否流式返回部分进度。开启后 token 以>from haystack_integrations.components.generators.togetherai import TogetherAIChatGenerator from haystack.dataclasses import ChatMessage messages [ChatMessage.from_user(Whats Natural Language Processing?)] client TogetherAIChatGenerator() response client.run(messages) print(response)输出示例replies中包含一个ChatMessage其_meta携带模型名、索引、结束原因与 token 用量{replies: [ChatMessage(_contentNatural 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., _roleChatRole.ASSISTANT: assistant, _nameNone, _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}})]}从底层看参考 openai.py 中OpenAIChatGenerator的实现run首先调用warm_up()初始化 OpenAI 兼容客户端将ChatMessage列表通过message.to_openai_dict_format()归一化为 OpenAI 消息格式再调用client.chat.completions端点。非流式场景下响应通过_convert_chat_completion_to_chat_message转换回ChatMessage含ToolCall解析并记录model、index、finish_reason、usage等元数据。3.4 消息构造ChatMessage 的四种角色TogetherAIChatMessageGenerator输入ChatMessage对象。ChatMessage是 Haystack 中面向 LLM 的消息核心抽象包含角色、元数据及多种内容类型文本、图片、文件、工具调用、工具调用结果、推理内容通过四个类方法构造源码见 chat_message.pyChatMessage.from_user(用户的提问) ChatMessage.from_system(你是乐于助人的助手) ChatMessage.from_assistant(助手的回复) ChatMessage.from_tool(tool_result, origintool_call)from_user还支持通过content_parts传入文本与图片/文件的混合内容为多模态对话留出扩展空间。对话历史就是由这些消息按时间顺序组成的列表。3.5 流式输出Streaming参考文档指出组件支持流式响应token 生成时即可被消费。启用方式是在初始化时传入streaming_callback可调用对象from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.togetherai import TogetherAIChatGenerator client TogetherAIChatGenerator( modelmeta-llama/Llama-3.3-70B-Instruct-Turbo, streaming_callbacklambda chunk: print(chunk.content, end, flushTrue), ) response client.run([ChatMessage.from_user(What are Agentic Pipelines? Be brief.)]) print(\n\nModel used:, response[replies][0].meta.get(model))底层实现中见 openai.py流式响应逐 chunk 被_convert_chat_completion_chunk_to_streaming_chunk转换为StreamingChunk并立即回调用户函数同时累积到_convert_streaming_chunks_to_chat_message最终合并为完整的ChatMessage。此外streaming_callback也可以在run调用时动态传入以覆盖初始化时的设置。3.6 在 Pipeline 中使用该组件在流水线中最常见的位置是ChatPromptBuilder 之后接收其生成的 prompt 消息。使用指南给出的完整示例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(modelmeta-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的prompt输出连接到llm的messages输入模板变量{{city}}在运行时被动态填充。此模式正是构建用户问题 → 提示模板 → Together AI 生成对话链路的典型范式。3.7 Tool / Toolset 函数调用TogetherAIChatGenerator通过tools参数支持函数调用function calling使用指南列出了三种灵活的配置方式Tool 对象列表逐个传入独立工具单个 Toolset直接传入整个工具集混合模式将多个 Toolset 与独立 Tool 组合在同一个列表中。from haystack.tools import Tool, Toolset from haystack_integrations.components.generators.togetherai import TogetherAIChatGenerator weather_tool Tool( nameweather, descriptionGet weather info, parameters..., function... ) news_tool Tool( namenews, descriptionGet latest news, parameters..., function... ) math_toolset Toolset([add_tool, subtract_tool, multiply_tool]) generator TogetherAIChatGenerator( tools[math_toolset, weather_tool, news_tool] # Mix of Toolset and Tool objects )从OpenAIChatGenerator的底层实现看tools会被flatten_tools_or_toolsets展平、经_check_duplicate_tool_names校验工具名唯一性再转换为{type: function, function: ...}格式随请求发出返回的工具调用被解析为ToolCall含id、tool_name、arguments并以 assistant 消息中的tool_calls内容返回。Tool 与 Toolset 的详细用法可参考 Tool 文档 与 Toolset 文档。3.8 序列化to_dictTogetherAIChatGenerator提供to_dict()方法将组件序列化为字典便于配置持久化与 Pipeline YAML 序列化to_dict() - dict[str, Any]返回值为组件的字典表示。序列化时继承自OpenAIChatGenerator见 openai.py会记录model、api_base_url、generation_kwargs、api_key、timeout、max_retries、序列化后的tools等初始化参数若generation_kwargs中的response_format是 Pydantic 模型会被转换为 OpenAI 严格模式的 JSON Schema。四、TogetherAIGenerator文本生成组件详解4.1 组件定位与弃用提示TogetherAIGenerator继承自TogetherAIChatGenerator提供使用运行在 Together AI 上的 LLM 生成文本的接口。需要特别注意的是当前使用指南已标注该组件弃用Deprecated将在未来版本移除官方建议切换到TogetherAIChatGenerator它同样接受普通字符串输入。本文按参考文档完整记录其用法但新项目应优先选用TogetherAIChatGenerator。4.2 初始化参数__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与TogetherAIChatGenerator相比差异点在于多了一个system_prompt参数文本生成时使用的系统提示词。不提供则省略此时使用模型自身的默认系统提示词。没有tools与http_client_kwargs参数。generation_kwargs除max_tokens、temperature、top_p外参考文档还补充了n每个 prompt 生成多少个补全。例如 3 个 prompt、n2则共生成 6 个补全。stop一个或多个停止序列LLM 遇到后停止生成 token。presence_penalty对已在文本中出现的 token 施加的惩罚值越大模型越不容易重复相同 token。frequency_penalty对已在文本中生成的 token 的惩罚值越大越不容易重复。logit_bias对特定 token 添加 logit 偏置键为 token值为要添加的偏置量。timeout与max_retries的兜底逻辑未显式设置时timeout从OPENAI_TIMEOUT环境变量推断否则默认 30max_retries从OPENAI_MAX_RETRIES环境变量推断否则默认 5与OpenAIChatGenerator._client_kwargs中的实现一致见 openai.py。4.3 run同步文本生成run( *, prompt: str, system_prompt: str | None None, streaming_callback: StreamingCallbackT | None None, generation_kwargs: dict[str, Any] | None None ) - dict[str, Any]参数要点prompt文本生成的输入 prompt 字符串。system_prompt可选提供上下文或指令不传时使用__init__中设置的system_prompt。streaming_callback流式回调传入时会覆盖__init__中的同名设置。generation_kwargs额外的生成参数会按 key 合并并优先于__init__中传入的同名参数。返回值字典包含两个 keyreplies生成的文本补全字符串列表meta每次生成的元数据字典列表包含模型名、结束原因finish reason与 token 用量统计。基础用法参考文档原例from haystack_integrations.components.generators.togetherai import TogetherAIGenerator generator TogetherAIGenerator(modeldeepseek-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( modelmeta-llama/Llama-3.3-70B-Instruct-Turbo, system_promptYou are a helpful assistant that provides concise answers., ) response client.run(Whats Natural Language Processing?) print(response[replies][0])4.4 run_async异步文本生成run_async( *, prompt: str, system_prompt: str | None None, streaming_callback: StreamingCallbackT | None None, generation_kwargs: dict[str, Any] | None None ) - dict[str, Any]run_async是run的异步版本参数与返回值完全一致可在async代码中使用await调用。从底层看它会通过warm_up_async()在事件循环上初始化AsyncOpenAI客户端并支持异步流式迭代见 openai.py适用于高并发 Agent 或异步 Web 服务场景。4.5 序列化to_dict 与 from_dictTogetherAIGenerator同时提供序列化与反序列化方法to_dict() - dict[str, Any] # 序列化为字典 from_dict(data: dict[str, Any]) - TogetherAIGenerator # 从字典反序列化from_dict接收组件的字典表示返回反序列化后的组件实例。这使得组件配置可以写入 YAML/JSON 文件在应用重启或分布式部署时无损恢复。4.6 在 RAG Pipeline 中使用TogetherAIGenerator的典型使用位置是PromptBuilder之后。使用指南给出了一个完整的 RAG 示例——检索法国首都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(contentRome is the capital of Italy), Document(contentParis 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_storedocstore)) pipe.add_component(prompt_builder, PromptBuilder(templatetemplate)) pipe.add_component( llm, TogetherAIGenerator(modelmeta-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, ...}]}}这条链路展示了 Haystack 的经典 RAG 范式InMemoryBM25Retriever召回相关文档 →PromptBuilder用 Jinja 模板将文档与问题组装成提示词 →TogetherAIGenerator基于上下文生成答案。五、输出元数据与结束原因解读无论使用哪个组件返回的meta中都包含三项核心信息与OpenAIChatGenerator的元数据处理一致见 openai.pymodel实际响应的模型名finish_reason结束原因常见为stop正常结束、length因达到 token 上限被截断、content_filter被内容过滤器截断usageprompt_tokens/completion_tokens/total_tokens用量统计。当finish_reason为length或content_filter时底层_check_finish_reason会输出警告日志前者提示增大max_tokens对应generation_kwargs中的max_tokens参数以获得更长补全后者提示生成被内容过滤器截断。解读这些元数据有助于监控成本与排查输出截断问题。六、参数优先级与合并规则generation_kwargs在__init__与run/run_async中均可传入。参考文档与底层实现openai.py共同确认的合并规则为run调用时的generation_kwargs与初始化时的generation_kwargs按 key 合并同一 key 同时出现时run传入的值优先仅在初始化时设置的 key 会被保留。streaming_callback遵循相同逻辑run时传入会覆盖初始化时的设置。此外n每次生成的补全数在流式场景下被限制为 1若流式且n 1会抛出ValueError。结构化输出场景下非流式走chat.completions.parse端点流式则走chat.completions.create端点并携带response_format。七、选型建议与进一步阅读新项目优先用TogetherAIChatGenerator它支持ChatMessage多轮对话、函数调用、Toolset且同样接受纯字符串输入是官方推荐的演进方向TogetherAIGenerator已标记弃用。文本补全 vs 聊天参考文档明确提示TogetherAIGenerator面向文本生成而非聊天聊天场景应使用TogetherAIChatGenerator。流式输出两者均支持通过streaming_callback启用适合打字机效果的对话 UI 或长文本渐进展示。密钥管理优先使用TOGETHER_API_KEY环境变量避免密钥硬编码与序列化泄露。延伸阅读仓库内资料Together AI 集成参考文档本文依据TogetherAIChatGenerator 使用指南TogetherAIGenerator 使用指南基类 OpenAIChatGenerator 源码ChatMessage 数据类源码 与 ChatMessage 概念文档Secret 密钥管理源码 与 Secret 概念文档生成器选型与流式支持指南【免费下载链接】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），仅供参考