Fix long-input evidence extraction and trim assignment context

backend/src/alphapulse/news_research/pipeline.py

@@ -1,6 +1,7 @@
 """Evidence contracts and lossless context preparation for event research."""
 import json
-from typing import Literal
+import re
+from typing import Annotated, Literal
 from pydantic import BaseModel, Field, ConfigDict
 from alphapulse.news_research.time_contracts import PackageTimePoint
 from alphapulse.news_research.model import complete, fits_context, SubmissionFailure
@@ -68,55 +69,110 @@ def validate_event(result, articles):
         if claim.verdict != 'insufficient' and not claim.evidence_refs:
             raise ValueError('核验结论缺少原文证据')
 
-async def extract_articles(provider, system, articles):
+class EvidenceSelection(BaseModel):
-    pending = [(a['id'], a['title']+'\n'+(a.get('summary') or '')+'\n'+(a.get('content') or '')) for a in articles]
+    model_config = ConfigDict(extra='forbid')
+    facts: str = Field(min_length=1)
+    evidence_ids: list[Annotated[int, Field(ge=0, strict=True)]] = Field(min_length=1)
+
+
+SELECTION_RULES = '\n本次只输出中文facts和evidence_ids，选择本次输入提供的证据编号；不生成quote或article_id。程序将按编号回填逐字原文，不得选择无关证据。'
+
+
+def selected_extraction(selection, evidence):
+    ids = selection.evidence_ids
+    if len(ids) != len(set(ids)) or not set(ids) <= evidence.keys():
+        raise ValueError('提取结果含未知或重复证据编号')
+    return Extraction(facts=selection.facts, evidence_refs=[evidence[i] for i in ids])
+
+
+def source_passages(article):
+    # Separate source fields so no quote can span a synthesized title/body boundary.
+    # Sentence boundaries keep evidence readable; long sentences still retain every character.
+    passages = []
+    for field in ('title', 'summary', 'content'):
+        for sentence in re.split(r'(?<=[。！？])|(?<=[.!?])(?=[ \t])', article.get(field) or ''):
+            passages.extend({'field': field, 'text': sentence[start:start+500]}
+                            for start in range(0, len(sentence), 500))
+    return passages
+
+
+async def extract_articles(provider, system, articles, *, scopes=None):
+    pending = [(article, source_passages(article)) for article in articles]
     extracts, usages = [], []
+    rules = system + SELECTION_RULES
     try:
-        schema = Extraction
         while pending:
-            identifier, content = pending.pop(0)
+            article, passages = pending.pop(0)
-            part = {'task': 'extract', 'article_id': identifier, 'content': content}
+            identifier = article['id']
-            if not fits_context(provider, system, part, schema):
+            if not passages:
-                if len(content) < 2:
+                raise ValueError('文章缺少可读取原文')
-                    raise ValueError('模型上下文不足以读取分段')
+            part = {'task': 'extract', 'article_id': identifier,
-                middle = len(content)//2
+                    'packages': [p for p in (scopes or []) if p['news_id'] == identifier],
-                pending[0:0] = [(identifier, content[:middle]), (identifier, content[middle:])]
+                    'passages': [{'id': i, **passage} for i, passage in enumerate(passages)]}
+            if not fits_context(provider, rules, part, EvidenceSelection):
+                if len(passages) > 1:
+                    middle = len(passages)//2
+                    halves = passages[:middle], passages[middle:]
+                elif passages and len(passages[0]['text']) > 1:
+                    passage = passages[0]
+                    middle = len(passage['text'])//2
+                    halves = ([{**passage, 'text': passage['text'][:middle]}],
+                              [{**passage, 'text': passage['text'][middle:]}])
+                else:
+                    raise ValueError(f'文章{identifier}的归属边界及最小原文片段超出模型上下文')
+                pending[0:0] = [(article, half) for half in halves]
                 continue
-            extract, used = await complete(provider, system, part, schema)
+            selection, used = await complete(provider, rules, part, EvidenceSelection)
             usages.append(used)
