多步骤研究循环设计
规划 → 搜索 → 阅读 → 提取 → 综合 → 重复，直到达到足够深度
多步骤研究循环设计 是 CoddyKit 上的免费 AI Agents 课时。 这是第 1 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
什么是研究循环
研究循环是一种代理模式，其中 LLM 会规划、搜索、阅读、提取信息、识别空白并重复执行，直到研究目标得到满足。
与一次性搜索不同，该循环会根据发现的内容进行调整——跟进意外线索，并舍弃无效路径。
第 1 阶段：分解问题
第一步是将研究问题拆分为多个子问题。这样可以创建有方向的搜索计划，避免代理漫无目的地浏览。
import openai, json
client = openai.OpenAI(api_key='YOUR_OPENAI_KEY')
def decompose_question(question: str) -> list[str]:
prompt = (
f'Break this research question into 3-5 focused sub-questions.\n'
f'Each sub-question should be independently searchable.\n'
f'Question: "{question}"\n'
f'Return JSON: {{"sub_questions": ["..."]}}'
)
resp = client.chat.completions.create(
model='gpt-4o',
messages=[{'role': 'user', 'content': prompt}],
response_format={'type': 'json_object'}
)
return json.loads(resp.choices[0].message.content)['sub_questions']
questions = decompose_question('What are the main causes of inflation in 2024?')
print(questions)第 2 阶段：网络搜索
针对每个子问题执行网络搜索。请使用搜索 API（Serper、Brave、Bing）获取网址和摘要列表。不要阅读所有页面——请优先处理高质量来源。
import requests
SERPER_KEY = 'YOUR_SERPER_API_KEY'
def search_web(query: str, num_results: int = 5) -> list[dict]:
resp = requests.post(
'https://google.serper.dev/search',
headers={'X-API-KEY': SERPER_KEY, 'Content-Type': 'application/json'},
json={'q': query, 'num': num_results}
)
resp.raise_for_status()
results = resp.json().get('organic', [])
return [
{'title': r['title'], 'url': r['link'], 'snippet': r.get('snippet', '')}
for r in results
]第 3 阶段：阅读并提取事实
获取每个网址，并提取与子问题相关的关键事实。请使用 LLM 阅读文章，并生成一份包含来源网址的事实列表。
import httpx
from bs4 import BeautifulSoup
def fetch_text(url: str, max_chars: int = 4000) -> str:
try:
resp = httpx.get(url, timeout=10, follow_redirects=True,
headers={'User-Agent': 'ResearchAgent/1.0'})
soup = BeautifulSoup(resp.text, 'html.parser')
for tag in soup(['script', 'style', 'nav', 'footer']):
tag.decompose()
return soup.get_text(separator=' ', strip=True)[:max_chars]
except Exception:
return ''
def extract_facts(text: str, question: str, url: str) -> list[dict]:
prompt = (
f'Extract key facts from the text that answer: "{question}"\n'
f'Return JSON: {{"facts": ["fact1", ...]}}\n\n'
f'TEXT:\n{text[:3000]}'
)
resp = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
response_format={'type': 'json_object'}
)
facts = json.loads(resp.choices[0].message.content).get('facts', [])
return [{'fact': f, 'source': url} for f in facts]第 4 阶段：识别知识空白
阅读完成后，请让 LLM 检查已积累的事实，并识别仍然未知的内容。这些空白会成为下一轮搜索查询。
def identify_gaps(original_question: str, facts: list[dict]) -> list[str]:
fact_text = '\n'.join(f'- {f["fact"]}' for f in facts[:20])
prompt = (
f'Original question: "{original_question}"\n'
f'Facts gathered so far:\n{fact_text}\n\n'
f'What key aspects are still unanswered? '
f'Return 0-3 follow-up search queries (0 if research is complete).\n'
f'JSON: {{"gaps": ["search query 1", ...]}}'
)
resp = client.chat.completions.create(
model='gpt-4o',
messages=[{'role': 'user', 'content': prompt}],
response_format={'type': 'json_object'}
)
return json.loads(resp.choices[0].message.content).get('gaps', [])终止条件
循环需要明确的停止条件，以避免无限运行。请在以下情况下停止：
- 未识别出新的空白
- 上一次迭代没有添加新的事实
- 达到最大迭代次数（安全限制）
- 事实总数超过阈值（深度已足够）
def should_continue(gaps: list[str], new_facts_this_round: int,
iteration: int, total_facts: int) -> bool:
if iteration >= 5:
return False # Hard cap: 5 iterations
if new_facts_this_round == 0:
return False # No new information found
if not gaps:
return False # LLM says research is complete
if total_facts >= 50:
return False # Sufficient depth reached
return True
if __name__ == '__main__':
print('Continue (has gaps, new facts)?', should_continue(['gap1'], 4, iteration=1, total_facts=10))
print('Continue (no new facts)?', should_continue(['gap1'], 0, iteration=1, total_facts=10))
去除重复事实
多个来源经常会报告同一个事实。请让 LLM 合并语义等价的事实，同时保留来源最可靠的版本，以去除重复内容。
def deduplicate_facts(facts: list[dict]) -> list[dict]:
if len(facts) <= 3:
return facts
fact_text = '\n'.join(
f'{i}: {f["fact"]} (source: {f["source"]})' for i, f in enumerate(facts)
)
prompt = (
f'Remove duplicate or near-duplicate facts. Keep the most informative version.\n'
f'Return JSON: {{"keep_indices": [0, 1, ...]}}\n\n'
f'FACTS:\n{fact_text}'
)
resp = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
response_format={'type': 'json_object'}
)
indices = json.loads(resp.choices[0].message.content).get('keep_indices', [])
return [facts[i] for i in indices if i < len(facts)]完整的研究循环
将所有阶段合并到一个 research() 函数中，由该函数驱动循环直至终止。
def research(question: str) -> dict:
all_facts = []
iteration = 0
# Phase 1: Decompose into sub-questions
queries = decompose_question(question)
while True:
new_facts_this_round = 0
for query in queries:
results = search_web(query, num_results=3)
for result in results:
text = fetch_text(result['url'])
if not text:
continue
facts = extract_facts(text, query, result['url'])
all_facts.extend(facts)
new_facts_this_round += len(facts)
all_facts = deduplicate_facts(all_facts)
gaps = identify_gaps(question, all_facts)
iteration += 1
if not should_continue(gaps, new_facts_this_round, iteration, len(all_facts)):
break
queries = gaps # next iteration searches for the gaps
return {'facts': all_facts, 'iterations': iteration}追踪来源信息
每个事实都必须携带其来源网址，以便最终报告包含引用。在处理过程中绝不要删除来源元数据——引用层需要使用这些信息。
def add_fact(fact_list: list, fact_text: str, source_url: str, iteration: int):
fact_list.append({
'fact': fact_text,
'source': source_url,
'iteration': iteration,
'verified': False # set to True after cross-referencing
})
# Example:
facts_store = []
add_fact(facts_store, 'Global inflation peaked at 9.1% in June 2022',
'https://bls.gov/news.release/cpi.htm', iteration=1)
print(f'Logged {len(facts_store)} fact(s):')
for f in facts_store:
print(f" - {f['fact']} (source: {f['source']})")
并行搜索以提升速度
顺序执行搜索的速度很慢——5 个查询 × 3 个网址 × 每个网址获取一次 = 15 次顺序 HTTP 调用。请使用 concurrent.futures 在每次迭代中并行获取内容。
from concurrent.futures import ThreadPoolExecutor, as_completed
def parallel_research_round(queries: list[str]) -> list[dict]:
collected = []
def process_query(query):
results = search_web(query, num_results=3)
facts = []
for r in results:
text = fetch_text(r['url'])
if text:
facts.extend(extract_facts(text, query, r['url']))
return facts
with ThreadPoolExecutor(max_workers=4) as ex:
futures = {ex.submit(process_query, q): q for q in queries}
for future in as_completed(futures):
collected.extend(future.result())
return collected监控循环进度
记录每次迭代，以便 debug 循环为何停止或运行时间超出预期。请包含查询数量、新增事实数量和已识别的空白。
import logging
log = logging.getLogger('research_loop')
import sys
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
def log_iteration(iteration: int, queries: list[str],
new_facts: int, total_facts: int, gaps: list[str]):
log.info(
'Iteration %d | Queries: %d | New facts: %d | Total: %d | Gaps: %d',
iteration, len(queries), new_facts, total_facts, len(gaps)
)
if gaps:
for g in gaps:
log.debug(' Gap query: %s', g)
if __name__ == '__main__':
log_iteration(iteration=2, queries=['who are our top competitors?'], new_facts=4, total_facts=12, gaps=['pricing data'])
找不到新信息时，什么会终止研究循环
理解终止条件可以防止无限循环，并确保代理在合理的时间范围内交付结果。
研究循环设计回顾
多步骤研究循环遵循以下流程：分解 → 搜索 → 阅读 → 提取 → 识别空白 → 重复。当空白为空、找不到新事实，或达到严格的迭代上限时停止。
请始终为每个事实记录来源网址，为提高速度并行执行搜索，并在综合处理前去除重复事实。
常见问题解答
「多步骤研究循环设计」课时是免费的吗？
是的 — 「多步骤研究循环设计」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「多步骤研究循环设计」这节课中我会学到什么？
规划 → 搜索 → 阅读 → 提取 → 综合 → 重复，直到达到足够深度 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 1 节课，共 4 节。
「多步骤研究循环设计」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。