diff --git a/backend/app/api/copilot_context.py b/backend/app/api/copilot_context.py index 15b7103e..6c8e807e 100644 --- a/backend/app/api/copilot_context.py +++ b/backend/app/api/copilot_context.py @@ -68,7 +68,7 @@ async def look_service(sources: list[Any], *, spec: ModelSpec | None, graph: Map kinds: tuple[str, ...] = ("table", "capability"), answer: bool = False) -> LookService: """按入口组装这一轮的查看服务。读目录、读能力登记都用各自的短会话，开流前后都能调。 - spec 是助手的模型规格：挑选模型第一次要用时才按它拿（关思考、额度小）。graph 是现有工作流，概览里列出它查过的表。 + spec 是助手的模型规格：挑选模型第一次要用时才按它拿（关思考、额度按 look/pick.PICK_MAX_TOKENS）。graph 是现有工作流，概览里列出它查过的表。 answer：问数据（意图为回答），能力只列只读的。 """ picker = Picker(spec) diff --git a/backend/app/look/base.py b/backend/app/look/base.py index 4b843595..84604fec 100644 --- a/backend/app/look/base.py +++ b/backend/app/look/base.py @@ -24,7 +24,9 @@ REPAIR_LOOK_LIMIT = 1 # 自查修正阶段另给几次 LOOK_RESULT_CHARS = 12_000 # 一次查看结果的字数上限 PICK_LIMIT = 12 # 挑选最多挑几个；候选不超过这个数时不挑、全给（沿用） CONTEXT_TABLE_LIMIT = 16 # 一次查看最多给几张表的详情（含中间表，沿用） -PICK_TIMEOUT_S = 20 # 挑选时限（沿用） +#: 挑选时限。经 OpenAI 兼容接入的推理模型关不掉思考（look/pick.PICK_MAX_TOKENS），先想再写：原来的 20 秒常常想到一半就 +#: 掐了，交回目录让主模型自己再看一遍反而更慢 +PICK_TIMEOUT_S = 45 class LookTarget(TypedDict, total=False): diff --git a/backend/app/look/pick.py b/backend/app/look/pick.py index 05c561b3..91074057 100644 --- a/backend/app/look/pick.py +++ b/backend/app/look/pick.py @@ -1,7 +1,8 @@ """按需求挑选：表、能力、工作流版本共用同一份挑选提示词（L3），每种对象只给自己的目录。 -挑选是一次分类：同一个助手模型，关掉思考、额度取小、温度 0，PICK_TIMEOUT_S 时限。只在主模型写了 look、只给了需求、 -候选又超过 PICK_LIMIT 时才调；失败（拿不到模型、超时、报错、回复认不出）不挡生成，由查看服务交回目录、让主模型点名。 +挑选是一次分类：同一个助手模型，关掉思考、温度 0，PICK_TIMEOUT_S 时限。只在主模型写了 look、只给了需求、 +候选又超过 PICK_LIMIT 时才调；失败（拿不到模型、超时、报错、被截断、回复认不出）不挡生成，由查看服务交回目录、让主模型 +点名。没成的原因照实分开说：到了额度上限被截断、回复是空的、写了字却不是约定的格式，不混成一句「格式不对」。 演示模型挑不了：pick_model 照实交回原因，不拿它乱挑。 """ @@ -9,14 +10,31 @@ from __future__ import annotations import asyncio import json +import logging import math +import re from typing import Any from app.look.base import PICK_LIMIT, PICK_TIMEOUT_S +from app.providers.budget import BudgetFallback, budget_refused, cut_off, cut_words, truncated_error from app.providers.factory import ModelSpec, get_chat_model -#: 挑选调用的输出额度：十几个名字的 JSON 用不了几百 token -PICK_MAX_TOKENS = 1024 +logger = logging.getLogger(__name__) + +#: 挑选调用的输出额度。挑出的十几个名字本身用不了几百 token，可经 OpenAI 兼容接入的推理模型（DeepSeek、GLM、Kimi 这一类） +#: 收不到 thinking=off，思考也算在额度里：原来的 1024 常常想完就没额度写结果，回复是空的或 JSON 只写了一半，被当成 +#: 「不是约定的格式」，每次都把目录交回主模型。和证据裁判、核查（engine/judge.MAX_TOKENS）同一档 +PICK_MAX_TOKENS = 16384 +#: 服务方嫌 PICK_MAX_TOKENS 太大（400 / 422，或 anthropic SDK 在本地就拒了）时退回这个。被嫌大过的按 (接入, 模型, 额度) +#: 记在进程里（providers/budget.BudgetFallback），和裁判、核查的记法相同、互相认 +PICK_FALLBACK_MAX_TOKENS = 4096 +#: 从正文里找 JSON 时最多试几个起点（每个 { 或 [ 是一个） +_MAX_STARTS = 200 +#: 有的网关把思考写在正文里，包在 里：先去掉，不然思考里打的草稿会被当成结果 +_THINK = re.compile(r".*?", re.S) +#: 挑选没成的原因（look 事件的 reason，也交回给模型） +REASON_FORMAT = "模型的回复不是约定的格式，无法识别挑出的对象" +REASON_EMPTY = "模型没有写出挑选结果（回复是空的）" PICK_SYSTEM_PROMPT = ( "你在帮工作流助手挑选它要查看的对象。用户这一轮提了一个需求，助手接下来要据此搭工作流；对象太多，没法每个都把" @@ -62,7 +80,7 @@ async def pick_model(spec: ModelSpec) -> tuple[Any | None, str | None]: """挑选用的模型：(模型, None)，拿不到时 (None, 原因)。spec 是助手的模型规格（copilot_model_spec）。 在流里拿：请求的会话那时已经不能用了，这里另开一个短会话读接入设置。没配接入时 get_chat_model 静默退回演示模型， - 演示模型挑不了，直接按挑选失败处理。 + 演示模型挑不了，直接按挑选失败处理。额度按 PICK_MAX_TOKENS，服务方嫌大时退回 PICK_FALLBACK_MAX_TOKENS（要退回时才建）。 """ from app.core.errors import explain from app.db.base import SessionLocal @@ -72,29 +90,50 @@ async def pick_model(spec: ModelSpec) -> tuple[Any | None, str | None]: "timeout": math.ceil(PICK_TIMEOUT_S)}) try: async with SessionLocal() as session: - model, _ = await get_chat_model(session, spec) + model, model_id = await get_chat_model(session, spec) except Exception as e: # noqa: BLE001 - 拿不到模型只是挑不了，生成照常 return None, f"助手使用的模型不可用：{explain(e)[0]}" if isinstance(model, MockChatModel): return None, "演示模型不能按需求挑选" - return model, None + + async def build() -> Any: + async with SessionLocal() as session: + return (await get_chat_model(session, spec.model_copy(update={"max_tokens": PICK_FALLBACK_MAX_TOKENS})))[0] + + return BudgetFallback(model, build=build, limits=(PICK_MAX_TOKENS, PICK_FALLBACK_MAX_TOKENS), + rejected=budget_refused, key=(spec.provider, model_id, PICK_MAX_TOKENS)), None def picked(raw: Any) -> list[tuple[str | None, str]] | None: """模型的回复 → [(组名或 None, 名字)]。结构化输出的对象、正文里的 JSON、字符串列表都认；键名认 items / tables、 - name / table（以前挑表时的写法）。对不上格式返回 None。""" + name / table（以前挑表时的写法）。对不上格式返回 None。 + + 正文里的 JSON 前后可以有说明、围栏，前面可以有 包着的思考：逐个起点试，第一个对得上形状的算数—— + {"items": [...]} 一类的对象（空列表也算，是「和需求无关」），或者至少有一个名字的列表（正文里顺手写的 [1] 不算）。""" if hasattr(raw, "model_dump"): raw = raw.model_dump() - if isinstance(raw, str): - text = raw.strip() - start = min((i for i in (text.find("{"), text.find("[")) if i >= 0), default=-1) - if start < 0: - return None - end = text.rfind("}" if text[start] == "{" else "]") + if not isinstance(raw, str): + return _items(raw) + text = _THINK.sub("", raw) + decoder = json.JSONDecoder() + starts = 0 + for i, ch in enumerate(text): + if ch not in "{[": + continue + starts += 1 + if starts > _MAX_STARTS: + break try: - raw = json.loads(text[start:end + 1]) + value, _ = decoder.raw_decode(text, i) except ValueError: - return None + continue + got = _items(value) + if got is not None and (got or isinstance(value, dict)): + return got + return None + + +def _items(raw: Any) -> list[tuple[str | None, str]] | None: if isinstance(raw, dict): raw = raw.get("items", raw.get("tables")) if not isinstance(raw, list): @@ -111,23 +150,61 @@ def picked(raw: Any) -> list[tuple[str | None, str]] | None: return out -async def _ask(model: Any, messages: list[tuple[str, str]]) -> list[tuple[str | None, str]]: - """问一次。先走结构化输出；不支持、报错、或给回来的不是约定的结构，再退回纯文本、从正文里取 JSON。 +def _unpack(out: Any) -> tuple[Any, Any]: + """with_structured_output(include_raw=True) 交回 {raw, parsed, parsing_error}；测试的假模型直接交回解析好的对象。""" + if isinstance(out, dict) and "raw" in out and ("parsed" in out or "parsing_error" in out): + return out.get("raw"), out.get("parsed") + return None, out - 结构化输出报错也退回（同 catalog._ask_model）：有的网关不支持工具调用，一调就报错，纯文本却能用。 + +def _in_raw(raw: Any) -> list[Any]: + """结构化输出没解析出来时，原始回复里还可能有结果的地方：工具调用的参数、正文。""" + from app.engine.state import message_text + + if raw is None: + return [] + found: list[Any] = [c.get("args") for c in getattr(raw, "tool_calls", None) or [] if isinstance(c, dict)] + found += [c.get("args") for c in getattr(raw, "invalid_tool_calls", None) or [] if isinstance(c, dict)] + return [*found, message_text(raw)] + + +def _cut(model: Any) -> str: + return f"{cut_words('按需求挑选的模型', getattr(model, 'max_tokens', None))}，没写完挑选结果" + + +async def _ask(model: Any, messages: list[tuple[str, str]]) -> list[tuple[str | None, str]]: + """问一次。先走结构化输出（带原始回复）：解析出来的不对，再从原始回复的工具调用参数、正文里找；不支持结构化输出 + （有的网关不支持工具调用，一调就报错，纯文本却能用，同 catalog._ask_model）或都没找到，再退回纯文本、从正文里找。 + + 到了额度上限被截断的照实说是截断（同证据裁判 engine/judge._ask）：结构化调用截断的不再退回纯文本——思考占满额度时 + 纯文本一样截断，白花一次额度；写了一半的清单也不采信，分不清最后一个名字写完了没有。 """ from app.engine.state import message_text try: - parsed = picked(await model.with_structured_output(PICK_SCHEMA).ainvoke(messages)) - except Exception: # noqa: BLE001 - 退回纯文本 - parsed = None - if parsed is not None: - return parsed - parsed = picked(message_text(await model.ainvoke(messages))) - if parsed is None: - raise PickFailed("模型的回复不是约定的格式，无法识别挑出的对象") - return parsed + raw, parsed = _unpack(await model.with_structured_output(PICK_SCHEMA, include_raw=True).ainvoke(messages)) + except Exception as e: # noqa: BLE001 - 退回纯文本 + if truncated_error(e): + raise PickFailed(_cut(model)) from e + else: + if cut_off(raw): + raise PickFailed(_cut(model)) + for found in (parsed, *_in_raw(raw)): + if (got := picked(found)) is not None: + return got + try: + reply = await model.ainvoke(messages) + except Exception as e: + if truncated_error(e): + raise PickFailed(_cut(model)) from e + raise + if cut_off(reply): + raise PickFailed(_cut(model)) + text = message_text(reply) + if (got := picked(text)) is not None: + return got + logger.warning("按需求挑选：回复认不出（正文 %d 字，开头：%r）", len(text), text[:200]) + raise PickFailed(REASON_FORMAT if text.strip() else REASON_EMPTY) class Picker: diff --git a/backend/tests/test_copilot_table_pick.py b/backend/tests/test_copilot_table_pick.py index d8b39e55..f34910b7 100644 --- a/backend/tests/test_copilot_table_pick.py +++ b/backend/tests/test_copilot_table_pick.py @@ -19,6 +19,7 @@ import uuid import pytest from httpx import ASGITransport, AsyncClient +from langchain_core.messages import AIMessage from app.data import catalog, schema_view from app.data.catalog import INFERRED_MARK, make_item @@ -26,7 +27,8 @@ from app.data.introspect import introspect from app.db.base import SessionLocal from app.db.models import DataSource from app.look.base import CONTEXT_TABLE_LIMIT, LookBudget -from app.look.pick import Picker +from app.look import pick as pick_module +from app.look.pick import REASON_EMPTY, REASON_FORMAT, Picker from app.look.service import LookService, parse_targets from app.look.tables import TableKind from app.main import app @@ -195,7 +197,7 @@ async def test_named_look_brings_in_the_bridge_without_the_picker(monkeypatch, s assert block["total"] == scenic.TABLE_COUNT + len(scenic.VIEWS) and block["catalog_tables"] == 0 -async def test_need_only_look_picks_with_a_small_fast_model(monkeypatch, scenic_source): +async def test_need_only_look_picks_with_thinking_off_and_room_to_think(monkeypatch, scenic_source): fake = FakePicker(PICKED) specs = _use_picker(monkeypatch, fake) looks = await _service([scenic_source]) @@ -205,7 +207,8 @@ async def test_need_only_look_picks_with_a_small_fast_model(monkeypatch, scenic_ # 挑选的输入是需求加紧凑的表目录（每张表一行），不是全部字段 assert NEED in fake.calls[0] and "channel_visits" in fake.calls[0] and "attributed_share" not in fake.calls[0] (spec,) = specs - assert spec.thinking == "off" and spec.max_tokens <= 2048 + # 经 OpenAI 兼容接入的推理模型收不到 thinking=off：额度要留出思考的余地，不然想完就没额度写结果 + assert spec.thinking == "off" and spec.max_tokens == pick_module.PICK_MAX_TOKENS >= 16384 (block,) = result.record["objects"] assert (block["named"], block["picked"], block["joined"]) == ([], ["visits", "channels"], ["channel_visits"]) assert f"按需求从 {scenic.TABLE_COUNT + len(scenic.VIEWS)} 张表中挑出 2 张，另为连接补上 1 张" in result.text @@ -245,6 +248,108 @@ async def test_pick_failures_hand_back_the_index(monkeypatch, scenic_source, fak assert "点名" in result.text +class ScriptedPicker: + """结构化输出和纯文本各回各的：structured 是 include_raw 的回包（或要抛的异常），plain 是纯文本调用的回复（或异常）。 + calls 按顺序记下调了哪一种。""" + + def __init__(self, *, structured, plain=None) -> None: + self.structured, self.plain = structured, plain + self.calls: list[str] = [] + + def with_structured_output(self, schema, **kwargs): + assert schema["title"] and kwargs.get("include_raw") + outer = self + + class _Structured: + async def ainvoke(self, _messages): + outer.calls.append("structured") + if isinstance(outer.structured, Exception): + raise outer.structured + return outer.structured + + return _Structured() + + async def ainvoke(self, _messages): + self.calls.append("plain") + if isinstance(self.plain, Exception): + raise self.plain + return self.plain + + +def _raw(content: str = "", **meta) -> dict: + return {"raw": AIMessage(content=content, response_metadata=meta), "parsed": None, "parsing_error": None} + + +async def _look(monkeypatch, source, fake) -> dict: + _use_picker(monkeypatch, fake) + looks = await _service([source]) + return (await looks.look({"op": "look", "need": NEED}, LookBudget())).record + + +async def test_a_pick_cut_off_at_the_budget_says_so(monkeypatch, scenic_source): + """思考占满额度、结构化调用停在上限：照实说被截断（带额度），不当成「格式不对」，也不再白调一次纯文本。""" + fake = ScriptedPicker(structured=_raw('{"items": [{"name": "vis', finish_reason="length"), + plain=AssertionError("截断后不该再调纯文本")) + record = await _look(monkeypatch, scenic_source, fake) + assert fake.calls == ["structured"] + assert record["status"] == "index" + assert record["reason"] == f"按需求挑选的模型的输出达到上限（{pick_module.PICK_MAX_TOKENS} token）被截断，没写完挑选结果" + + +async def test_a_plain_reply_with_thinking_and_prose_is_still_read(monkeypatch, scenic_source): + """网关不支持结构化输出、正文里先是 包着的思考（里面打了个草稿），再是一句说明和围栏里的 JSON：认围栏里的。""" + text = ('先打个草稿 {"items": [{"name": "stores"}]}\n好的，挑出这些：\n```json\n' + '{"items": [{"name": "visits"}, {"name": "channels"}]}\n```') + fake = ScriptedPicker(structured=NotImplementedError("不支持工具调用"), plain=AIMessage(content=text)) + record = await _look(monkeypatch, scenic_source, fake) + assert fake.calls == ["structured", "plain"] + assert record["status"] == "ok" and record["objects"][0]["picked"] == ["visits", "channels"] + + +async def test_structured_reply_written_in_the_text_is_read_without_a_second_call(monkeypatch, scenic_source): + """网关收了结构化输出的参数却把 JSON 写在正文里：从原始回复里认出来，不再调纯文本。""" + fake = ScriptedPicker(structured=_raw('结果：{"items": [{"name": "visits"}]}', finish_reason="stop"), + plain=AssertionError("原始回复里已经有了")) + record = await _look(monkeypatch, scenic_source, fake) + assert fake.calls == ["structured"] and record["objects"][0]["picked"] == ["visits"] + + +@pytest.mark.parametrize(("plain", "reason"), [ + (AIMessage(content=""), REASON_EMPTY), + (AIMessage(content="这几张都要看，见第[1]条"), REASON_FORMAT), +], ids=["empty", "prose"]) +async def test_empty_and_prose_replies_are_told_apart(monkeypatch, scenic_source, plain, reason): + """空回复和写了字却不是约定格式的分开说；正文里顺手写的 [1] 不算一个空的挑选结果。""" + fake = ScriptedPicker(structured=_raw("", finish_reason="stop"), plain=plain) + record = await _look(monkeypatch, scenic_source, fake) + assert record["status"] == "index" and record["reason"] == reason + + +async def test_a_budget_the_service_rejects_falls_back_to_a_smaller_one(monkeypatch, scenic_source): + """服务方一开口就嫌额度太大：按 PICK_FALLBACK_MAX_TOKENS 重建再问，挑选照常成。""" + import app.look.pick as pick + from app.providers import budget + + monkeypatch.setattr(budget, "_REJECTED", set()) + + class TooBig(Exception): + status_code = 400 + + big = FakePicker(error=TooBig("max_tokens: 16384 > 8192, which is the maximum allowed")) + small = FakePicker(PICKED) + specs: list = [] + + async def _model(_session, spec): + specs.append(spec) + return (big if spec.max_tokens == pick.PICK_MAX_TOKENS else small), "fake-pick-small-gateway" + + monkeypatch.setattr(pick, "get_chat_model", _model) + looks = await _service([scenic_source]) + record = (await looks.look({"op": "look", "need": NEED}, LookBudget())).record + assert [s.max_tokens for s in specs] == [pick.PICK_MAX_TOKENS, pick.PICK_FALLBACK_MAX_TOKENS] + assert record["status"] == "ok" and record["objects"][0]["picked"] == ["visits", "channels"] + + async def test_an_empty_pick_is_reported_as_such(monkeypatch, scenic_source): """挑选模型照提示说「需求和这份目录无关」返回空：照实说没挑出来，不当成失败、不塞表。""" _use_picker(monkeypatch, FakePicker({"items": []}))