>Parallel 是一个专为 LLM 和 AI 应用构建的实时网络搜索和内容提取平台。

该 Task API 在分层处理器菜单上运行研究级任务 (lite → ultra，以及匹配的 -fast 变体)。 langchain-parallel 将其作为四个 LangChain 接口公开，全部在本页面上，以便您为工作负载选择合适的形态。

我应该使用哪个接口？

• 一个临时问题，可由代理调用: ParallelTaskRunTool (BaseTool).
• 一份长期运行的多源报告: ParallelDeepResearch (Runnable).
• 对列表进行批量丰富化，具有类型化输入和输出: ParallelEnrichment (Runnable).
• 当您需要对运行环境进行完全控制时的低级批处理: ParallelTaskGroup （普通类）。

所有四个都默认为 -fast 处理器变体（在相似精度下比相应的非 fast 层级快 2-5 倍）。去掉 -fast 后缀，当延迟不如最高质量重要时。请参阅 选择处理器 查看完整菜单。

概览

集成详情

设置

集成位于 langchain-parallel package.

    pip install -U langchain-parallel
    

    uv add langchain-parallel
    

凭证

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

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

ParallelTaskRunTool

ParallelTaskRunTool 是一个可由代理调用的 BaseTool。它同步运行一个任务并返回结构化的 output，按字段的 basis 引用，以及 run_id.

from langchain_parallel import ParallelTaskRunTool

tool = ParallelTaskRunTool()
result = tool.invoke({"input": "Who founded SpaceX, in one sentence?"})

print(result["output"]["content"])
print("run_id:", result["run"]["run_id"])

result["output"] 始终是一个字典；答案文本位于 result["output"]["content"] ，按字段引用位于 result["output"]["basis"]。传递一个 task_output_schema 使 content 作为解析后的 pydantic 形态字典到达，而不是自由文本字符串：

from pydantic import BaseModel, Field

class FounderFact(BaseModel):
    founder: str = Field(description="Full name of the founder")
    year: int = Field(description="Year the company was founded")

structured = ParallelTaskRunTool(task_output_schema=FounderFact)
res = structured.invoke({"input": "Who founded SpaceX and in what year?"})

print(res["output"]["content"])
print(res["output"]["basis"])  # citations + reasoning + confidence per field

{'founder': 'Elon Musk', 'year': 2002}
[{'field': 'founder', 'citations': [...], 'reasoning': '...', 'confidence': 'high'}, ...]

parse_依据：引用 + 低置信度字段

每个关注置信度的消费者最终都会编写相同的样板代码来遍历结果以获取引用、低置信度字段和 interaction_id. parse_basis() 为您完成这些操作：

from langchain_parallel import parse_basis

parsed = parse_basis(res)
print("citations per field:", {f: len(c) for f, c in parsed["citations_by_field"].items()})
print("low-confidence fields:", parsed["low_confidence_fields"])
print("interaction_id:", parsed["interaction_id"])

多轮链式调用

结果字典在顶层显示 interaction_id 。将其作为 previous_interaction_id 在下一调用中传递，以跨轮次链接上下文：

first = tool.invoke({"input": "Who founded SpaceX?"})
followup = tool.invoke({
    "input": "And what year was it founded?",
    "previous_interaction_id": first["interaction_id"],
})

ParallelDeepResearch

ParallelDeepResearch is a Runnable。它默认为 pro-fast （ -fast 的变体（"探索性网络研究"）。要获得最全面的多源报告，请传递 processor="ultra".

from langchain_parallel import ParallelDeepResearch

research = ParallelDeepResearch()
result = research.invoke("Summarize the state of net-energy-gain fusion research as of 2026.")
print(result["output"]["content"])

对于类型化的深度研究，请传递一个 output_schema:

class CityFact(BaseModel):
    capital: str
    population_millions: float

typed = ParallelDeepResearch(output_schema=CityFact)
out = typed.invoke("Capital and population (in millions) of France?")
print(out["output"]["content"])

