ChatParallel 集成

使用 LangChain Python 集成 ChatParallel 对话模型。

并行是一个为大型语言模型（LLM）和AI应用构建的实时网络搜索和内容提取平台。

ChatParallel是Parallel模型的OpenAI兼容聊天界面。speed模型是一个低延迟的对话模型，不提供引用；研究模型（lite、base、core）浏览网络，并通过JSON模式返回按字段组织的引用和结构化输出。

ChatParallel是规范类名。较早的ChatParallelWeb继续作为同一类的别名工作。

概述

集成细节

类	包	可序列化	JS/TS支持	下载	最新版本
@[ChatParallel]	@[langchain-parallel]	✅	❌

模型特性

工具调用	结构化输出	图像输入	音频输入	视频输入	Token级流式输出	原生异步	Token使用	Logprobs
❌	✅ (研究模型)	❌	❌	❌	✅	✅	❌	❌

选择模型

模型	延迟	网络浏览	引用	结构化输出	使用场景
speed	低	❌	❌	❌	从模型的参数化知识中获得对话回答。
lite	中等	✅	✅	✅	带引用的事实查找。
base	中高	✅	✅	✅	带引用的中深度研究。
core	较高	✅	✅	✅	带引用的多源研究。

speed不遵循response_format，因此with_structured_output()在那里会引发清晰的错误。当您需要解析的pydantic输出或按字段引用时，请使用研究模型。

设置

要访问Parallel模型，请安装langchain-parallel集成包并获取并行 API密钥。

安装

pip

pip install -U langchain-parallel

凭证

前往并行注册并生成API密钥。在您的环境中设置PARALLEL_API_KEY：

import getpass
import os

if not os.environ.get("PARALLEL_API_KEY"):
    os.environ["PARALLEL_API_KEY"] = getpass.getpass("Parallel API key:\n")

实例化

from langchain_parallel import ChatParallel

llm = ChatParallel(
    model="speed",
    # timeout=None,
    # max_retries=2,
    # api_key="...",  # optional if PARALLEL_API_KEY is set
    # base_url="https://api.parallel.ai",  # default
)

参见 @[ChatParallel] API参考以获取所有可用参数的完整集。

调用

messages = [
    ("system", "You are a helpful assistant with access to real-time web information."),
    ("human", "What are the latest developments in AI?"),
]
ai_msg = llm.invoke(messages)
print(ai_msg.content)

链接

将模型与提示词模板链接：

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate(
    [
        (
            "system",
            "You are a research assistant with access to real-time web information. "
            "Answer questions about {topic} using current sources.",
        ),
        ("human", "{question}"),
    ]
)

chain = prompt | llm
chain.invoke(
    {
        "topic": "artificial intelligence",
        "question": "What are the most significant AI breakthroughs in 2026?",
    }
)

结构化输出

在研究模型（lite、base、core）上，ChatParallel.with_structured_output(...)绑定一个JSON-schema response_format，并返回一个解析的pydantic对象（或字典）。在speed上调用它将引发ValueError，因为speed会静默忽略response_format。

from pydantic import BaseModel, Field

class Founder(BaseModel):
    name: str = Field(description="Full name of the founder")
    company: str = Field(description="Company they founded")

structured = ChatParallel(model="lite").with_structured_output(Founder)
parsed = structured.invoke([("human", "Who founded SpaceX?")])
print(parsed)

name='Elon Musk' company='SpaceX'

method="json_schema"（默认）、method="json_mode"和method="function_calling"都受支持。传递include_raw=True以接收完整的{"raw", "parsed", "parsing_error"}信封并捕获解析器失败：

structured = ChatParallel(model="lite").with_structured_output(Founder, include_raw=True)
res = structured.invoke([("human", "Who founded SpaceX?")])
res["parsed"]          # Founder(...) or None
res["parsing_error"]   # Exception or None
res["raw"]             # original AIMessage

引用

研究模型用AIMessage.response_metadata["basis"]填充按字段引用、模型的推理和置信度标签。response_metadata["interaction_id"]用于多轮上下文链接；system_fingerprint在存在时会被转发。

cited = ChatParallel(model="lite").invoke([
    ("human", "Who is the current CEO of OpenAI? One sentence."),
])
print(cited.content)
print("\nbasis:", cited.response_metadata.get("basis"))
print("interaction_id:", cited.response_metadata.get("interaction_id"))

流式输出

ChatParallel支持按Token流式输出：

stream = llm.stream_events(messages, version="v3")
for token in stream.text:
    print(token, end="", flush=True)

异步

ai_msg = await llm.ainvoke(messages)

stream = await llm.astream_events(messages, version="v3")
async for token in stream.text:
    print(token, end="", flush=True)

Token使用

Parallel目前不提供Token使用元数据。usage_metadata是None。

ai_msg = llm.invoke(messages)
print(ai_msg.usage_metadata)
# None

响应元数据

ai_msg = llm.invoke(messages)
print(ai_msg.response_metadata)
# {'model_name': 'speed', 'finish_reason': 'stop', 'created': 1764043410}

对于研究模型，response_metadata还携带basis（按字段引用）、interaction_id（用于多轮链接）和system_fingerprint（当可用时）。

错误处理

集成在常见失败模式下引发ValueError，并附带描述性消息：

from langchain_parallel import ChatParallel

try:
    llm = ChatParallel(api_key="invalid-key")
    response = llm.invoke([("human", "Hello")])
except ValueError as e:
    if "Authentication failed" in str(e):
        print("Invalid API key provided")
    elif "Rate limit exceeded" in str(e):
        print("API rate limit exceeded, please try again later")

OpenAI兼容性

ChatParallel接受许多OpenAI Chat Completions API参数，用于无缝迁移OpenAI客户端。高级参数如tools、tool_choice、top_p和frequency_penalty被接受，但会被Parallel API忽略。

llm = ChatParallel(
    model="speed",
    # accepted but ignored by Parallel:
    tools=[{"type": "function", "function": {"name": "example"}}],
    tool_choice="auto",
    top_p=1.0,
    frequency_penalty=0.0,
    presence_penalty=0.0,
    logit_bias={},
    seed=42,
    user="user-123",
)

对于结构化输出，优先选择ChatParallel.with_structured_output(...)（参见结构化输出），而不是直接传递response_format。它在研究模型上工作，并返回解析的对象。

消息处理

集成将连续的同类型消息合并以满足API要求：

from langchain.messages import HumanMessage, SystemMessage

# Consecutive system messages are automatically merged before the API call.
messages = [
    SystemMessage("You are a helpful assistant."),
    SystemMessage("Always be polite and concise."),
    HumanMessage("What is the weather like today?"),
]

response = llm.invoke(messages)

API参考

有关详细文档，请前往 @[ChatParallel] API参考或Parallel聊天API快速入门。

---

官方原文：src/oss/python/integrations/chat/parallel.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。