feat(research): multi-note pipeline — outline + parallel section synthesis + index note
Replaces the single monolithic research note with topic-driven section notes plus an index note. Two new LLM calls: _generate_outline (JSON outline, 3-8 sections) and _synthesize_section (300-600 word focused note per section, parallelised via asyncio.gather). Public signature of run_research_pipeline unchanged; falls back to single-note synthesis on outline failure or if all sections fail. Also extracts _build_sources_block helper and adds full test suite. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -21,6 +21,106 @@ MAX_SYNTHESIS_SOURCES = 12 # deduplicated sources passed to synthesis LLM
|
||||
CHARS_PER_SOURCE = 2000 # content chars per source sent to synthesis
|
||||
|
||||
|
||||
def _build_sources_block(sources: list[dict]) -> str:
|
||||
"""Format fetched sources into a text block for LLM prompts."""
|
||||
parts = []
|
||||
for i, s in enumerate(sources, 1):
|
||||
content = (s.get("content") or s.get("snippet") or "")[:CHARS_PER_SOURCE]
|
||||
parts.append(
|
||||
f"[Source {i}] {s['title']}\nURL: {s['url']}\nSearch query: {s['query']}\n\n{content}"
|
||||
)
|
||||
return "\n\n" + ("─" * 60) + "\n\n".join(parts)
|
||||
|
||||
|
||||
async def _generate_outline(topic: str, sources: list[dict], model: str) -> list[dict]:
|
||||
"""Generate a topic outline from fetched research sources.
|
||||
|
||||
Returns a list of {"title": str, "focus": str} dicts (3–8 entries).
|
||||
Returns [] on failure — callers must fall back to single-note synthesis.
|
||||
"""
|
||||
import json as _json
|
||||
|
||||
sources_block = _build_sources_block(sources) if sources else "(no sources)"
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": (
|
||||
"You are a research organizer. Given research sources on a topic, produce a JSON array "
|
||||
"of section objects that together cover the topic comprehensively from distinct angles.\n\n"
|
||||
"Rules:\n"
|
||||
"- Return exactly 3–7 sections\n"
|
||||
"- Each section must cover a unique angle — no overlap between sections\n"
|
||||
"- Titles must work as standalone note titles (specific, not generic like 'Overview')\n"
|
||||
"- focus: one sentence describing exactly what this section covers\n"
|
||||
"- Respond with ONLY a JSON array, no other text\n\n"
|
||||
'Example: [{"title": "CRISPR: Molecular Mechanisms", "focus": "How Cas9 identifies and cuts DNA at guide-RNA-specified sites"}]'
|
||||
),
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": f"Topic: {topic}\n\nSources:\n{sources_block}",
|
||||
},
|
||||
]
|
||||
try:
|
||||
raw = await generate_completion(messages, model, max_tokens=400)
|
||||
raw = raw.strip()
|
||||
raw = re.sub(r"^```(?:json)?\s*", "", raw)
|
||||
raw = re.sub(r"\s*```$", "", raw)
|
||||
idx = raw.find("[")
|
||||
if idx >= 0:
|
||||
parsed, _ = _json.JSONDecoder().raw_decode(raw[idx:])
|
||||
if isinstance(parsed, list):
|
||||
sections = [
|
||||
s for s in parsed
|
||||
if isinstance(s, dict) and s.get("title") and s.get("focus")
|
||||
]
|
||||
if len(sections) >= 3:
|
||||
return sections[:8]
|
||||
except Exception:
|
||||
logger.warning("Outline generation failed for topic '%s'", topic, exc_info=True)
|
||||
return []
|
||||
|
||||
|
||||
async def _synthesize_section(
|
||||
section_title: str,
|
||||
section_focus: str,
|
||||
sources: list[dict],
|
||||
model: str,
|
||||
) -> tuple[str, str]:
|
||||
"""Synthesize one focused note section.
|
||||
|
||||
Returns (section_title, body_markdown). Does not stream.
|
||||
"""
|
||||
sources_block = _build_sources_block(sources) if sources else "(no sources provided)"
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": (
|
||||
"You are a focused research writer. Write a single well-structured note section "
|
||||
"on the specific topic provided.\n\n"
|
||||
"Requirements:\n"
|
||||
f"- Focus strictly on: {section_focus}\n"
|
||||
"- 300–600 words of substantive prose\n"
|
||||
"- Use ### for subsections only when they genuinely aid clarity\n"
|
||||
"- Do NOT include a top-level # heading — the title is set separately\n"
|
||||
"- Write in detailed prose paragraphs — not bullet points\n"
|
||||
"- End with a '## Sources' section listing relevant source URLs as markdown hyperlinks\n"
|
||||
"- Ignore source material that falls outside your assigned focus"
|
||||
),
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
f"Section title: {section_title}\n"
|
||||
f"Focus: {section_focus}\n\n"
|
||||
f"Sources:\n{sources_block}"
|
||||
),
|
||||
},
|
||||
]
|
||||
raw = await generate_completion(messages, model, max_tokens=2048, num_ctx=16384)
|
||||
return section_title, raw.strip()
|
||||
|
||||
|
||||
async def run_research_pipeline(
|
||||
topic: str,
|
||||
user_id: int,
|
||||
@@ -28,14 +128,17 @@ async def run_research_pipeline(
|
||||
buf=None,
|
||||
project_id: int | None = None,
|
||||
) -> Note:
|
||||
"""Full research pipeline: search → fetch → synthesize → create note.
|
||||
"""Full research pipeline: search → fetch → outline → section notes → index note.
|
||||
|
||||
Emits status events via buf.append_event throughout (when buf is provided).
|
||||
Returns the created Note.
|
||||
Emits status events via buf throughout (when buf is provided).
|
||||
Returns the index note (or a single fallback note on outline failure).
|
||||
"""
|
||||
def _status(msg: str) -> None:
|
||||
if buf is not None:
|
||||
buf.append_event("status", {"status": msg})
|
||||
|
||||
# Step 1: Generate sub-queries
|
||||
if buf is not None:
|
||||
buf.append_event("status", {"status": "Generating search queries..."})
|
||||
_status("Generating search queries...")
|
||||
queries = await _generate_sub_queries(topic, model)
|
||||
logger.info("Research: generated %d sub-queries for topic '%s'", len(queries), topic)
|
||||
|
||||
@@ -43,8 +146,7 @@ async def run_research_pipeline(
|
||||
async def _search_with_stagger(i: int, query: str) -> tuple[str, list[dict]]:
|
||||
if i > 0:
|
||||
await asyncio.sleep(0.2 * i)
|
||||
if buf is not None:
|
||||
buf.append_event("status", {"status": f"Searching: {query}..."})
|
||||
_status(f"Searching: {query}...")
|
||||
results = await _search_searxng(query)
|
||||
logger.info("Research: query '%s' → %d results", query, len(results))
|
||||
return query, results
|
||||
@@ -66,8 +168,7 @@ async def run_research_pipeline(
|
||||
# Fetch all unique URLs in parallel
|
||||
async def _fetch_source(url: str, result: dict, query: str) -> dict:
|
||||
title = result.get("title", url)
|
||||
if buf is not None:
|
||||
buf.append_event("status", {"status": f"Reading: {title[:60]}..."})
|
||||
_status(f"Reading: {title[:60]}...")
|
||||
content = await fetch_url_content(url)
|
||||
return {
|
||||
"url": url,
|
||||
@@ -84,39 +185,97 @@ async def run_research_pipeline(
|
||||
if not all_sources:
|
||||
raise ValueError(f"No results found for '{topic}'")
|
||||
|
||||
# Step 3: Filter failed fetches
|
||||
good_sources = [
|
||||
s for s in all_sources
|
||||
if not s["content"].startswith("[Failed to fetch")
|
||||
]
|
||||
good_sources = [s for s in all_sources if not s["content"].startswith("[Failed to fetch")]
|
||||
|
||||
if not good_sources:
|
||||
raise ValueError(f"Could not read any sources for '{topic}'")
|
||||
|
||||
# Limit to top N sources for synthesis (already deduplicated by URL)
|
||||
synthesis_sources = good_sources[:MAX_SYNTHESIS_SOURCES]
|
||||
logger.info(
|
||||
"Research: %d/%d sources successfully fetched, using %d for synthesis",
|
||||
len(good_sources), len(all_sources), len(synthesis_sources),
|
||||
)
|
||||
|
||||
# Step 4: Synthesize (streams tokens into chat as the note is being written)
|
||||
if buf is not None:
|
||||
buf.append_event("status", {"status": f"Synthesizing report from {len(synthesis_sources)} sources..."})
|
||||
title, body = await _synthesize_note(topic, synthesis_sources, model, buf)
|
||||
# Step 3: Generate topic outline
|
||||
_status("Generating outline...")
|
||||
outline = await _generate_outline(topic, synthesis_sources, model)
|
||||
|
||||
# Step 5: Create note
|
||||
if buf is not None:
|
||||
buf.append_event("status", {"status": "Saving note..."})
|
||||
note = await create_note(
|
||||
# Fallback: outline failed or too short → single monolithic note
|
||||
if not outline:
|
||||
logger.warning("Research outline empty, falling back to single note for '%s'", topic)
|
||||
_status("Synthesizing report...")
|
||||
title, body = await _synthesize_note(topic, synthesis_sources, model, buf=None)
|
||||
note = await create_note(
|
||||
user_id=user_id, title=title, body=body, tags=["research"], project_id=project_id,
|
||||
)
|
||||
logger.info("Research (fallback): created note id=%d title='%s'", note.id, note.title)
|
||||
return note
|
||||
|
||||
# Step 4: Synthesize each section in parallel
|
||||
for section in outline:
|
||||
_status(f"Writing: {section['title']}...")
|
||||
|
||||
raw_results = await asyncio.gather(
|
||||
*[_synthesize_section(s["title"], s["focus"], synthesis_sources, model) for s in outline],
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
# Step 5: Create section notes sequentially
|
||||
_status(f"Saving {len(outline)} notes...")
|
||||
section_note_pairs: list[tuple[dict, Note]] = []
|
||||
for section, result in zip(outline, raw_results):
|
||||
if isinstance(result, Exception):
|
||||
logger.warning("Section synthesis failed for '%s': %s", section["title"], result)
|
||||
continue
|
||||
sec_title, sec_body = result
|
||||
try:
|
||||
note = await create_note(
|
||||
user_id=user_id,
|
||||
title=sec_title,
|
||||
body=sec_body,
|
||||
tags=["research"],
|
||||
project_id=project_id,
|
||||
)
|
||||
section_note_pairs.append((section, note))
|
||||
except Exception:
|
||||
logger.warning("Failed to save section note '%s'", sec_title, exc_info=True)
|
||||
|
||||
# All sections failed — fall back to single note
|
||||
if not section_note_pairs:
|
||||
logger.warning("All section syntheses failed, falling back to single note for '%s'", topic)
|
||||
_status("Synthesizing report (fallback)...")
|
||||
title, body = await _synthesize_note(topic, synthesis_sources, model, buf=None)
|
||||
note = await create_note(
|
||||
user_id=user_id, title=title, body=body, tags=["research"], project_id=project_id,
|
||||
)
|
||||
return note
|
||||
|
||||
# Step 6: Create index note
|
||||
from datetime import date as _date
|
||||
index_lines = [
|
||||
f"Research overview for **{topic}** — {_date.today().isoformat()}",
|
||||
"",
|
||||
f"Generated from {len(synthesis_sources)} web sources across {len(section_note_pairs)} sections.",
|
||||
"",
|
||||
"## Sections",
|
||||
"",
|
||||
]
|
||||
for section, note in section_note_pairs:
|
||||
index_lines.append(f"- **{note.title}** — {section['focus']}")
|
||||
index_lines += ["", "*Search for any section title to read it.*"]
|
||||
|
||||
index_note = await create_note(
|
||||
user_id=user_id,
|
||||
title=title,
|
||||
body=body,
|
||||
tags=["research"],
|
||||
title=f"Research: {topic}",
|
||||
body="\n".join(index_lines),
|
||||
tags=["research", "research-index"],
|
||||
project_id=project_id,
|
||||
)
|
||||
logger.info("Research: created note id=%d title='%s'", note.id, note.title)
|
||||
return note
|
||||
logger.info(
|
||||
"Research: created %d section notes + index id=%d for topic '%s'",
|
||||
len(section_note_pairs), index_note.id, topic,
|
||||
)
|
||||
return index_note
|
||||
|
||||
|
||||
async def _generate_sub_queries(topic: str, model: str) -> list[str]:
|
||||
@@ -248,13 +407,7 @@ async def _synthesize_note(
|
||||
When buf is provided, tokens are streamed into the chat buffer in real time
|
||||
so the user can see the note being written. Uses an extended context window.
|
||||
"""
|
||||
sources_text_parts = []
|
||||
for i, s in enumerate(sources, 1):
|
||||
content = (s.get("content") or s.get("snippet") or "")[:CHARS_PER_SOURCE]
|
||||
sources_text_parts.append(
|
||||
f"[Source {i}] {s['title']}\nURL: {s['url']}\nSearch query: {s['query']}\n\n{content}"
|
||||
)
|
||||
sources_block = "\n\n" + ("─" * 60) + "\n\n".join(sources_text_parts)
|
||||
sources_block = _build_sources_block(sources)
|
||||
|
||||
messages = [
|
||||
{
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
"""Tests for the multi-note research pipeline."""
|
||||
import json
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_outline_parses_valid_json():
|
||||
"""_generate_outline returns parsed sections when model returns valid JSON."""
|
||||
from fabledassistant.services.research import _generate_outline
|
||||
|
||||
outline_json = json.dumps([
|
||||
{"title": "Section One", "focus": "Focus one"},
|
||||
{"title": "Section Two", "focus": "Focus two"},
|
||||
{"title": "Section Three", "focus": "Focus three"},
|
||||
])
|
||||
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value=outline_json):
|
||||
result = await _generate_outline("test topic", [], "test-model")
|
||||
|
||||
assert len(result) == 3
|
||||
assert result[0]["title"] == "Section One"
|
||||
assert result[0]["focus"] == "Focus one"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_outline_returns_empty_on_parse_failure():
|
||||
"""_generate_outline returns [] when model output is not parseable JSON."""
|
||||
from fabledassistant.services.research import _generate_outline
|
||||
|
||||
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value="not json at all"):
|
||||
result = await _generate_outline("test topic", [], "test-model")
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_outline_returns_empty_when_fewer_than_3_sections():
|
||||
"""_generate_outline returns [] when model returns fewer than 3 valid sections."""
|
||||
from fabledassistant.services.research import _generate_outline
|
||||
|
||||
outline_json = json.dumps([{"title": "Only One", "focus": "Focus"}])
|
||||
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value=outline_json):
|
||||
result = await _generate_outline("test topic", [], "test-model")
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_outline_truncates_to_8():
|
||||
"""_generate_outline truncates results to at most 8 sections."""
|
||||
from fabledassistant.services.research import _generate_outline
|
||||
|
||||
sections = [{"title": f"Section {i}", "focus": f"Focus {i}"} for i in range(10)]
|
||||
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value=json.dumps(sections)):
|
||||
result = await _generate_outline("test topic", [], "test-model")
|
||||
|
||||
assert len(result) == 8
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_outline_skips_malformed_entries():
|
||||
"""_generate_outline filters out entries missing title or focus."""
|
||||
from fabledassistant.services.research import _generate_outline
|
||||
|
||||
outline_json = json.dumps([
|
||||
{"title": "Good One", "focus": "Good focus"},
|
||||
{"title": "Missing focus"},
|
||||
{"focus": "Missing title"},
|
||||
{"title": "Good Two", "focus": "Good focus two"},
|
||||
{"title": "Good Three", "focus": "Good focus three"},
|
||||
])
|
||||
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value=outline_json):
|
||||
result = await _generate_outline("test topic", [], "test-model")
|
||||
|
||||
assert len(result) == 3
|
||||
assert all(s["title"] and s["focus"] for s in result)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_synthesize_section_returns_title_and_body():
|
||||
"""_synthesize_section returns (section_title, body) from model output."""
|
||||
from fabledassistant.services.research import _synthesize_section
|
||||
|
||||
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value="The body of this section."):
|
||||
title, body = await _synthesize_section(
|
||||
section_title="Quantum Entanglement: Mechanisms",
|
||||
section_focus="How entanglement works at the particle level",
|
||||
sources=[],
|
||||
model="test-model",
|
||||
)
|
||||
|
||||
assert title == "Quantum Entanglement: Mechanisms"
|
||||
assert body == "The body of this section."
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_synthesize_section_strips_whitespace():
|
||||
"""_synthesize_section strips leading/trailing whitespace from body."""
|
||||
from fabledassistant.services.research import _synthesize_section
|
||||
|
||||
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value="\n\n body text \n\n"):
|
||||
title, body = await _synthesize_section("Title", "Focus", [], "model")
|
||||
|
||||
assert body == "body text"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pipeline_creates_section_notes_and_index():
|
||||
"""run_research_pipeline creates N section notes + 1 index note, returns index."""
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
outline = [
|
||||
{"title": "Section A", "focus": "Focus A"},
|
||||
{"title": "Section B", "focus": "Focus B"},
|
||||
{"title": "Section C", "focus": "Focus C"},
|
||||
]
|
||||
|
||||
note_id_counter = iter(range(10, 20))
|
||||
|
||||
def _make_note(user_id, title, body, tags, project_id=None):
|
||||
n = MagicMock()
|
||||
n.id = next(note_id_counter)
|
||||
n.title = title
|
||||
return n
|
||||
|
||||
with patch("fabledassistant.services.research._generate_sub_queries", new_callable=AsyncMock, return_value=["q1"]), \
|
||||
patch("fabledassistant.services.research._search_searxng", new_callable=AsyncMock, return_value=[{"url": "http://x.com", "title": "X", "snippet": "s"}]), \
|
||||
patch("fabledassistant.services.research.fetch_url_content", new_callable=AsyncMock, return_value="content"), \
|
||||
patch("fabledassistant.services.research._generate_outline", new_callable=AsyncMock, return_value=outline), \
|
||||
patch("fabledassistant.services.research._synthesize_section", new_callable=AsyncMock, side_effect=lambda t, f, s, m: (t, f"Body for {t}")), \
|
||||
patch("fabledassistant.services.research.create_note", new_callable=AsyncMock, side_effect=_make_note):
|
||||
|
||||
from fabledassistant.services.research import run_research_pipeline
|
||||
result = await run_research_pipeline("test topic", user_id=1, model="test-model")
|
||||
|
||||
assert result.title == "Research: test topic"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pipeline_falls_back_to_single_note_when_outline_empty():
|
||||
"""run_research_pipeline falls back to single-note synthesis when _generate_outline returns []."""
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
single_note = MagicMock()
|
||||
single_note.id = 99
|
||||
single_note.title = "Research: test topic"
|
||||
|
||||
with patch("fabledassistant.services.research._generate_sub_queries", new_callable=AsyncMock, return_value=["q1"]), \
|
||||
patch("fabledassistant.services.research._search_searxng", new_callable=AsyncMock, return_value=[{"url": "http://x.com", "title": "X", "snippet": "s"}]), \
|
||||
patch("fabledassistant.services.research.fetch_url_content", new_callable=AsyncMock, return_value="content"), \
|
||||
patch("fabledassistant.services.research._generate_outline", new_callable=AsyncMock, return_value=[]), \
|
||||
patch("fabledassistant.services.research._synthesize_note", new_callable=AsyncMock, return_value=("Research: test topic", "body")), \
|
||||
patch("fabledassistant.services.research.create_note", new_callable=AsyncMock, return_value=single_note):
|
||||
|
||||
from fabledassistant.services.research import run_research_pipeline
|
||||
result = await run_research_pipeline("test topic", user_id=1, model="test-model")
|
||||
|
||||
assert result.id == 99
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pipeline_falls_back_when_all_sections_fail():
|
||||
"""run_research_pipeline falls back to single note when all section syntheses raise."""
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
outline = [
|
||||
{"title": "Section A", "focus": "Focus A"},
|
||||
{"title": "Section B", "focus": "Focus B"},
|
||||
{"title": "Section C", "focus": "Focus C"},
|
||||
]
|
||||
fallback_note = MagicMock()
|
||||
fallback_note.id = 77
|
||||
fallback_note.title = "Research: test topic"
|
||||
|
||||
with patch("fabledassistant.services.research._generate_sub_queries", new_callable=AsyncMock, return_value=["q1"]), \
|
||||
patch("fabledassistant.services.research._search_searxng", new_callable=AsyncMock, return_value=[{"url": "http://x.com", "title": "X", "snippet": "s"}]), \
|
||||
patch("fabledassistant.services.research.fetch_url_content", new_callable=AsyncMock, return_value="content"), \
|
||||
patch("fabledassistant.services.research._generate_outline", new_callable=AsyncMock, return_value=outline), \
|
||||
patch("fabledassistant.services.research._synthesize_section", new_callable=AsyncMock, side_effect=RuntimeError("synthesis failed")), \
|
||||
patch("fabledassistant.services.research._synthesize_note", new_callable=AsyncMock, return_value=("Research: test topic", "body")), \
|
||||
patch("fabledassistant.services.research.create_note", new_callable=AsyncMock, return_value=fallback_note):
|
||||
|
||||
from fabledassistant.services.research import run_research_pipeline
|
||||
result = await run_research_pipeline("test topic", user_id=1, model="test-model")
|
||||
|
||||
assert result.id == 77
|
||||
Reference in New Issue
Block a user