-            validate_evidence(extract.evidence_refs, {a['id']: a for a in articles})
+            evidence = {i: Evidence(article_id=identifier, quote=p['text']) for i, p in enumerate(passages)}
-            if any(r.article_id != identifier or r.quote not in content for r in extract.evidence_refs):
+            extract = selected_extraction(selection, evidence)
-                raise ValueError('分段证据越界')
+            validate_evidence(extract.evidence_refs, {identifier: article})
-            if extract.evidence_refs:
             extracts.append(extract.model_dump())
         return extracts, usages
     except Exception as error:
-        raise SubmissionFailure(str(error), {'parts': usages, 'failed': getattr(error, 'usage', None)}) from error
+        raise SubmissionFailure(f'文章{identifier}长输入原文提取失败：{error}', {'parts': usages, 'failed': getattr(error, 'usage', None)}) from error
+
+
+def consolidation_input(extracts):
+    evidence, identities, parts = {}, {}, []
+    for extract in extracts:
+        ids = []
+        for ref in extract['evidence_refs']:
+            key = (ref['article_id'], ref['quote'])
+            if key not in identities:
+                identifier = len(evidence)
+                identities[key] = identifier
+                evidence[identifier] = Evidence.model_validate(ref)
+            ids.append(identities[key])
+        parts.append({'facts': extract['facts'], 'evidence_ids': ids})
+    return {'task': 'consolidate', 'extractions': parts,
+            'evidence': [{'id': i, **ref.model_dump()} for i, ref in evidence.items()]}, evidence
+
 
 async def compact_extractions(provider, system, payload, schema, fits=fits_context):
     usages = []
     try:
-        rules = system + '\nconsolidate任务只合并给定分段事实，保留冲突和时间线；每个article_id至少保留一条给定原文引用，不新增引用，压缩重复表述。'
+        rules = system + '\nconsolidate任务只合并给定分段事实，保留冲突和时间线；每个article_id至少保留一条给定原文证据，压缩重复表述。' + SELECTION_RULES
         while not fits(provider, system, payload, schema):
             extracts = payload['extractions']
             groups, group = [], []
             for extract in extracts:
-                if not fits_context(provider, rules, {'task': 'consolidate', 'extractions': [*group, extract]}, Extraction):
+                if not fits_context(provider, rules, consolidation_input([*group, extract])[0], EvidenceSelection):
                     if not group:
                         raise ValueError('模型上下文不足以归纳分段结果，请增加上下文配置')
                     groups.append(group); group = []
-                    if not fits_context(provider, rules, {'task': 'consolidate', 'extractions': [extract]}, Extraction):
+                    if not fits_context(provider, rules, consolidation_input([extract])[0], EvidenceSelection):
                         raise ValueError('模型上下文不足以归纳分段结果，请增加上下文配置')
                 group.append(extract)
             if group: groups.append(group)
             compacted = []
             for group in groups:
-                result, used = await complete(provider, rules, {'task': 'consolidate', 'extractions': group}, Extraction)
+                request, evidence = consolidation_input(group)
+                selection, used = await complete(provider, rules, request, EvidenceSelection)
                 usages.append(used)
-                refs = {(r['article_id'], r['quote']) for part in group for r in part['evidence_refs']}
+                result = selected_extraction(selection, evidence)
-                actual = {(r.article_id, r.quote) for r in result.evidence_refs}
+                if {ref.article_id for ref in result.evidence_refs} != {ref.article_id for ref in evidence.values()}:
-                if not actual <= refs or {r[0] for r in actual} != {r[0] for r in refs}:
+                    raise ValueError('分层归纳遗漏来源')
-                    raise ValueError('分层归纳遗漏来源或改写原文引用')
                 compacted.append(result.model_dump())
             if len(json.dumps(compacted, ensure_ascii=False).encode()) >= len(json.dumps(extracts, ensure_ascii=False).encode()):
                 raise ValueError('分层归纳未缩减输入，无法在模型上下文内完整汇总')

backend/src/alphapulse/news_research/service.py

@@ -373,7 +373,6 @@ async def candidate_context(session, catalog):
             "id": e.id,
             "revision": e.revision,
             "title": e.title,
