diff --git a/backend/src/alphapulse/news_research/model.py b/backend/src/alphapulse/news_research/model.py index 1ccc78a..36ab07f 100644 --- a/backend/src/alphapulse/news_research/model.py +++ b/backend/src/alphapulse/news_research/model.py @@ -5,6 +5,9 @@ from alphapulse.ai import llm from alphapulse.models import AiProvider +MAX_CONTEXT_WINDOW = 250_000 + + class OutputTruncated(ValueError): """A read-only model call may be repeated with a smaller input batch.""" @@ -87,7 +90,7 @@ async def complete(provider: AiProvider, system: str, payload: dict, schema: typ def fits_context(provider, system, payload, schema): # UTF-8 bytes bound source tokens conservatively, including the tool schema. text = system + json.dumps(payload, ensure_ascii=False) + json.dumps(schema.model_json_schema(), ensure_ascii=False) - return len(text.encode('utf-8')) + provider.max_tokens <= provider.context_window + return len(text.encode('utf-8')) + provider.max_tokens <= min(provider.context_window, MAX_CONTEXT_WINDOW) async def provider_for(session, settings): provider = await session.get(AiProvider, settings.provider_id) if settings.provider_id else None diff --git a/backend/src/alphapulse/news_research/service.py b/backend/src/alphapulse/news_research/service.py index 22e9073..ef136ce 100644 --- a/backend/src/alphapulse/news_research/service.py +++ b/backend/src/alphapulse/news_research/service.py @@ -459,6 +459,7 @@ async def _process_package(identifier, config, expected_stage): attempt_id = attempt.id output = None used = None + output_audit = {} try: vocabulary = await vocab.snapshot(session) if stage == "assign": @@ -480,9 +481,12 @@ async def _process_package(identifier, config, expected_stage): ) if stage == "index": cleaned, changes = clean_aliases(output, source) + output_audit = {"output_adjustments": changes} + if changes: + output_audit["raw_output"] = output.model_dump(mode="json") + attempt.rules = {**attempt.rules, **output_audit} validate_keywords(cleaned, source) output = cleaned - attempt.rules = {**attempt.rules, "output_adjustments": changes} else: validate_assignment(output, source, {e.id for e in catalog}, payload["keywords"]) async with publication_lock: @@ -586,6 +590,7 @@ async def _process_package(identifier, config, expected_stage): except Exception as error: await session.rollback() attempt = await session.get(Attempt, attempt_id) + attempt.rules = {**attempt.rules, **output_audit} attempt.state = "failed" attempt.error = str(error)[:2000] attempt.output = output.model_dump(mode="json") if output else None @@ -859,6 +864,7 @@ async def process_event(identifier, config, force=False): researcher = Research(event={"id": identifier, "revision": revision}) output = None used = None + output_audit = {} preparations = [] try: attempts = {a.package_id: a for a in await session.scalars( @@ -930,9 +936,11 @@ async def process_event(identifier, config, force=False): sources = {**{a["id"]: a for a in payload["articles"]}, **researcher.sources} raw_output = output.model_dump(mode="json") changes = align_quotes(raw_output, sources) + output_audit = {"output_adjustments": changes} if changes: + output_audit["raw_output"] = output.model_dump(mode="json") output = ResearchResult.model_validate(raw_output) - version.rules = {**version.rules, "output_adjustments": changes} + version.rules = {**version.rules, **output_audit} validate_event(output, sources) if output.id != identifier: raise ValueError("返回事件ID不匹配") @@ -985,6 +993,7 @@ async def process_event(identifier, config, force=False): except Exception as error: await session.rollback() version = await session.get(Version, version_id) + version.rules = {**version.rules, **output_audit} version.state = "failed" version.error = str(error)[:2000] version.evidence = researcher.records diff --git a/backend/src/alphapulse/news_research/tools.py b/backend/src/alphapulse/news_research/tools.py index d47848e..ebf1801 100644 --- a/backend/src/alphapulse/news_research/tools.py +++ b/backend/src/alphapulse/news_research/tools.py @@ -6,7 +6,7 @@ from pydantic import BaseModel, Field from alphapulse.ai import llm from alphapulse.core.db import SessionLocal from alphapulse.models import NewsItem -from alphapulse.news_research.model import fits_context, validate_submission, model_stream +from alphapulse.news_research.model import MAX_CONTEXT_WINDOW, fits_context, validate_submission, model_stream class Search(BaseModel): @@ -54,7 +54,8 @@ class Research: def fits(self, provider, system, payload, schema): reserved = copy.copy(provider) - reserved.context_window -= min(32768, max(0, (provider.context_window-provider.max_tokens)//5)) + reserved.context_window = min(provider.context_window, MAX_CONTEXT_WINDOW) + reserved.context_window -= min(32768, max(0, (reserved.context_window-provider.max_tokens)//5)) return self.fits_messages(reserved, system, [{'role': 'user', 'text': json.dumps(payload, ensure_ascii=False)}], schema) def restore(self, records): @@ -125,11 +126,13 @@ class Research: usages = self.usages = [] from alphapulse.news_research.progress import step, current final_only = False + limited = {'status':'context_limit', 'instruction':'资料未完整交付,不得引用;依据已有证据提交并说明缺口。'} + limited_result = json.dumps(limited, ensure_ascii=False) for round_no in range(9): if current.get() is not None: step('模型研判', round=round_no+1, model=provider.model, reasoning_effort=provider.reasoning_effort) final_only |= round_no == 8 if not self.fits_messages(provider, system, messages, schema, final=final_only): - raise ValueError('研究上下文不足,请增大上下文配置;证据未截断') + raise ValueError('研究上下文不足;证据未截断或继续发送') completion = None async for item in model_stream(provider, system, messages, [submit] if final_only else [*specs, submit], provider.reasoning_effort): if isinstance(item, llm.Completion): completion = item @@ -148,20 +151,22 @@ class Research: snapshot = {name: copy.deepcopy(getattr(self, name)) for name in ( 'sources', 'provided_news', 'provided_packages', 'read_event_versions', 'read_event_ids')} records_before = len(self.records) - try: result = await self.tool(call['name'], call['arguments']) if not final_only else {'status':'context_limit', 'instruction':'停止查询,依据已有证据提交并说明缺口。'} + try: result = await self.tool(call['name'], call['arguments']) if not final_only else limited except Exception as error: result = {'error': str(error)} self.records.append({'tool': call['name'], 'arguments': call['arguments'], 'result': result}) response = {'role':'tool', 'tool_call_id':call['id'], 'result':json.dumps(result, ensure_ascii=False, default=str)} - remaining = [{'role':'tool', 'tool_call_id':c['id'], 'result':json.dumps({'status':'context_limit'})} for c in calls[index+1:]] + remaining = [{'role':'tool', 'tool_call_id':c['id'], 'result':limited_result} for c in calls[index+1:]] if not self.fits_messages(provider, system, [*messages, response, *remaining], schema, final=True): for name, value in snapshot.items(): setattr(self, name, value) - result = {'status':'context_limit', 'instruction':'资料未完整交付,不得引用;依据已有证据提交并说明缺口。'} + result = limited for record in self.records[records_before:]: record['undelivered_result'] = record['result'] record['result'] = result - response['result'] = json.dumps(result, ensure_ascii=False) + response['result'] = limited_result final_only = True + if not self.fits_messages(provider, system, [*messages, response, *remaining], schema, final=True): + raise ValueError('研究上下文不足;证据未截断或继续发送') messages.append(response) if not self.fits_messages(provider, system, messages, schema): final_only = True raise ValueError('研究轮次已用尽') diff --git a/backend/tests/test_news_pipeline.py b/backend/tests/test_news_pipeline.py index fef2ffb..c911a57 100644 --- a/backend/tests/test_news_pipeline.py +++ b/backend/tests/test_news_pipeline.py @@ -246,3 +246,48 @@ def test_undelivered_source_is_not_restored_as_read(trading): reader.restore([{'tool':'read_news','arguments':{'id':99}, 'result':{'status':'context_limit'},'undelivered_result':{'article_id':-1,'content':'原文'}}]) assert not reader.sources and not reader.cache and not reader.provided_news + + +@pytest.mark.parametrize('context_window,body_length', [(8000, 7310), (250000, 249310)]) +def test_parallel_tool_responses_leave_room_for_final_submission(trading, monkeypatch, context_window, body_length): + from alphapulse.news_research.tools import Research + from alphapulse.news_research import model + from alphapulse.ai.llm import Completion + from pydantic import BaseModel + + class Result(BaseModel): + facts: str + + reader = Research() + requests = [] + tool_calls = [{'id': str(i), 'name': 'read_news', 'arguments': {'id': i}} for i in (1, 2)] + + async def stream(provider, system, messages, tools, effort): + requests.append(messages.copy()) + if len(requests) == 1: + yield Completion(tool_calls=tool_calls, usage={'input_tokens': 10}) + else: + assert [t.name for t in tools] == ['submit'] + assert reader.fits_messages(provider, system, messages, Result, final=True) + responses = [m for m in messages if m['role'] == 'tool'] + assert [r['tool_call_id'] for r in responses] == ['1', '2'] + assert all('context_limit' in r['result'] for r in responses) + yield Completion(tool_calls=[{'id': '3', 'name': 'submit', 'arguments': {'facts': '依据已有资料提交,补证不足。'}}], + usage={'input_tokens': 20}) + + async def tool(name, args): + assert args['id'] == 1 + result = {'content': 'x' * body_length} + reader.sources[-1] = result + reader.provided_news[1] = -1 + reader.records.append({'tool': name, 'arguments': args, 'result': result}) + return result + + monkeypatch.setattr(reader, 'tool', tool) + monkeypatch.setattr(model.llm, 'stream', stream) + provider = SimpleNamespace(model='test', reasoning_effort='medium', max_tokens=100, context_window=context_window) + output, usage = asyncio.run(reader.run(provider, '规则', {}, Result)) + assert output.facts and len(requests) == len(usage['rounds']) == 2 + assert not reader.sources and not reader.provided_news + assert len(reader.records) == 1 + assert reader.records[0]['undelivered_result']['content'] == 'x' * body_length diff --git a/backend/tests/test_news_research_workspace.py b/backend/tests/test_news_research_workspace.py index 19e8dd4..04f8627 100644 --- a/backend/tests/test_news_research_workspace.py +++ b/backend/tests/test_news_research_workspace.py @@ -685,6 +685,132 @@ def test_long_event_compaction_uses_final_research_context(trading, monkeypatch, assert reader.fits(provider, skill, compressed, ResearchResult) != needs_compaction +@pytest.mark.parametrize("failure_at", [None, "evidence", "timeline", "publication"]) +def test_quote_adjustment_audit_survives_validation_and_publication_failures(trading, monkeypatch, failure_at): + import asyncio + import copy + from unittest.mock import AsyncMock + from alphapulse.news_research import service + from alphapulse.news_research.contracts import ResearchResult + from alphapulse.news_research.models import Version + + quote, original_quote = "官员“感到担忧”。", "官员“感到担忧” 。" + event = N(id=1, revision=1, title="官员回应", state="pending", result=None, + due_at=datetime.now(UTC) - timedelta(seconds=1)) + news = N(id=101, title="官员回应", summary="", content_text=quote, content_hash="hash", + url="https://example.invalid/news", published_at=datetime.now(UTC)) + package = N(id=7, news_id=101, state="done", content_hash="hash") + member = N(event_id=1, package_id=7, evidence=quote, reason="同一回应") + assignment = N(package_id=7, content_hash="hash", input={"keywords": [ + {"canonical": "官员", "aliases": ["官员"], "quote": quote}]}, + rules={"membership_targets": {"1": {"keyword_canonicals": ["官员"]}}}) + output = ResearchResult( + id=1, title="官员回应", summary="官员表达担忧。", category="政策", entities=[], + risk_stars=1, importance_stars=2, direction="unknown", verification="reported", confidence="low", + reasoning="依据报道。", uncertainty="缺少独立佐证。", facts="官员表达担忧。", timeline="官员回应。", + asset_impacts="未知。", opinions="持续观察。", counter_evidence="未发现。", watchpoints="后续回应。", + claims=[{"statement": "官员表达担忧。", "verdict": "supported", "evidence_refs": [ + {"article_id": 101, "quote": "不存在的引文" if failure_at == "evidence" else quote}]}], + evidence_refs=[{"article_id": 101, "quote": original_quote}], + timeline_points=[{"occurred_at": None, "time_precision": "unknown", "time_text": "", + "reported_at": None, "description": "官员表达担忧。", "verification": "supported", + "package_ids": [999 if failure_at == "timeline" else 7], + "evidence_refs": [{"article_id": 101, "quote": quote}]}], references=[]) + original = output.model_dump(mode="json") + versions, saved_rules = [], {} + + class Session: + async def __aenter__(self): return self + async def __aexit__(self, *args): pass + def add(self, row): + row.id = 1 + versions.append(row) + async def commit(self): + nonlocal saved_rules + if versions[0].state == "done" and failure_at == "publication": + raise RuntimeError("发布写入失败") + saved_rules = copy.deepcopy(versions[0].rules) + async def rollback(self): + # Rollback discards uncommitted JSON changes, as the real ORM does. + versions[0].rules = copy.deepcopy(saved_rules) + + session = Session() + session.get = AsyncMock(side_effect=lambda model, *args, **kwargs: versions[0] if model is Version else event) + session.execute = AsyncMock(side_effect=[[(package, member, news)], N(all=lambda: [(101, "hash")])]) + session.scalars = AsyncMock(return_value=[assignment]) + session.scalar = AsyncMock(return_value=None) + monkeypatch.setattr(service, "SessionLocal", lambda: session) + monkeypatch.setattr(service, "model_config", AsyncMock(return_value=( + N(max_tokens=32000, context_window=250000), "研究规则", {"system_prompt": "研究规则"}))) + monkeypatch.setattr(service.vocab, "snapshot", AsyncMock(return_value={})) + monkeypatch.setattr(service.Research, "run", AsyncMock(return_value=(output, {"input_tokens": 10}))) + + asyncio.run(service.process_event(1, {})) + version = versions[0] + assert version.state == ("failed" if failure_at else "done"), version.error + if failure_at: + assert {"evidence": "原文引用", "timeline": "未知新闻包", "publication": "发布写入失败"}[failure_at] in version.error + assert version.output["evidence_refs"][0]["quote"] == quote + assert version.rules["raw_output"] == original + assert version.rules["output_adjustments"] == [{ + "path": ".evidence_refs.0.quote", "original": original_quote, "replacement": quote, + "reason": "unique_punctuation_whitespace_alignment"}] + assert version.rules["system_prompt"] == "研究规则" + assert news.content_text == quote and output.model_dump(mode="json") == original + + +@pytest.mark.parametrize("failure_at", [None, "validation", "publication"]) +def test_alias_adjustment_audit_survives_index_failures(trading, monkeypatch, failure_at): + import asyncio + import copy + from unittest.mock import AsyncMock + from alphapulse.news_research import service + from alphapulse.news_research.contracts import IndexResult + from alphapulse.news_research.models import Attempt, Package + + news = N(id=101, title="", summary="", content_text="插件调用单独计费", content_hash="hash", + url="https://example.invalid/news", published_at=datetime.now(UTC)) + package = N(id=7, news_id=101, state="index_pending", content_hash="hash") + output = IndexResult(keywords=[{"canonical": "插件计费", "aliases": ["插件调用", "插件调用收费"], + "quote": "不存在的引文" if failure_at == "validation" else "插件调用单独计费"}]) + original = output.model_dump(mode="json") + attempts, saved_rules = [], {} + + class Session: + async def __aenter__(self): return self + async def __aexit__(self, *args): pass + def add(self, row): + row.id = 1 + attempts.append(row) + async def commit(self): + nonlocal saved_rules + if attempts[0].state == "done" and failure_at == "publication": + raise RuntimeError("发布写入失败") + saved_rules = copy.deepcopy(attempts[0].rules) + async def rollback(self): + attempts[0].rules = copy.deepcopy(saved_rules) + + session = Session() + session.get = AsyncMock(side_effect=lambda model, *args, **kwargs: + attempts[0] if model is Attempt else package if model is Package else news) + session.execute = AsyncMock() + monkeypatch.setattr(service, "SessionLocal", lambda: session) + monkeypatch.setattr(service.vocab, "snapshot", AsyncMock(return_value={"entries": []})) + monkeypatch.setattr(service, "model_config", AsyncMock(return_value=(None, "索引规则", {}))) + monkeypatch.setattr(service, "complete", AsyncMock(return_value=(output, {"input_tokens": 10}))) + + asyncio.run(service._process_package(7, {}, "index")) + attempt = attempts[0] + assert attempt.state == ("failed" if failure_at else "done"), attempt.error + if failure_at: + assert {"validation": "原文引用", "publication": "发布写入失败"}[failure_at] in attempt.error + assert attempt.rules["raw_output"] == original + assert attempt.rules["output_adjustments"] == [{ + "path": "keywords.0.aliases", "original": ["插件调用", "插件调用收费"], + "replacement": ["插件调用"], "reason": "aliases_absent_from_source"}] + assert news.content_text == "插件调用单独计费" and output.model_dump(mode="json") == original + + @pytest.mark.parametrize("task", ["index", "assign", "research"]) def test_model_config_sends_only_selected_task_and_records_exact_prompt(trading, monkeypatch, task): import asyncio diff --git a/backend/tests/test_news_risk.py b/backend/tests/test_news_risk.py index 3677ad4..6095044 100644 --- a/backend/tests/test_news_risk.py +++ b/backend/tests/test_news_risk.py @@ -36,3 +36,38 @@ def test_context_budget_includes_unicode_and_tool_schema(risk): from types import SimpleNamespace provider = SimpleNamespace(max_tokens=256, context_window=3500) assert not risk.fits_context(provider, '只使用原文证据', {'content': '风'*1000}, IndexResult) + + +@pytest.mark.parametrize('context_window', [250000, 1000000]) +def test_news_context_is_capped_at_250k_including_output(risk, monkeypatch, context_window): + from pydantic import BaseModel + from types import SimpleNamespace + from unittest.mock import AsyncMock + + class Result(BaseModel): + facts: str + + provider = SimpleNamespace(max_tokens=32000, context_window=context_window) + assert risk.fits_context(provider, '', {'content': 'x' * 200000}, Result) + stream = AsyncMock() + monkeypatch.setattr(risk.llm, 'stream', stream) + with pytest.raises(ValueError, match='超出模型上下文'): + asyncio.run(risk.complete(provider, '', {'content': 'x' * 220000}, Result)) + stream.assert_not_called() + + +def test_research_reserves_tool_room_within_250k_cap(risk): + from alphapulse.news_research.tools import Research + from pydantic import BaseModel + from types import SimpleNamespace + + class Result(BaseModel): + facts: str + + reader = Research() + payload = {'content': 'x' * 190000} + for context_window in (250000, 1000000): + provider = SimpleNamespace(max_tokens=32000, context_window=context_window) + assert risk.fits_context(provider, '', payload, Result) + assert not reader.fits(provider, '', payload, Result) + assert provider.context_window == context_window