ParallelTaskGroup

ParallelTaskGroup 创建一个任务组，分配运行并收集结果。当您需要对批处理信封进行细粒度控制时直接使用它；否则请优先选择 ParallelEnrichment 用于类型化的批量运行。

from langchain_parallel import ParallelTaskGroup

group = ParallelTaskGroup()
results = group.run([
    "Founder of Anthropic? One sentence.",
    "Founder of OpenAI? One sentence.",
])

for r in results:
    print(r["output"]["content"])

ParallelTaskGroup 暴露 run （同步）和 arun （异步）。延迟由批处理中最慢的运行和所选处理器决定—— lite-fast 通常在几秒内解决，更高级别需要几分钟。

ParallelEnrichment

ParallelEnrichment 包装 ParallelTaskGroup 和一个 default_task_spec built from your input/output pydantic schemas. It coerces pydantic instances into dicts, fans out the batch, and returns results in input order.

from langchain_parallel import ParallelEnrichment

class CompanyInput(BaseModel):
    company: str = Field(description="Company name")

class CompanyOutput(BaseModel):
    headquarters: str
    founding_year: int

enricher = ParallelEnrichment(
    input_schema=CompanyInput,
    output_schema=CompanyOutput,
)
results = enricher.invoke([
    CompanyInput(company="Anthropic"),
    {"company": "OpenAI"},
])

for r in results:
    print(r["output"]["content"])

ParallelEnrichment 块直到每个输入都解析完成。使用默认 core-fast 处理器，对于非平凡的批处理需要几分钟；为短表单字段传递更快的处理器，或为大型批处理在后台工作器中运行。

构建_任务_规范

build_task_spec 接受 pydantic 类、原始 JSON-schema 字典或文本描述，并返回一个适合 TaskSpec 的字典 client.task_run.create or add_runs(default_task_spec=...)。当您想完全控制 ParallelTaskRunTool or ParallelTaskGroup.

from langchain_parallel import build_task_spec

spec = build_task_spec(input_schema=CompanyInput, output_schema=CompanyOutput)
print(list(spec.keys()))  # ['output_schema', 'input_schema']

BYOMCP：使用您自己的 MCP 服务器

ParallelTaskRunTool 和 ParallelDeepResearch 接受 mcp_servers=[McpServer(...)] 以向运行暴露 Streamable-HTTP MCP 端点。

from langchain_parallel import McpServer

mcp = McpServer(
    name="my_internal_mcp",
    url="https://example.com/mcp",
    headers={"Authorization": "Bearer ..."},
)

tool_with_mcp = ParallelTaskRunTool(mcp_servers=[mcp])
result = tool_with_mcp.invoke({"input": "What's our latest internal release?"})

Webhook 签名验证

长时间运行的任务可以通过 webhook 传递结果。使用以下方式验证签名 verify_webhook （标准 Webhook 方案：HMAC-SHA256 哈希 <webhook-id>.<webhook-timestamp>.<body>，base64 编码， v1,<sig> 包含重放保护）。参见 webhook 设置 了解交付契约。

from langchain_parallel import verify_webhook

ok = verify_webhook(
    body,
    webhook_id=request.headers["webhook-id"],
    webhook_timestamp=request.headers["webhook-timestamp"],
    webhook_signature=request.headers["webhook-signature"],
    secret=os.environ["PARALLEL_WEBHOOK_SECRET"],
)
if not ok:
    raise PermissionError("invalid webhook signature")

链接

绑定 ParallelTaskRunTool 到任何工具调用聊天模型，并使用 create_agent:

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

llm = init_chat_model(model="claude-opus-4-8")
agent = create_agent(model=llm, tools=[tool])

agent.invoke({"messages": [("human", "Look up Anthropic's founding year.")]})

API 参考

要获取详细文档，请前往 ParallelTaskRunTool, ParallelDeepResearch, ParallelTaskGroup, or ParallelEnrichment API 参考，或 并行任务 API 指南.