From 2cc5598ebd1644abf35defd75b45ae6c7cd19567 Mon Sep 17 00:00:00 2001 From: eezd Date: Fri, 6 Nov 2026 10:52:02 +0800 Subject: [PATCH] Bound research context and audit source-grounded output corrections --- .../src/alphapulse/news_research/contracts.py | 4 +- .../src/alphapulse/news_research/pipeline.py | 43 +++++++++++ .../src/alphapulse/news_research/service.py | 48 ++++++++++--- backend/src/alphapulse/news_research/tools.py | 39 +++++++--- backend/tests/test_news_pipeline.py | 72 +++++++++++++++++++ backend/tests/test_news_research_workspace.py | 2 +- 6 files changed, 186 insertions(+), 22 deletions(-) diff --git a/backend/src/alphapulse/news_research/contracts.py b/backend/src/alphapulse/news_research/contracts.py index f711663..3ade63d 100644 --- a/backend/src/alphapulse/news_research/contracts.py +++ b/backend/src/alphapulse/news_research/contracts.py @@ -16,6 +16,7 @@ ASSIGN_SKILL = """你负责将一个完整新闻包归属到具体事件。外 无论原文语言，生成的title、reason均用简体中文，主体名称沿用中文规范名；没有可靠中文名的专有名称、产品型号及通用缩写保留。quote逐字保留原语言，不翻译。keyword_canonicals必须逐字复用输入TAG，即使历史TAG是英文也不得在归属时翻译或重做索引。JSON键、event_id、new_key等标识及change_kind枚举值保持协议格式，不翻译。 一个包可以关联多个事件：综述涉及多件独立事件时，各填一个归属，以对应连续原文quote说明相关部分，不能因主题相同而合并独立事件。 事件以主体、具体事项和发生批次或统计期为边界；归属reason明确说明同一事实主线及原文依据。直接后果、回应、调查、更正、赔付属于原事件；独立决策或行动、新事故、新统计期另建事件，不能因为相关而强行合并。原文明示由某次数据发布引起的直接市场反应归入该次发布；独立政策决策另建。例：瑞士10月CPI环比、同比和受该次CPI影响的欧元兑瑞郎反应同属发布事件；瑞士11月CPI另建。币安同一次被盗及数月后追偿同属一事，第二次被盗另建。单次事故的资金追踪、冻结或追回即使由另一国家机关执行，也属于原事故直接处置，不能仅因执行主体或国家不同另建；只有原文明示独立案件范围（例如同一平台处理多起无关犯罪的执法）才另建并引用事故。关联不明时独立建事件并在reason说明缺口，不编造因果。 +已有事件尚未包含这次回应，不是另建事件的理由。例如同一次美国中期选举后，万斯解释败选原因、特朗普回应，应归入该次选举事件；正式宣布的新政策决定或独立行动另建事件。 优先检查给定的长期事件候选；同一次事故、回应、调查、损失修订和处置可属于同一事件，即使横跨数月。仅主体相同不是同一事件。 候选reporting_range是报道时间范围，evidence包含首条与近期成员原文线索，不等于事件发生时间；结合quote中的日期、统计期及具体事项区分每日数据、不同次事故等独立事件，不能仅因名称或TAG相同而归并，也不能仅因报道间隔长而拆开同一事件。 每个target填写event_id或new_key且恰好一个非空、title、reason、quote、keyword_canonicals和change_kind、urgent。已有事件沿用ID；无合适候选则创建可独立识别的新事件名称，不编写事件总结。至少归属一个事件，单包事件也成立。title与quote只包含本事件事项，quote保留其主体、动作和对象；综述的其他事项不要混入。不能仅引用“下跌”“另有”等失去主体的片段；原文确实未具名时不补造。 @@ -31,7 +32,8 @@ summary为事件总结；facts为已知事实；reasoning为判断依据；opini importance_stars独立评估：1背景、2局部、3重要、4广泛、5系统性。risk_stars：0无具体负面迹象、1关注、2警惕、3重大损害、4严重广泛冲击、5系统性危机。证据不充分降低confidence，不机械提高风险。verification=insufficient时direction=unknown。 claims聚焦原文明示的关键主张、实质冲突及会改变判断的误读；不凭空增加“已获官方认证”“已决定全国控制权”等主张再机械判为不足。报道某人作出预测与预测已经实现分开。先核对数字、单位、期间和口径，保留算术反证能力。summary概括结论，reasoning说明推理，uncertainty集中写缺口，counter_evidence写实质反证或说明未发现，不在各字段重复整段限制。official仅表示核心主张得到实际读取的官方正文支持，不因有一个官方链接就覆盖其他未核实主张。 每次站内查询先明确要解决的关键缺口：是否会改变事件身份、核心事实、核验状态、评级或判断。普通行情可直接按reported提交，不强查原因或收盘状态；重大事故、矛盾数字和关键政策优先通过已采集的站内原文核验。不得外部搜索、读取网页或访问URL。不为补齐字段查询；没有有效站内线索时提交，并集中说明证据局限。可独立进行的站内查询合并在同轮调用，不为每项claims各查一轮。 -articles已有成员完整原文时直接复用正文和文章id，不再逐包调用read_package/read_news，也不把重复读取当独立核验；仅输入extractions而无全文时，可按需读取对应成员原文。工具返回already_provided即使用先前已提供正文及article_id，不重试同一成员，也不继续逐个读取其他已提供成员。先列出会改变判断的最少证据缺口，将无需依赖彼此结果的站内查询合并同一轮；拿到线索后才执行依赖它的读取。每轮检查是否有新增有效证据；没有可解决关键缺口的新具体线索时立即submit，未解决项写uncertainty，不为凑核验轮次或空字段继续查询，不跳过重大事实所需的有效取证。 +articles已有成员完整原文时直接复用正文和文章id，不再逐包调用read_package/read_news，也不把重复读取当独立核验；仅输入extractions而无全文时，可按需读取对应成员原文。工具返回context_limit表示该资料未完整交付，不得引用该资料，停止查询并依据已有证据提交，将缺口写入uncertainty。检索结果仅用于定位，不能替代原文。 +工具返回already_provided即使用先前已提供正文及article_id，不重试同一成员，也不继续逐个读取其他已提供成员。先列出会改变判断的最少证据缺口，将无需依赖彼此结果的站内查询合并同一轮；拿到线索后才执行依赖它的读取。每轮检查是否有新增有效证据；没有可解决关键缺口的新具体线索时立即submit，未解决项写uncertainty，不为凑核验轮次或空字段继续查询，不跳过重大事实所需的有效取证。 claims逐项核验，只有可能改变结论的缺口才查询站内新闻包、事件或经济日历。不机械耗尽预算；同稿转载和多个事件引用同一原文不是独立确认。URL不是已读证据，不得联网核验。站内已采集的官方原文可支持official；证据不足时明确缺口并据实降低核验状态和confidence，不编造补证。恢复客户资产余额不等于追回被盗资金；金额估值、可追踪比例须保留各自时点与来源。 时间线timeline_points按实际发展组织，转载可共用一个节点，不强制每包一节点；成员包进展的节点填写package_ids及各包对应原文evidence_refs；packages.id是包ID，packages.news_id是文章ID，二者不能混用。仅由工具补充材料支撑的背景节点package_ids填空数组，不能冒用成员包ID。发生、计划、报道时间和统计期分开；未知为null/unknown。time_text逐字取自某条quote，不用报道时间替代发生时间。 日和月精度的occurred_at/planned_at统一以UTC零点存储日期标记（日取当日、月取首日），它不是实际发生瞬间，不按事件所在国家猜时区；明确的当地日期保留其原日期，不先换算UTC日期。分钟精度仅在原文明确时区或可由具名交易场所明确确定时填写，否则保留time_text并填null/unknown。只有报道日期不能推定发生日期。季度及年度保留period_text，不伪造月或日精度。 diff --git a/backend/src/alphapulse/news_research/pipeline.py b/backend/src/alphapulse/news_research/pipeline.py index 2e7d970..abd7c11 100644 --- a/backend/src/alphapulse/news_research/pipeline.py +++ b/backend/src/alphapulse/news_research/pipeline.py @@ -54,6 +54,49 @@ def validate_evidence(refs, articles): if source is None or not ref.quote.strip() or not any(ref.quote in (source.get(k) or '') for k in ('title', 'summary', 'content')): raise ValueError('模型证据并非对应文章原文引用') +def align_quotes(value, articles): + """Recover unique original spans only when punctuation-adjacent whitespace differs.""" + punctuation = set('，。！？；：、,.;:!?“”‘’"\'()（）[]【】') + def normalized(text): + chars, positions = [], [] + for match in re.finditer(r'\S|\s+', text): + part = match.group() + if part.isspace() and ((match.start() and text[match.start()-1] in punctuation) + or (match.end() < len(text) and text[match.end()] in punctuation)): + continue + chars.extend(part) + positions.extend(range(match.start(), match.end())) + return ''.join(chars), positions + changes = [] + def visit(item, path): + if isinstance(item, list): + for i, child in enumerate(item): visit(child, f'{path}.{i}') + elif isinstance(item, dict): + if 'article_id' in item and 'quote' in item: + quote = item['quote'] + source = articles.get(item['article_id'], {}) + if not quote.strip() or any(quote in (source.get(k) or '') for k in ('title','summary','content')): + return + needle, _ = normalized(quote) + matches = [] + if needle: + for field in ('title','summary','content'): + original = source.get(field) or '' + haystack, positions = normalized(original) + start = haystack.find(needle) + while start >= 0: + matches.append(original[positions[start]:positions[start+len(needle)-1]+1]) + start = haystack.find(needle, start+1) + if len(matches) == 1 and len(matches[0]) <= 500: + item['quote'] = matches[0] + changes.append({'path':path+'.quote','original':quote,'replacement':matches[0], + 'reason':'unique_punctuation_whitespace_alignment'}) + else: + for key, child in item.items(): visit(child, f'{path}.{key}') + visit(value, '') + return changes + + def validate_event(result, articles): if result.excluded_article_ids: raise ValueError('事件包含不相关文章：'+str(result.excluded_article_ids)) diff --git a/backend/src/alphapulse/news_research/service.py b/backend/src/alphapulse/news_research/service.py index fb6e18f..22e9073 100644 --- a/backend/src/alphapulse/news_research/service.py +++ b/backend/src/alphapulse/news_research/service.py @@ -14,6 +14,7 @@ from alphapulse.news_research.pipeline import ( validate_evidence, Evidence, validate_event, + align_quotes, extract_articles, compact_extractions, ) @@ -74,6 +75,22 @@ def validate_keywords(result, source): raise ValueError("关键词别名必须实际出现在本篇原文中") +def clean_aliases(result, source): + """Retain every TAG; audit invalid aliases only when a valid alias remains.""" + cleaned = result.model_copy(deep=True) + changes = [] + for i, keyword in enumerate(cleaned.keywords): + valid = [alias for alias in keyword.aliases if alias.strip() and any( + alias in (source.get(field) or '') for field in ('title','summary','content'))] + if not valid: + raise ValueError('关键词没有任何原文有效别名') + if valid != keyword.aliases: + changes.append({'path':f'keywords.{i}.aliases','original':keyword.aliases, + 'replacement':valid,'reason':'aliases_absent_from_source'}) + keyword.aliases = valid + return cleaned, changes + + def validate_assignment(result, source, candidate_ids, keywords): for target in result.targets: if target.event_id is not None and target.event_id not in candidate_ids: @@ -462,7 +479,10 @@ async def _process_package(identifier, config, expected_stage): IndexResult if stage == "index" else Assignment, ) if stage == "index": - validate_keywords(output, source) + cleaned, changes = clean_aliases(output, source) + 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: @@ -691,14 +711,15 @@ class Research(ResearchTools): .limit(20) ) result = [ - {"id": r.id, "title": r.title, "tags": r.tags} for r in rows + {"id": r.id, "title": r.title, "tags": [t for t in r.tags + if any(term.lower() in t.lower() for term in terms[:6])]} for r in rows ] else: # Retain the phrase in addition to the bounded search terms. condition, score = package_matching([query, *terms[:6]], vocabulary) phrase, _ = package_matching([query], vocabulary) rows = await session.execute( - select(Package, NewsItem.content_hash) + select(Package, NewsItem.content_hash, NewsItem.title) .join(NewsItem) .where(condition) .order_by( @@ -710,9 +731,13 @@ class Research(ResearchTools): .limit(20) ) result = [ - {"id": r.id, "news_id": r.news_id, - "keywords": r.keywords if r.content_hash == source_hash else []} - for r, source_hash in rows + {"id": r.id, "news_id": r.news_id, "title": title, + "keywords": [{"canonical": k["canonical"], "aliases": [a for a in k["aliases"] + if any(t.lower() in a.lower() for t in terms[:6])]} + for k in r.keywords if any(t.lower() in word.lower() + for t in terms[:6] for word in [k["canonical"], *k["aliases"]])] + if r.content_hash == source_hash else []} + for r, source_hash, title in rows ] elif name == "read_package": row = await session.get(Package, arguments["id"]) @@ -902,10 +927,13 @@ async def process_event(identifier, config, force=False): ) if preparations: used = {"preparation": preparations, "research": used} - validate_event( - output, - {**{a["id"]: a for a in payload["articles"]}, **researcher.sources}, - ) + sources = {**{a["id"]: a for a in payload["articles"]}, **researcher.sources} + raw_output = output.model_dump(mode="json") + changes = align_quotes(raw_output, sources) + if changes: + output = ResearchResult.model_validate(raw_output) + version.rules = {**version.rules, "output_adjustments": changes} + validate_event(output, sources) if output.id != identifier: raise ValueError("返回事件ID不匹配") package_news = {p.id: n.id for p, m, n in members} diff --git a/backend/src/alphapulse/news_research/tools.py b/backend/src/alphapulse/news_research/tools.py index 5804bb4..d47848e 100644 --- a/backend/src/alphapulse/news_research/tools.py +++ b/backend/src/alphapulse/news_research/tools.py @@ -1,5 +1,6 @@ """Read-only news and external research tools with bounded calls.""" import json +import copy from sqlalchemy import select, or_ from pydantic import BaseModel, Field from alphapulse.ai import llm @@ -47,18 +48,20 @@ class Research: def tool_specs(): return [llm.ToolSpec(n, description, model.model_json_schema()) for n, (model, description) in TOOLS.items()] - def fits_messages(self, provider, system, messages, schema): + def fits_messages(self, provider, system, messages, schema, *, final=False): return fits_context(provider, system, {'messages': messages, 'tools': [ - {'name': t.name, 'description': t.description, 'parameters': t.parameters} for t in self.tool_specs()]}, schema) + {'name': t.name, 'description': t.description, 'parameters': t.parameters} for t in ([] if final else self.tool_specs())]}, schema) def fits(self, provider, system, payload, schema): - return self.fits_messages(provider, system, [{'role': 'user', 'text': json.dumps(payload, ensure_ascii=False)}], schema) + reserved = copy.copy(provider) + reserved.context_window -= min(32768, max(0, (provider.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): for record in records: name, args, result = record['tool'], record['arguments'], record['result'] if name not in TOOLS: continue - if isinstance(result, dict) and 'error' in result: continue + if isinstance(result, dict) and ('error' in result or result.get('status') == 'context_limit'): continue # Membership and revisions can change between attempts; reread local objects. if name in ('read_event', 'read_package'): continue self.cache[name+json.dumps(args, sort_keys=True)] = result @@ -121,12 +124,14 @@ class Research: messages = [{'role': 'user', 'text': json.dumps(payload, ensure_ascii=False)}] usages = self.usages = [] from alphapulse.news_research.progress import step, current + final_only = 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) - if not self.fits_messages(provider, system, messages, schema): + final_only |= round_no == 8 + if not self.fits_messages(provider, system, messages, schema, final=final_only): raise ValueError('研究上下文不足，请增大上下文配置；证据未截断') completion = None - async for item in model_stream(provider, system, messages, [*specs, submit] if round_no < 8 else [submit], provider.reasoning_effort): + 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 if completion is not None: usages.append(completion.usage) if completion is None or completion.truncated: raise ValueError('研究输出未完成或被截断') @@ -135,14 +140,28 @@ class Research: if len(calls) != 1: raise ValueError('提交结果不能同时执行查询') result = await validate_submission(provider, system, messages, schema, completion, usages) return result, {'rounds': usages, 'tools': self.records} - if round_no == 8 or not calls: raise ValueError('研究模型未提交结果') + if final_only or not calls: raise ValueError('研究模型未提交结果') assistant = {'role': 'assistant', 'text': completion.text, 'tool_calls': calls} if completion.thinking_blocks: assistant['thinking_blocks'] = completion.thinking_blocks messages.append(assistant) - for call in calls: - try: result = await self.tool(call['name'], call['arguments']) + for index, call in enumerate(calls): + 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':'停止查询，依据已有证据提交并说明缺口。'} except Exception as error: result = {'error': str(error)} self.records.append({'tool': call['name'], 'arguments': call['arguments'], 'result': result}) - messages.append({'role': 'tool', 'tool_call_id': call['id'], 'result': json.dumps(result, ensure_ascii=False, default=str)}) + 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:]] + 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':'资料未完整交付，不得引用；依据已有证据提交并说明缺口。'} + for record in self.records[records_before:]: + record['undelivered_result'] = record['result'] + record['result'] = result + response['result'] = json.dumps(result, ensure_ascii=False) + final_only = True + 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 b42d27a..fef2ffb 100644 --- a/backend/tests/test_news_pipeline.py +++ b/backend/tests/test_news_pipeline.py @@ -174,3 +174,75 @@ def test_oversized_scope_fails_before_model_call_with_article_identity(pipeline, 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() + + +def test_alias_cleanup_preserves_tags_valid_aliases_and_raw_output(trading): + from alphapulse.news_research.contracts import IndexResult + from alphapulse.news_research.service import clean_aliases, validate_keywords + raw = IndexResult(keywords=[dict(canonical='插件调用计费', aliases=['插件调用','插件调用收费'], quote='插件调用单独计费')]) + source = {'id':1, 'content':'插件调用单独计费'} + output, changes = clean_aliases(raw, source) + validate_keywords(output, source) + assert output.keywords[0].aliases == ['插件调用'] + assert raw.keywords[0].aliases == ['插件调用','插件调用收费'] + assert changes[0]['original'] == raw.keywords[0].aliases + with pytest.raises(ValueError, match='没有任何原文有效别名'): + clean_aliases(raw, {'id':1,'content':'其他文字'}) + + +@pytest.mark.parametrize('source,quote,corrected', [ + ('官员“感到担忧”。','官员“感到担忧” 。',True), + ('官员“感到担忧”。官员“感到担忧”。','官员“感到担忧” 。',False), + ('金额100美元。','金额1000美元。',False), + ('US D company.','USD company.',False), +]) +def test_quote_alignment_only_recovers_unique_punctuation_whitespace(pipeline, source, quote, corrected): + value = {'claims':[{'evidence_refs':[{'article_id':-2,'quote':quote}]}]} + changes = pipeline.align_quotes(value, {-2:{'content':source}}) + assert bool(changes) == corrected + if corrected: + pipeline.validate_evidence([pipeline.Evidence.model_validate(value['claims'][0]['evidence_refs'][0])], {-2:{'content':source}}) + assert changes[0]['original'] == quote + else: + assert value['claims'][0]['evidence_refs'][0]['quote'] == quote + + +def test_oversized_tool_response_is_not_delivered_or_registered(trading, monkeypatch): + 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() + calls = [] + async def stream(provider, system, messages, tools, effort): + calls.append(messages.copy()) + if len(calls) == 1: + yield Completion(tool_calls=[{'id':'1','name':'read_news','arguments':{'id':99}}],usage={'input_tokens':10}) + else: + assert [t.name for t in tools] == ['submit'] + assert 'context_limit' in messages[-1]['result'] + assert 'x'*100 not in messages[-1]['result'] + yield Completion(tool_calls=[{'id':'2','name':'submit','arguments':{'facts':'证据不足'}}],usage={'input_tokens':20}) + async def tool(name, args): + result = {'article_id':-1,'content':'x'*20000} + reader.sources[-1] = result + reader.provided_news[99] = -1 + reader.records.append({'tool':name,'arguments':args,'result':result}) + return result + monkeypatch.setattr(reader, 'tool', tool) + monkeypatch.setattr(model.llm, 'stream', stream) + output, usage = asyncio.run(reader.run(SimpleNamespace(model='test',reasoning_effort='medium',max_tokens=100,context_window=8000), '规则', {}, Result)) + assert output.facts == '证据不足' and len(usage['rounds']) == 2 + assert not reader.sources and not reader.provided_news + assert reader.records[0]['result']['status'] == 'context_limit' + assert len(reader.records[0]['undelivered_result']['content']) == 20000 + + +def test_undelivered_source_is_not_restored_as_read(trading): + from alphapulse.news_research.tools import Research + reader = Research() + 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 diff --git a/backend/tests/test_news_research_workspace.py b/backend/tests/test_news_research_workspace.py index 6d99661..19e8dd4 100644 --- a/backend/tests/test_news_research_workspace.py +++ b/backend/tests/test_news_research_workspace.py @@ -221,7 +221,7 @@ def test_retry_refreshes_local_search_with_current_vocabulary( session = AsyncMock() session.__aenter__.return_value = session session.execute.return_value = [ - (N(id=2, news_id=12, content_hash="current", keywords=[]), "current") + (N(id=2, news_id=12, content_hash="current", keywords=[]), "current", "Fresh news") ] session.scalars.return_value = [N(id=2, title="Fresh event", tags=["binance"])] monkeypatch.setattr(service, "SessionLocal", lambda: session)