企业官网建设流程全解析

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 是面向生产级 LLM 应用的开源 AI 编排框架，通过模块化 Pipeline 让开发者精确控制检索、路由、记忆与生成。togetherai-haystack集成把 Together AI 托管的开源模型（如 Llama、DeepSeek）无缝接入 Haystack 的组件体系，本文基于 Together AI 集成 API 参考 系统讲解两个核心生成组件的参数、用法、流式输出、工具调用与序列化机制，并结合仓库源码剖析其底层实现。读完本文，你将掌握如何在独立脚本与 Pipeline 中调用 Together AI 模型、如何配置generation_kwargs、如何实现流式输出与函数调用，以及两个组件各自的适用场景。

集成概览：两个组件，两种生成范式

togetherai-haystack集成位于haystack_integrations.components.generators.togetherai命名空间下，提供两个生成组件：

两者共享同一套 Together AI chat completion 端点（OpenAI 兼容协议），默认模型均为meta-llama/Llama-3.3-70B-Instruct-Turbo。在官方文档中，TogetherAIGenerator已被标记为弃用（deprecated），官方建议迁移到TogetherAIChatGenerator，因为后者同样接受纯字符串输入（OpenAIChatGenerator.run会把字符串自动转换为 user 角色的ChatMessage，见 openai.py）。

环境准备与安装

安装集成包

pip install togetherai-haystack

安装后即可从haystack_integrations.components.generators.togetherai导入两个组件。

获取并配置 API Key

使用该集成需要一份有效的 Together AI 订阅与 API Key，两种提供方式：

• 推荐：设置TOGETHER_API_KEY环境变量。这也是api_key参数的默认值来源：api_key: Secret = Secret.from_env_var("TOGETHER_API_KEY")。
• 显式传入：通过api_key初始化参数配合 Haystack 的 Secret API，例如Secret.from_token("your-api-key-here")。

默认端点与可替换性

组件默认使用 Together AI 的 OpenAI 兼容端点https://api.together.xyz/v1，可通过api_base_url覆盖。这正是该集成能直接继承OpenAIChatGenerator全部能力（流式、工具、序列化）的原因：Together AI 暴露了与 OpenAI 高度兼容的 REST 接口，基类只需把api_base_url指向 Together 的地址即可。

TogetherAIChatGenerator：多轮对话与 Agent 场景

初始化签名与参数详解

TogetherAIChatGenerator继承自OpenAIChatGenerator（见 openai.py），__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：Together API Key，类型为 HaystackSecret，默认从TOGETHER_API_KEY环境变量读取。
• model：要调用的 Together AI chat completion 模型名。完整模型清单以 Together AI 官方文档 为准。
• streaming_callback：流式回调函数，每收到一个新 token 即以StreamingChunk为参数被调用。
• api_base_url：Together AI API 基础 URL，默认https://api.together.xyz/v1。
• generation_kwargs：直接透传给 Together AI 端点的生成参数（详见下文）。
• tools：ToolsType，接受单个Toolset、多个Tool的序列、或 Tool 与 Toolset 混合的序列。ToolsType的类型定义为Sequence[Tool | Toolset] | Toolset（见 tool_types.py），要求每个工具名称唯一。
• timeout：API 调用超时（秒）。
• max_retries：遇到内部错误时重试 Together AI 的最大次数。未设置时默认取OPENAI_MAX_RETRIES环境变量，否则为 5。
• http_client_kwargs：用于配置自定义httpx.Client或httpx.AsyncClient的关键字参数字典。

generation_kwargs：直接透传的生成参数

所有generation_kwargs都会原样发送到 Together AI 端点，常用参数包括：

• max_tokens：输出文本的最大 token 数上限。
• temperature：采样温度。值越高模型越"冒险"；创意类任务可试 0.9，答案明确的任务用 0（argmax 采样）。
• top_p：核采样（nucleus sampling）替代温度采样。top_p=0.1表示只考虑概率质量前 10% 的 token。
• stream：是否流式返回部分进度；开启后 token 以 server-sent events 形式持续推送，并以data: [DONE]结束。
• safe_prompt：是否在所有对话前注入安全提示。
• random_seed：随机采样种子，用于复现结果。
• response_format：JSON schema 或 Pydantic 模型，强制约束模型输出结构（模型返回工具调用时除外）。注意：流式 + 结构化输出时response_format必须是 JSON schema 而非 Pydantic 模型。

从源码看，generation_kwargs在run时与初始化时传入的字典按 key 合并：run中的 key 优先覆盖，初始化时设置的 key 若在run中未出现则保留（见 openai.py）。因此既可以把稳定的参数放在__init__，把每次调用变化的参数放在run。

独立使用示例

基础调用：

from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.togetherai import TogetherAIChatGenerator messages = [ChatMessage.from_user("What's Natural Language Processing?")] client = TogetherAIChatGenerator() response = client.run(messages) print(response)

输出示例（replies中的ChatMessage带角色、模型名与用量元数据）：

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

带流式输出的调用：

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

工具调用（Function Calling）

通过tools参数传入Tool与Toolset即可让模型准备函数调用，支持三种组织方式：

