这篇能做出什么
照着本文做完，你会得到一条"并行研究流水线"：把一个笼统的大问题丢进去，它自动拆成若干个彼此独立的子问题，几路同时开工，跑完自动汇总成一份能直接改改就交出去的报告草稿。
举个具体例子。任务是"调研宠物智能硬件出海东南亚的机会"。一个人串行做，通常是：先查各国市场规模，再查竞品和定价，再查电商平台入驻规则，再查认证合规，最后查物流和售后成本。每换一个环节，都要重新翻资料、重新在脑子里组织语言，串行跑完半天就过去了。
改造之后，这五件事同时开工，各自带着独立的上下文去查。整体耗时接近"最慢那一趟"，而不是五趟相加。汇总阶段再做一次交叉校验，把五份结果里互相矛盾的地方挑出来单独标红。
需要说清楚一点：并行本身不会自动省钱。省钱的地方在于——每个子任务只带自己需要的那点背景，不用把一个超长上下文反复滚来滚去；同时能快速发现"这条路查不出东西"，及时掐掉。所以"时间和成本减半"是个目标，实际幅度取决于任务怎么拆、上下文怎么控。下面给的是通用做法。
前置条件清单
- 一个能调用 GPT-6 Astra 的 API Key，以及对应的 base_url。具体申请入口、可用区域、计费方式，以官方页面为准。
- Python 3.9 以上，具体版本要求以官方 SDK 文档为准。
- 装两个库：
pip install openai python-dotenv。 - 项目目录里放一个
.env文件，写三行：ASTRA_API_KEY=、ASTRA_BASE_URL=、ASTRA_MODEL=。 - 一颗"结果要人工核验"的心。这一点后面会反复提到。
关键的模型标识、接口路径、计费口径，都从官方文档里抄，别从二手文章里抄。
第 1 步：准备调用入口
先做一个薄封装，后面所有步骤都复用它。
```python
client.py
import os
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
client = OpenAI(
api_key=os.environ["ASTRA_API_KEY"],
base_url=os.environ["ASTRA_BASE_URL"], # 以官方文档为准
)
MODEL = os.environ["ASTRA_MODEL"] # 模型标识以官方文档为准
def ask(system: str, user: str, max_tokens: int = 2000) -> str:
resp = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
temperature=0.3,
max_tokens=max_tokens,
)
return resp.choices[0].message.content.strip()
```
temperature 调低，研究类任务要的是稳定，不是花样。
第 2 步：把大问题拆成可并行的子问题
这一步决定了整条流水线的质量。拆得干净，后面顺；拆得含糊，汇总时全是矛盾。
```python
plan.py
import json
from client import ask
PLANNER_SYSTEM = """你是研究项目负责人，负责把一个大问题拆成可以同时开工的子问题。
硬性要求：
1. 每个子问题都能独立完成，不需要等其他子问题的结论；
2. 子问题之间内容尽量不重叠；
3. 每个子问题必须给出一个具体的交付物（一段数据、一张对比表、一份清单等），
不接受"了解背景""分析意义"这类没有交付物的伪任务；
4. 子问题数量控制在 4 到 6 个；
5. 只输出 JSON，不要任何解释文字。
输出格式：
{"subtasks":[{"id":"T1","title":"简短标题","question":"要回答的具体问题",
"deliverable":"交付物形式","key_points":["要点1","要点2"]}]}
"""
def make_plan(big_question: str) -> list[dict]:
raw = ask(PLANNER_SYSTEM, f"研究问题：{big_question}", max_tokens=1500)
raw = raw.removeprefix("``json").removeprefix("`").removesuffix("``").strip()
return json.loads(raw)["subtasks"]
```
拆解本身也是一次模型调用，所以它有自己的成本。但这一次调用换来的是后面几路并行的结构，值得。
第 3 步：写"单路研究"函数
每一路研究都是独立的一次调用，输入是那个子问题和它的要点，输出是一段结构化的结果。
```python
worker.py
from client import ask
RESEARCH_SYSTEM = """你是一名严谨的研究员，负责回答一个具体子问题。
写作规则：
1. 把"事实"和"推断"分开写，推断要明确标出"这是推断"；
2. 不确定的地方直接写"未确认"，不要用看起来很像真的数字填空；
3. 涉及具体数字、政策、价格时，说明该到哪里去核实（例如"以官方发布为准"）；
4. 输出结构：结论摘要（3 句以内）→ 分点展开 → 待核实清单。
"""
def research_one(task: dict) -> str:
user = (
f"子问题标题：{task['title']}\n"
f"需要回答：{task['question']}\n"
f"交付物：{task['deliverable']}\n"
f"必须覆盖的要点：{', '.join(task.get('key_points', []))}"
)
return ask(RESEARCH_SYSTEM, user, max_tokens=2500)
```
第 2 条规则是整个流程里最值钱的一行。让模型把"不知道"写出来，比让它编一个漂亮的数字有用得多。
第 4 步：并发跑起来，同时把结果落盘
```python
run.py
import json, pathlib
from concurrent.futures import ThreadPoolExecutor, as_completed
from plan import make_plan
from worker import research_one
OUT = pathlib.Path("results")
OUT.mkdir(exist_ok=True)
def run(big_question: str, max_workers: int = 4):
tasks = make_plan(big_question)
(OUT / "plan.json").write_text(
json.dumps(tasks, ensure_ascii=False, indent=2), encoding="utf-8"
)
results = {}
with ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = {}
for t in tasks:
fp = OUT / f"{t['id']}.md"
if fp.exists(): # 断点续跑
results[t["id"]] = fp.read_text(encoding="utf-8")
continue
futures[pool.submit(research_one, t)] = t
for fut in as_completed(futures):
t = futures[fut]
try:
text = fut.result()
except Exception as e:
text = f"[该子任务失败：{e}]"
(OUT / f"{t['id']}.md").write_text(text, encoding="utf-8")
results[t["id"]] = text
print("完成", t["id"])
return tasks, results
if __name__ == "__main__":
run("宠物智能硬件出海东南亚的机会与主要风险")
```
max_workers 先设 3 到 4，跑稳了再往上加。落盘那几行别省：并发跑到一半挂了，重跑一遍的时间和钱都是白花的。
第 5 步：汇总与交叉校验
汇总不是把几段文字拼起来。拼起来的结果读着像几个人各说各话，一点用没有。汇总要显式做三件事：去重、标冲突、补缺口。
```python
merge.py
from client import ask
MERGE_SYSTEM = """你是研究报告的主编，手上是几位研究员针对同一大问题的子报告。
请完成：
1. 合并：把重复的结论合并成一条；
2. 标冲突：不同子报告之间数据或结论不一致的地方，单独列成"冲突点"，
写清楚分歧在哪、建议怎么核实，不要自行裁决谁对；
3. 补缺口：指出还需要补充什么信息才能下结论；
4. 输出一份报告提纲：执行摘要 → 分主题正文（保留各子报告的要点）→ 冲突点
→ 待核实清单 → 建议的下一步动作。
不要把子报告原文照抄，要重写成一份连贯的文档。
"""
def merge(big_question: str, tasks: list[dict], results: dict) -> str:
blocks = []
for t in tasks:
blocks.append(f"### {t['id']} {t['title']}\n{results.get(t['id'], '（缺失）')}")
user = f"研究总问题：{big_question}\n\n" + "\n\n".join(blocks)
return ask(MERGE_SYSTEM, user, max_tokens=4000)
```
如果子报告加起来太长，先让每一路自己压成一份 300 字要点，再送进汇总，别硬塞。
常见坑与排错
子问题互相依赖。 表现是汇总时两路结论打架，而且打架的原因是它们各自假设了对方没给出的前提。排错办法：拆完之后人工扫一眼，凡是需要读另一份结果才能写下去的，合并成一个子任务。
拆出一堆没有交付物的伪任务。 "分析行业背景""探讨发展趋势"这类任务，模型会写得很漂亮但没法用。在拆解提示词里强制要求 deliverable 字段，就是为了卡住这一类。
报 429 限流。 并发调低到 2 到 3，给重试加指数退避：失败后等 2 秒、4 秒、8 秒。多数 SDK 自带重试参数，具体写法以官方文档为准。
数字看起来很真但是编的。 这是研究类任务的头号风险。三道防线一起上：提示词里要求分开写事实与推断；要求输出"待核实清单"；最后人工抽查所有关键数字。涉及金额、政策、市场份额这类内容，一律回到官方或权威页面确认。
汇总被偷懒。 最省事的做法是把五份结果直接拼起来交差。判断标准很简单：读一遍汇总稿，如果同一条结论出现了两三次，就是没做好去重。
上下文超长报错。 每路先压缩再汇总；必要时把汇总拆成两轮，先合并同类主题，再统一成稿。
中途失败丢结果。 有落盘和断点续跑就不会。这是第 4 步里那几行 pathlib 的意义。
下一步建议
- 接检索。目前模型只能靠训练时的记忆，接一个搜索接口或内部知识库，让每条事实带上可点开的来源，可信度会上一个台阶。
- 加一层评审。让另一个模型专门挑汇总稿的毛病：哪些结论没有支撑、哪些地方自相矛盾。写和挑错分开，质量差别很明显。
- 递归拆解。某个子问题如果本身还是太大，就再走一遍第 2 步，往下拆一层。注意控制深度，两层通常够用。
- 做成命令行工具。把
run.py包一层参数，python run.py "研究问题"直接出稿，日常用起来才顺手。 - 沉淀模板。常见的研究类型（竞品调研、供应商比价、政策梳理）各自固化一份拆解提示词，比每次临场发挥稳定得多。
最后提醒一句：这套流程省的是翻资料和起草的时间，不省判断。报告里每一个要拿去决策的数字，仍然需要你自己去核实一遍。