-            "tags": e.tags,
             "reporting_range": None,
             "evidence": [],
         }
@@ -876,12 +875,12 @@ async def process_event(identifier, config, force=False):
             if not researcher.fits(provider, skill, request, ResearchResult):
                 extraction_rules = (
                     skill
-                    + "\n当前事件及各成员归属边界："
+                    + "\n当前事件："
-                    + json.dumps({"event": payload["event"], "packages": payload["packages"]}, ensure_ascii=False)
+                    + json.dumps(payload["event"], ensure_ascii=False)
                     + "\nextract只提取当前片段对本事件有用的事实与连续原文引用，不作评级。consolidate保留各来源、冲突、时间与引用，不新增事实。"
                 )
                 extracts, preparations = await extract_articles(
-                    provider, extraction_rules, payload["articles"]
+                    provider, extraction_rules, payload["articles"], scopes=payload["packages"]
                 )
                 if {
                     ref["article_id"]

backend/tests/test_news_pipeline.py

@@ -23,7 +23,7 @@ def event_result(pipeline, refs):
         timeline_points=[dict(occurred_at=None, time_precision='unknown', time_text='Unknown', reported_at=None, description='Synthetic', verification='insufficient', evidence_refs=[pipeline.Evidence.model_validate(r).model_dump() for r in refs])], facts='Synthetic', timeline='Synthetic', asset_impacts='Unknown', opinions='Unknown', counter_evidence='Unknown', watchpoints='Unknown', claims=[pipeline.Claim(statement='Synthetic', verdict='insufficient', evidence_refs=[])], excluded_article_ids=[], evidence_refs=refs)
 
 
-@pytest.mark.parametrize('invalid', ['missing_source', 'rewritten_quote', 'no_progress'])
+@pytest.mark.parametrize('invalid', ['missing_source', 'unknown_reference', 'no_progress'])
 def test_compaction_rejects_lost_provenance_and_stops_without_progress(pipeline, monkeypatch, invalid):
     provider = SimpleNamespace(max_tokens=100, context_window=8000)
     extracts = [{'facts': 'x'*3000, 'evidence_refs': [{'article_id': i, 'quote': 'original'}]} for i in (1, 2)]