• 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", parameters=..., function=... ) news_tool = Tool( name="news", description="Get latest news", parameters=..., function=... ) math_toolset = Toolset([add_tool, subtract_tool, multiply_tool]) generator = TogetherAIChatGenerator( tools=[math_toolset, weather_tool, news_tool] # Toolset 与 Tool 混合 )

基类OpenAIChatGenerator.__init__中会通过_check_duplicate_tool_names(flatten_tools_or_toolsets(self.tools))校验工具名唯一性，并在warm_up阶段调用warm_up_tools预热工具（见 openai.py）。Tool 与 Toolset 的详细用法可参考 Tool 文档 与 Toolset 文档。

在 Pipeline 中使用

TogetherAIChatGenerator最常见的 Pipeline 位置是在ChatPromptBuilder之后，通过builder.prompt → llm.messages连接：

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)

序列化

to_dict()将组件序列化为字典，便于 Pipeline 的 YAML/JSON 持久化与反序列化。序列化时会处理streaming_callback（通过serialize_callable）、generation_kwargs（若response_format是 Pydantic 模型则转换为 OpenAI 的json_schema格式）、tools（通过serialize_tools_or_toolset）等字段（见 openai.py）。

TogetherAIGenerator：单轮文本生成（已弃用）

初始化签名与参数详解

TogetherAIGenerator继承自TogetherAIChatGenerator，提供面向纯字符串 prompt 的生成接口：

__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：生成文本时的系统提示。不提供则省略，此时使用模型的默认系统提示。
• timeout：Together AI 客户端调用超时。未设置时取OPENAI_TIMEOUT环境变量，默认 30 秒。
• max_retries：内部错误时的最大重试次数，未设置时取OPENAI_MAX_RETRIES环境变量，默认 5。

generation_kwargs支持参数在聊天版基础上还包含：

• n：每个 prompt 生成的补全数。例如 3 个 prompt 且n=2时共生成 6 条补全。
• stop：一个或多个停止序列，LLM 遇到后停止生成。
• presence_penalty：对已出现 token 的惩罚，值越大模型越不容易重复同一 token。
• frequency_penalty：对已生成过 token 的惩罚，值越大越不容易重复。
• logit_bias：对特定 token 的 logit 偏置，字典 key 为 token，value 为偏置值。

独立使用示例

基础调用：

from haystack_integrations.components.generators.togetherai import TogetherAIGenerator client = TogetherAIGenerator(model="meta-llama/Llama-3.3-70B-Instruct-Turbo") response = client.run("What's Natural Language Processing? Be brief.") print(response) # >> {'replies': ['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.'], # >> '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}}]}

带系统提示：

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)

指定生成参数（如 DeepSeek-R1 模型 + 高温度）：

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

run 与 run_async 方法

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：运行时传入的流式回调，若提供则覆盖__init__中的设置。
• generation_kwargs：运行时的额外生成参数，可能覆盖__init__中传入的参数（如 temperature、max_new_tokens、top_p 等）。

返回字典包含两个键：

• replies：生成的文本补全字符串列表。
• meta：每个生成的元数据字典列表，含模型名、finish reason、token 用量统计。

run_async提供异步生成能力，适合在异步 Pipeline（Pipeline.run_async）或async应用中调用。

在 Pipeline 中使用

TogetherAIGenerator最常见的 Pipeline 位置是在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(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', ...}]}}

序列化与反序列化

TogetherAIGenerator额外提供：

• to_dict()：序列化为字典（继承自聊天生成器）。
• from_dict(data: dict[str, Any]) -> TogetherAIGenerator：从字典反序列化恢复组件实例，返回TogetherAIGenerator。

这两个方法让组件可以安全地写入 YAML Pipeline 配置并重新加载，是 Haystack 声明式 Pipeline 的基石。

迁移建议与最佳实践

由于TogetherAIGenerator已标记为弃用，新项目应优先选择TogetherAIChatGenerator，原因如下：

1. 功能超集：TogetherAIChatGenerator继承自OpenAIChatGenerator，同时支持ChatMessage与纯字符串输入，功能完全覆盖旧组件；
2. 工具调用：只有聊天生成器支持tools参数，Agent 工作流与函数调用场景必须使用它；
3. 流式一致：两者均支持streaming_callback流式输出；
4. 面向未来：多轮对话、Agent 化是当前 LLM 应用的主流形态，TogetherAIChatGenerator与之天然契合。

实际选型建议：

• 构建 RAG 问答、多轮对话、Agent 系统 → 使用TogetherAIChatGenerator；
• 需要结构化输出（response_format+ JSON schema）→ 使用TogetherAIChatGenerator；
• 单轮、简单的 prompt→answer 脚本 → 可直接迁移到TogetherAIChatGenerator，把字符串传给run即可；
• 异步高并发服务 → 使用run_async或在异步 Pipeline 中运行。

进一步阅读

• TogetherAIChatGenerator 组件文档：组件定位、Tool 支持与流式的完整说明
• TogetherAIGenerator 组件文档：含弃用说明与 RAG 示例
• 生成器选型指南：不同生成器的流式支持对比
• ChatMessage 数据类文档：输入输出的消息结构
• Haystack 组件总览：更多生成器与组合方式

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