按需求挑选不再总是「没挑出来，已把表目录交给助手」：挑选以为能关掉思考、只给 1024 的输出额度，可经 OpenAI 兼容接入的推理模型（DeepSeek、GLM、Kimi 这一类）收不到 thinking=off，想完就没额度写结果，回复是空的或 JSON 只写了一半，又被报成「不是约定的格式」。额度改成和证据裁判、核查同一档 16384，服务方嫌大时退回 4096（按接入和模型记在进程里、互相认），时限 20 秒放宽到 45 秒；结构化输出带原始回复，正文、工具调用参数里的 JSON 也认，<think> 里的草稿不算，正文里顺手写的 [1] 不算空结果；没成的原因照实分开说：被截断（带额度）、回复是空的、写了字却不是约定的格式，认不出的回复开头记进日志

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)

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):

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> 里：先去掉，不然思考里打的草稿会被当成结果
+_THINK = re.compile(r"<think>.*?</think>", 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 前后可以有说明、围栏，前面可以有 <think> 包着的思考：逐个起点试，第一个对得上形状的算数——
+    {"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:

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):
+    """网关不支持结构化输出、正文里先是 <think> 包着的思考（里面打了个草稿），再是一句说明和围栏里的 JSON：认围栏里的。"""
+    text = ('<think>先打个草稿 {"items": [{"name": "stores"}]}</think>\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": []}))