@@ -31,14 +31,14 @@ def test_compaction_rejects_lost_provenance_and_stops_without_progress(pipeline,
 
     async def complete(provider, system, payload, schema):
         calls.append(payload)
-        refs = [r for e in payload['extractions'] for r in e['evidence_refs']]
+        ids = [r['id'] for r in payload['evidence']]
-        if invalid == 'missing_source': refs = refs[:1]
+        if invalid == 'missing_source': ids = ids[:1]
-        if invalid == 'rewritten_quote': refs[0] = {**refs[0], 'quote': 'invented'}
+        if invalid == 'unknown_reference': ids[0] = 999
         facts = 'x'*10000 if invalid == 'no_progress' else 'compressed'
-        return pipeline.Extraction(facts=facts, evidence_refs=refs), {}
+        return schema(facts=facts, evidence_ids=ids), {}
 
     monkeypatch.setattr(pipeline, 'complete', complete)
-    message = '未缩减输入' if invalid == 'no_progress' else '遗漏来源或改写'
+    message = {'no_progress': '未缩减输入', 'missing_source': '遗漏来源', 'unknown_reference': '未知或重复证据编号'}[invalid]
     with pytest.raises(ValueError, match=message):
         asyncio.run(pipeline.compact_extractions(provider, 'rules', {'task': 'event', 'id': 1, 'extractions': extracts}, pipeline.ReviewResult))
     assert len(calls) == 1
@@ -57,20 +57,80 @@ def test_extraction_requires_original_evidence(pipeline, monkeypatch):
 
 def test_long_extraction_preserves_complete_originals_and_source_ids(pipeline, monkeypatch):
     provider = SimpleNamespace(max_tokens=1000, context_window=12000)
-    articles = [{'id': i, 'title': f'来源{i}', 'summary': '', 'content': '完整原文。' * 5000} for i in (1, 2)]
+    articles = [{'id': i, 'title': f'来源{i}', 'summary': '原文摘要。', 'content': '完整原文。' * 5000} for i in (1, 2)]
-    fragments = {1: [], 2: []}
+    fragments = {i: {field: [] for field in ('title', 'summary', 'content')} for i in (1, 2)}
+    scopes = [{'id': 10+i, 'news_id': i, 'scope_quote': f'来源{i}', 'other_scopes': []} for i in (1, 2)]
 
     async def complete(provider, rules, payload, schema):
         assert pipeline.fits_context(provider, rules, payload, schema)
-        identifier, content = payload['article_id'], payload['content']
+        identifier = payload['article_id']
-        fragments[identifier].append(content)
+        assert payload['packages'] == [scopes[identifier-1]]
-        return pipeline.Extraction(facts='保留事实。', evidence_refs=[
+        for passage in payload['passages']:
-            pipeline.Evidence(article_id=identifier, quote='完整原文。')]), {'input_tokens': 10}
+            fragments[identifier][passage['field']].append(passage['text'])
+        assert all(len(p['text']) <= 500 for p in payload['passages'])
+        return schema(facts='保留事实。', evidence_ids=[0]), {'input_tokens': 10}
 
     monkeypatch.setattr(pipeline, 'complete', complete)
-    extracts, usage = asyncio.run(pipeline.extract_articles(provider, '按原文提取事实。', articles))
+    extracts, usage = asyncio.run(pipeline.extract_articles(provider, '按原文提取事实。', articles, scopes=scopes))
     for article in articles:
-        assert ''.join(fragments[article['id']]) == article['title']+'\n\n'+article['content']
+        for field in ('title', 'summary', 'content'):
-        assert len(fragments[article['id']]) > 1
+            assert ''.join(fragments[article['id']][field]) == article[field]
-    assert len(usage) == len(extracts) == sum(len(parts) for parts in fragments.values())
+        assert len(fragments[article['id']]['content']) > 1
+    assert len(usage) == len(extracts)
     assert {ref['article_id'] for part in extracts for ref in part['evidence_refs']} == {1, 2}
+    pipeline.validate_evidence([pipeline.Evidence.model_validate(ref) for part in extracts for ref in part['evidence_refs']], {a['id']: a for a in articles})
+
+
+@pytest.mark.parametrize('selection', [[999], [0, 0]])
+def test_extraction_rejects_unknown_or_repeated_ids_and_retains_usage(pipeline, monkeypatch, selection):
+    async def complete(provider, rules, payload, schema):
+        return schema(facts='事实。', evidence_ids=selection), {'input_tokens': 17, 'output_tokens': 3}
+    monkeypatch.setattr(pipeline, 'complete', complete)
+    with pytest.raises(pipeline.SubmissionFailure, match='未知或重复证据编号') as failure:
+        asyncio.run(pipeline.extract_articles(SimpleNamespace(max_tokens=100, context_window=12000), '规则',
+            [{'id': 42, 'title': '原文标题', 'summary': '原文摘要', 'content': '原文正文'}]))
+    assert failure.value.usage['parts'] == [{'input_tokens': 17, 'output_tokens': 3}]
+
+
+def test_selected_quotes_preserve_fields_whitespace_and_foreign_language(pipeline, monkeypatch):
+    article = {'id': 42, 'title': 'Official title', 'summary': 'Another field', 'content': 'Binance\n(SignalPlus) access was cut.\nNext report.'}
+    async def complete(provider, rules, payload, schema):
+        assert schema is pipeline.EvidenceSelection
+        assert '不生成quote或article_id' in rules
+        return schema(facts='币安切断访问。', evidence_ids=[p['id'] for p in payload['passages']]), {}
+    monkeypatch.setattr(pipeline, 'complete', complete)
+    extracts, _ = asyncio.run(pipeline.extract_articles(SimpleNamespace(max_tokens=100, context_window=12000), '规则', [article]))
+    quotes = [r['quote'] for part in extracts for r in part['evidence_refs']]
+    assert ''.join(quotes) == article['title']+article['summary']+article['content']
+    assert all(any(quote in article[field] for field in ('title','summary','content')) for quote in quotes)
+    with pytest.raises(ValueError, match='原文引用'):
+        pipeline.validate_evidence([pipeline.Evidence(article_id=42, quote='Official title\nAnother field')], {42:article})
+
+
+def test_extraction_schema_never_accepts_model_generated_quotes(pipeline):
+    with pytest.raises(ValueError):
+        pipeline.EvidenceSelection.model_validate({'facts':'事实。','evidence_ids':[0],
+            'evidence_refs':[{'article_id':42,'quote':'编造引文'}]})
+
+
+def test_single_long_sentence_splits_without_losing_characters(pipeline, monkeypatch):
+    provider = SimpleNamespace(max_tokens=100, context_window=1500)
+    article = {'id': 9, 'title': '', 'summary': '', 'content': '长' * 500}
+    received = []
+    async def complete(provider, rules, payload, schema):
+        assert pipeline.fits_context(provider, rules, payload, schema)
+        received.extend(p['text'] for p in payload['passages'])
+        return schema(facts='分段事实。', evidence_ids=[0]), {}
+    monkeypatch.setattr(pipeline, 'complete', complete)
+    asyncio.run(pipeline.extract_articles(provider, '规则', [article]))
+    assert len(received) > 1 and ''.join(received) == article['content']
+
+
+def test_oversized_scope_fails_before_model_call_with_article_identity(pipeline, monkeypatch):
+    from unittest.mock import AsyncMock
+    complete = AsyncMock()
+    monkeypatch.setattr(pipeline, 'complete', complete)
+    with pytest.raises(pipeline.SubmissionFailure, match='文章9.*最小原文片段超出模型上下文'):
+        asyncio.run(pipeline.extract_articles(SimpleNamespace(max_tokens=100, context_window=1500), '规则',
+            [{'id': 9, 'title': '标题', 'content': '正文'}], scopes=[{'news_id': 9, 'scope_quote': '长' * 5000}]))
+    complete.assert_not_awaited()

backend/tests/test_news_research_workspace.py

@@ -2,6 +2,7 @@
 
 from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace as N
+import json
 import pytest
 
 
@@ -648,13 +649,12 @@ def test_long_event_compaction_uses_final_research_context(trading, monkeypatch,
     monkeypatch.setattr(service, "validate_timeline", lambda *args: None)
 
     async def consolidate(provider, rules, payload, schema):
-        assert schema is pipeline.Extraction
+        assert schema is pipeline.EvidenceSelection
-        assert "当前事件及各成员归属边界" in rules and quote in rules
+        assert "当前事件：" in rules and "当前事件及各成员归属边界" not in rules
         assert "extract和consolidate生成的facts等叙述也必须用简体中文" in rules
         assert "time_text及取自原文的period_text逐字保留原语言和字符" in rules
         assert pipeline.fits_context(provider, rules, payload, schema)
-        refs = [ref for part in payload["extractions"] for ref in part["evidence_refs"]]
+        return schema(facts="调查继续。", evidence_ids=[ref["id"] for ref in payload["evidence"]]), {"mock": True}
-        return pipeline.Extraction(facts="调查继续。", evidence_refs=refs), {"mock": True}
 
     complete = AsyncMock(side_effect=consolidate)
     monkeypatch.setattr(pipeline, "complete", complete)
@@ -679,7 +679,9 @@ def test_long_event_compaction_uses_final_research_context(trading, monkeypatch,
     compressed = {**versions[0].input, "articles": [], "extractions": extracts}
     reader = service.Research()
     assert not reader.fits(provider, skill, versions[0].input, ResearchResult)
-    assert not reader.fits(provider, boundary_rules, compressed, ResearchResult)
+    assert len(extract.await_args.kwargs["scopes"]) == 40
+    assert all(p["keywords"] == keywords for p in extract.await_args.kwargs["scopes"])
+    assert len(boundary_rules) < len(json.dumps(versions[0].input["packages"], ensure_ascii=False))
     assert reader.fits(provider, skill, compressed, ResearchResult) != needs_compaction
 
 

backend/tests/test_time_repair.py

@@ -101,7 +101,7 @@ def usage_tokens(usage):
 def test_extraction_failure_keeps_completed_and_failed_call_usage(trading, monkeypatch, failure):
     from unittest.mock import AsyncMock
     from alphapulse.news_research import pipeline, model as service
-    extract = pipeline.Extraction(facts='Source', evidence_refs=[pipeline.Evidence(article_id=1, quote='Source')])
+    extract = pipeline.EvidenceSelection(facts='Source', evidence_ids=[0 if failure == 'second_call' else 999])
     complete = AsyncMock(side_effect=[(extract, {'input_tokens': 10}), service.SubmissionFailure('synthetic failure', {'input_tokens': 40})])
     monkeypatch.setattr(pipeline, 'complete', complete)
     monkeypatch.setattr(pipeline, 'fits_context', lambda *args: True)
@@ -109,6 +109,7 @@ def test_extraction_failure_keeps_completed_and_failed_call_usage(trading, monke
     with pytest.raises(service.SubmissionFailure) as error:
         asyncio.run(pipeline.extract_articles(SimpleNamespace(), 'rules', articles))
     assert usage_tokens(error.value.usage) == (50 if failure == 'second_call' else 10)
+    assert complete.await_count == (2 if failure == 'second_call' else 1)
 
 
 @pytest.mark.parametrize('failure', ['second_call', 'evidence', 'no_shrink'])
@@ -116,14 +117,15 @@ def test_compaction_failure_keeps_every_paid_call(trading, monkeypatch, failure)
     from unittest.mock import AsyncMock
     from alphapulse.news_research import pipeline, model as service
     group = {'facts': 'Very long source fact ' * 50, 'evidence_refs': [{'article_id': 1, 'quote': 'Source'}]}
-    extract = pipeline.Extraction(facts='Source' if failure != 'no_shrink' else 'Longer' * 1000,
+    extract = pipeline.EvidenceSelection(facts='Source' if failure != 'no_shrink' else 'Longer' * 1000,
-        evidence_refs=[pipeline.Evidence(article_id=1, quote='Changed' if failure == 'evidence' else 'Source')])
+        evidence_ids=[999 if failure == 'evidence' else 0])
     complete = AsyncMock(side_effect=[(extract, {'input_tokens': 10}), service.SubmissionFailure('synthetic failure', {'input_tokens': 40})])
     monkeypatch.setattr(pipeline, 'complete', complete)
     monkeypatch.setattr(pipeline, 'fits_context', lambda *args: True)
     with pytest.raises(service.SubmissionFailure) as error:
         asyncio.run(pipeline.compact_extractions(SimpleNamespace(), 'rules', {'extractions': [group]}, pipeline.ReviewResult, fits=lambda *args: False))
     assert usage_tokens(error.value.usage) == (50 if failure == 'second_call' else 10)
+    assert complete.await_count == (2 if failure == 'second_call' else 1)
 
 
 

scripts/smoke_news_research.py

@@ -579,6 +579,7 @@ async def main():
             await s.commit()
             event = await s.get(Event, event_id)
             context = (await service.candidate_context(s, [event]))[0]
+            assert "tags" not in context and context["id"] == event.id and context["title"] == event.title
             assert context["reporting_range"] == {
                 "first": (now - timedelta(days=180)).isoformat(),
                 "latest": now.isoformat(),
@@ -766,9 +767,9 @@ async def review_regressions(config, complete, package_id, news_id):
         ).output
 
     # Force compression, then exercise both member read tools with frozen originals.
-    async def extract(provider, rules, articles):
+    async def extract(provider, rules, articles, *, scopes):
-        assert "当前事件及各成员归属边界" in rules
+        assert "当前事件：" in rules
-        assert '"keyword_canonicals": ["serialincident"]' in rules
+        assert all(p["keyword_canonicals"] == ["serialincident"] for p in scopes)
         return [
             {
                 "evidence_refs": [