feat：完成数据文件上传解析及测试

app/agent/data_tools.py

@@ -1,5 +1,7 @@
 """数据文件的本地工具：describe_data 查看列概况，query_data 执行只读 SQL；问答和 screen agent 共用
 
+apply_actions 里的 $query 也通过 DataContext.query 执行，见 tools.py
+
 工具只能看到 DataContext 里的表，也就是当前用户在本会话里未过期的文件，查询引擎也只为这些表建视图
 工具通过 runtime.context.data 读取 DataContext，用到它们的 agent 上下文都要有 data 字段；
 上下文类型因 agent 而异，所以写 ToolRuntime[Any]（不写类型参数时按 None 处理，序列化会告警）
@@ -14,7 +16,7 @@ from langchain_core.messages import ToolMessage
 from pydantic import Field
 
 from app.config import settings
-from app.datafiles.query import QueryError, run_query
+from app.datafiles.query import MAX_ROWS, QueryError, QueryResult, run_query
 from app.schemas.data_file import RequestDataFile, RequestDataTable
 
 # 列超过这么多时，不指定 columns 只返回列名和类型，避免一次塞进太多概况
@@ -39,6 +41,22 @@ class DataContext:
             for table in file.tables
         }
 
+    def query(self, sql: str, *, max_rows: int = MAX_ROWS) -> QueryResult:
+        """执行一条只读 SQL；失败时抛 QueryError，原因可以直接返回给模型修正"""
+        paths = self.paths()
+        # 任务执行期间被删除或清理的文件跳过，不影响查询其他表
+        existing = {name: path for name, path in paths.items() if path.exists()}
+        try:
+            return run_query(
+                sql, existing, timeout=settings.data_query_timeout, max_rows=max_rows
+            )
+        except QueryError as error:
+            if gone := [name for name in paths if name not in existing]:
+                raise QueryError(
+                    f"{error}；{'、'.join(gone)} 的文件已删除或过期，不能再查询"
+                ) from None
+            raise
+
 
 def _error(runtime: ToolRuntime, message: str) -> ToolMessage:
     # 返回给模型修正，不中断 agent
@@ -111,16 +129,10 @@ def query_data(sql: str, runtime: ToolRuntime[Any]) -> dict[str, Any] | ToolMess
     回答和图表里的数字以查询结果为准，不要自己心算
     """
     data: DataContext = runtime.context.data
-    paths = data.paths()
-    # 任务执行期间被删除或清理的文件跳过，不影响查询其他表
-    existing = {name: path for name, path in paths.items() if path.exists()}
     try:
-        return run_query(sql, existing, timeout=settings.data_query_timeout).to_dict()
+        return data.query(sql).to_dict()
     except QueryError as error:
-        message = str(error)
-        if gone := [name for name in paths if name not in existing]:
-            message += f"；{'、'.join(gone)} 的文件已删除或过期，不能再查询"
-        return _error(runtime, message)
+        return _error(runtime, str(error))
 
 
 DATA_TOOLS = [describe_data, query_data]

app/agent/qa_agent.py

@@ -0,0 +1,50 @@
+"""问答 agent：回答关于大屏和上传数据的问题
+
+会话里有数据文件时挂 describe_data、query_data，先查询再回答；没有文件时不挂工具，只调用一次模型
+"""
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from langchain.agents import create_agent
+from langchain.agents.middleware import ModelCallLimitMiddleware
+
+from app.agent.data_tools import DATA_TOOLS, DataContext
+from app.clients.llm import get_llm
+from app.config import settings
+from app.prompt.qa import QA_DATA_RULES, QA_SYSTEM_PROMPT
+
+
+@dataclass(frozen=True)
+class QAContext:
+    data: DataContext = field(default_factory=DataContext)  # 本会话可用的数据表
+
+
+# 是否挂数据工具 → agent
+_agents: dict[bool, Any] = {}
+
+
+def get_qa_agent(with_data: bool = False):
+    """第一次使用时组装，之后复用；没有数据文件的请求看不到数据工具和相关提示"""
+    if with_data not in _agents:
+        _agents[with_data] = create_agent(
+            model=get_llm(),
+            tools=DATA_TOOLS if with_data else [],
+            system_prompt=(
+                f"{QA_SYSTEM_PROMPT}\n{QA_DATA_RULES}"
+                if with_data
+                else QA_SYSTEM_PROMPT
+            ),
+            context_schema=QAContext,
+            middleware=[
+                # 超过上限时抛 ModelCallLimitExceededError，由节点按失败处理；
+                # 写成 end 时中间件只插入一条英文说明，节点分不清是正常结束还是用尽了次数
+                ModelCallLimitMiddleware(
+                    run_limit=settings.max_qa_steps, exit_behavior="error"
+                ),
+            ],
+            # 不继承外层图的检查点：会话历史只保存回答文本，不保存查询过程
+            checkpointer=False,
+            name="qa_agent",
+        )
+    return _agents[with_data]

app/agent/query_values.py

@@ -0,0 +1,171 @@
+"""apply_actions 里的 $query：把字段值写成 {"$query": "SELECT ..."}，提交时替换成上传数据的查询结果
+
+模型不用把查询结果逐行抄进动作，又慢又容易抄错；替换发生在 run_batch 之前，替换后照常做字段校验
+结果是对象数组，键是查询结果的列名，正好对应图表物料的 dataset.source；series 的 encode 按列名引用
+"""
+
+from dataclasses import dataclass
+from typing import Any
+
+from pydantic import JsonValue
+
+from app.actions.batch import BatchError
+from app.agent.data_tools import DataContext
+from app.config import settings
+from app.schemas.actions import AddAction, AgentAction, UpdateAction
+from app.schemas.screen import Screen
+
+QUERY_KEY = "$query"
+
+
+@dataclass(frozen=True)
+class EmbeddedQuery:
+    """一处替换：第几个动作、写到哪个字段、结果的列和行数，写进 apply_actions 的结果供模型核对"""
+
+    index: int
+    node_id: str
+    path: str
+    columns: list[str]
+    rows: int
+
+
+def fill_queries(
+    actions: list[AgentAction], data: DataContext
+) -> tuple[list[AgentAction], list[EmbeddedQuery]]:
+    """替换 add 的 node 和 update 的 value 里所有的 $query；查询失败抛 BatchError，整批不生效
+
+    add 用了 $query 又没写 sample_data 时设为 false：数据来自用户上传的文件，不是示例数据
+    """
+    filled: list[AgentAction] = []
+    embedded: list[EmbeddedQuery] = []
+    for index, action in enumerate(actions):
+        found: list[tuple[str, list[str], int]] = []
+        node_id = ""  # 只有 add 和 update 有字段值，其他动作不会有替换
+        try:
+            if isinstance(action, AddAction):
+                node = _fill(
+                    action.node.model_dump(mode="json", exclude_unset=True),
+                    data,
+                    "",
+                    found,
+                )
+                if found:
+                    fields = action.model_dump(mode="json", exclude_unset=True)
+                    fields["node"] = node
+                    fields.setdefault("sample_data", False)
+                    action = AddAction.model_validate(fields)
+                node_id = action.node.id
+            elif isinstance(action, UpdateAction):
+                value = _fill(action.value, data, action.field_path, found)
+                if found:
+                    action = action.model_copy(update={"value": value})
+                node_id = action.target_id
+        except ValueError as error:  # 含 QueryError
+            raise BatchError(index, action, str(error)) from error
+        embedded.extend(
+            EmbeddedQuery(index, node_id, path, columns, rows)
+            for path, columns, rows in found
+        )
+        filled.append(action)
+    return filled, embedded
+
+
+def _fill(
+    value: JsonValue,
+    data: DataContext,
+    path: str,
+    found: list[tuple[str, list[str], int]],
+) -> JsonValue:
+    if isinstance(value, dict):
+        if QUERY_KEY in value:
+            return _query(value, data, path, found)
+        return {
+            key: _fill(item, data, _join(path, key), found)
+            for key, item in value.items()
+        }
+    if isinstance(value, list):
+        return [
+            _fill(item, data, _join(path, position), found)
+            for position, item in enumerate(value)
+        ]
+    return value
+
+
+def _query(
+    value: dict[str, Any],
+    data: DataContext,
+    path: str,
+    found: list[tuple[str, list[str], int]],
+) -> list[dict[str, Any]]:
+    sql = value[QUERY_KEY]
+    if len(value) != 1 or not isinstance(sql, str):
+        raise ValueError(
+            f'{path} 的 $query 要写成 {{"$query": "SELECT ..."}}，对象里不能有其他键'
+        )
+    if not data.files:
+        raise ValueError("会话里没有上传的数据文件，不能使用 $query")
+    limit = settings.data_embed_max_rows
+    result = data.query(sql, max_rows=limit)
+    if result.truncated:
+        raise ValueError(
+            f"{path} 的 $query 结果超过 {limit} 行，组件放不下；在 SQL 里聚合、筛选或 LIMIT"
+        )
+    if duplicated := sorted(
+        {name for name in result.columns if result.columns.count(name) > 1}
+    ):
+        raise ValueError(
+            f"{path} 的 $query 结果有重名的列 {'、'.join(duplicated)}，用 AS 起不同的别名"
+        )
+    found.append((path, result.columns, len(result.rows)))
+    return [dict(zip(result.columns, row, strict=True)) for row in result.rows]
+
+
+def _join(path: str, key: str | int) -> str:
+    return f"{path}.{key}" if path else str(key)
+
+
+def encode_problems(screen: Screen, node_ids: set[str]) -> list[str]:
+    """用了 $query 的组件，series 的 encode 引用的列必须在 dataset.source 里
+
+    数据换成查询结果后列名变了，encode 还引用物料默认的列名时图表是空的，前端也不报错，所以在这里拦下
+    """
+    problems = []
+    for node in screen.nodes:
+        if node.id not in node_ids:
+            continue
+        option = node.props.get("option")
+        if not isinstance(option, dict):
+            continue
+        dataset = option.get("dataset")
+        source = dataset.get("source") if isinstance(dataset, dict) else None
+        # 空结果或不是对象数组时无从核对
+        if not (
+            isinstance(source, list)
+            and source
+            and all(isinstance(row, dict) for row in source)
+        ):
+            continue
+        columns = list(dict.fromkeys(key for row in source for key in row))
+        series = option.get("series")
+        items = series if isinstance(series, list) else [series]
+        wrong = []
+        for position, item in enumerate(items):
+            encode = item.get("encode") if isinstance(item, dict) else None
+            if not isinstance(encode, dict):
+                continue
+            for key, names in encode.items():
+                missing = [
+                    name
+                    for name in (names if isinstance(names, list) else [names])
+                    if isinstance(name, str) and name not in columns
+                ]
+                if missing:
+                    wrong.append(
+                        f"series.{position}.encode.{key} 用了 {'、'.join(missing)}"
+                    )
+        if wrong:
+            problems.append(
+                f"组件 {node.id}：{'；'.join(wrong)}，但 dataset.source 的列是 {'、'.join(columns)}；"
+                "把 encode 改成查询结果的列名，或在 SQL 里用 AS 起对应的别名"
+            )
+    return problems

app/agent/tools.py

@@ -3,6 +3,7 @@
 它们读写的是本次请求的大屏和 agent 状态，所以放在本地；物料等能力由 MCP 提供
 """
 
+from collections.abc import Sequence
 from dataclasses import dataclass, field
 from typing import Annotated, NotRequired
 
@@ -14,6 +15,7 @@ from pydantic import Field
 
 from app.actions.batch import BatchError, run_batch
 from app.agent.data_tools import DataContext
+from app.agent.query_values import EmbeddedQuery, encode_problems, fill_queries
 from app.schemas.actions import Action, AddAction, AgentAction
 from app.schemas.component_resource import ComponentCatalog
 from app.schemas.screen import Screen
@@ -58,6 +60,7 @@ def _result_text(
     after: Screen,
     actions: list[Action],
     undeclared_sample: frozenset[str] = frozenset(),
+    queries: Sequence[EmbeddedQuery] = (),
 ) -> str:
     """执行结果：除了动作数量，还给出执行后的画布和每个组件的布局
 
@@ -75,10 +78,19 @@ def _result_text(
     if guessed := [item for item in sample if item in undeclared_sample]:
         lines.append(
             f"未写 sample_data，按物料默认标成了示例数据：{'、'.join(guessed)}；"
-            "如果这些组件展示的数据都来自大屏上已有的组件，把 sample_data 设为 false 重新提交整批，否则不用处理"
+            "如果这些组件展示的数据都来自大屏上已有的组件或上传的数据文件，把 sample_data 设为 false 重新提交整批，否则不用处理"
         )
     if declared := [item for item in sample if item not in undeclared_sample]:
         lines.append(f"标为示例数据的新增组件：{'、'.join(declared)}")
+    # 模型提交时只写了 SQL，看不到结果；给出列名和行数，用来核对 encode 和数据量
+    if queries:
+        lines.append("$query 已替换为查询结果：")
+        for item in queries:
+            note = "（结果为空，组件会没有数据）" if not item.rows else ""
+            lines.append(
+                f"- 第 {item.index + 1} 个动作 {item.node_id} 的 {item.path}："
+                f"{item.rows} 行{note}，列 {'、'.join(item.columns)}"
+            )
 
     old_layouts = {node.id: node.layout for node in before.nodes}
     lines.append(
@@ -114,11 +126,17 @@ def apply_actions(
     失败时整批不生效，返回带动作序号的错误，修正后重新提交整批；以最后一次调用的结果为准
     """
     try:
+        actions, queries = fill_queries(actions, runtime.context.data)
         completed, after = run_batch(
             runtime.context.screen, runtime.context.catalog, actions
         )
+        problems = encode_problems(after, {item.node_id for item in queries})
     except BatchError as error:
-        message = f"执行失败，整批未生效：\n- {error}"
+        problems = [str(error)]
+    if problems:
+        message = "执行失败，整批未生效：\n" + "\n".join(
+            f"- {item}" for item in problems
+        )
         return Command(
             update={
                 "actions": None,
@@ -155,7 +173,9 @@ def apply_actions(
             "apply_error": None,
             "messages": [
                 ToolMessage(
-                    _result_text(runtime.context.screen, after, completed, undeclared),
+                    _result_text(
+                        runtime.context.screen, after, completed, undeclared, queries
+                    ),
                     tool_call_id=runtime.tool_call_id,
                 )
             ],

app/config.py

@@ -59,6 +59,10 @@ class Settings:
     max_agent_steps: int = field(
         default_factory=lambda: int(os.getenv("MAX_AGENT_STEPS", "12"))
     )
+    # 问答 agent 单次运行的模型调用上限；查询数据时每次查询多一次调用
+    max_qa_steps: int = field(
+        default_factory=lambda: int(os.getenv("MAX_QA_STEPS", "8"))
+    )
     # 单次 MCP 工具调用的超时（秒），服务列表见 mcp_servers.json
     mcp_tool_timeout: float = field(
         default_factory=lambda: float(os.getenv("MCP_TOOL_TIMEOUT", "5"))
@@ -109,6 +113,11 @@ class Settings:
         default_factory=lambda: float(os.getenv("DATA_QUERY_TIMEOUT", "10"))
     )
 
+    # apply_actions 里一个 $query 写进组件的最多行数，超过时报错，让模型在 SQL 里聚合
+    data_embed_max_rows: int = field(
+        default_factory=lambda: int(os.getenv("DATA_EMBED_MAX_ROWS", "1000"))
+    )
+
     # 用户中心：请求携带的 token 交给它校验，换回 user_id
     user_center_url: str = field(
         default_factory=lambda: (
@@ -145,6 +154,8 @@ class Settings:
             raise ValueError("HISTORY_MAX_TOKENS 必须是正整数")
         if self.max_agent_steps < 1:
             raise ValueError("MAX_AGENT_STEPS 必须是正整数")
+        if self.max_qa_steps < 1:
+            raise ValueError("MAX_QA_STEPS 必须是正整数")
         if self.mcp_tool_timeout <= 0:
             raise ValueError("MCP_TOOL_TIMEOUT 必须是正数")
         if self.data_file_max_mb < 1:
@@ -153,6 +164,8 @@ class Settings:
             raise ValueError("DATA_FILE_TTL 必须是正整数")
         if self.data_query_timeout <= 0:
             raise ValueError("DATA_QUERY_TIMEOUT 必须是正数")
+        if self.data_embed_max_rows < 1:
+            raise ValueError("DATA_EMBED_MAX_ROWS 必须是正整数")
         if not 0 <= self.intent_qa_min_probability <= 1:
             raise ValueError("INTENT_QA_MIN_PROBABILITY 必须在 0 到 1 之间")
 

app/processor/nodes/qa_node.py

@@ -1,14 +1,24 @@
-"""基于当前大屏和会话历史流式回答问题"""
+"""基于当前大屏、上传的数据和会话历史流式回答问题
 
-from langchain_core.messages import HumanMessage, SystemMessage
+由问答 agent 执行：会话里有数据文件时先查询再回答，否则只调用一次模型；只把模型输出的文字推送给前端
+"""
+
+import json
+
+from langchain.agents.middleware.model_call_limit import ModelCallLimitExceededError
+from langchain_core.messages import AIMessageChunk, HumanMessage
 from langchain_core.output_parsers import StrOutputParser
 from langgraph.config import get_stream_writer
 
-from app.clients.llm import get_llm
+from app.agent.data_tools import DataContext
+from app.agent.qa_agent import QAContext, get_qa_agent
 from app.processor.base import BaseNode
 from app.processor.history import history_messages
 from app.processor.state import QueryState
-from app.prompt.qa import QA_SYSTEM_PROMPT
+from app.processor.summary import data_file_summary
+
+# 模型在一次回答里分几次输出文字（例如查询前后各说一段）时，段落之间空一行
+PARAGRAPH_BREAK = "\n\n"
 
 
 class QANode(BaseNode):
@@ -23,23 +33,42 @@ class QANode(BaseNode):
         except RuntimeError:
             writer = None
 
+        context = state["input_context"]
+        payload = context.model_dump(
+            mode="json", exclude_unset=True, exclude={"history", "data_files"}
+        )
+        if context.data_files:
+            # 只给表名、行数和列名，文件路径和列概况不进提示词，需要时由 describe_data 查看
+            payload["data_files"] = [
+                data_file_summary(item) for item in context.data_files
+            ]
+        agent = get_qa_agent(with_data=bool(context.data_files))
+
         parser = StrOutputParser()
         parts: list[str] = []
-        context = state["input_context"]
-        async for chunk in get_llm().astream(
-            [
-                SystemMessage(content=QA_SYSTEM_PROMPT),
-                # 会话历史作为此前的对话消息，本次请求是最后一条用户消息
-                *history_messages(context.history),
-                HumanMessage(
-                    content=context.model_dump_json(
-                        exclude_unset=True, exclude={"history"}
-                    )
-                ),
-            ]
-        ):
-            # 工具调用等无文本片段会得到空串
-            if delta := parser.invoke(chunk):
+        last_message_id = None
+        try:
+            async for chunk, metadata in agent.astream(
+                {
+                    "messages": [
+                        # 会话历史作为此前的对话消息，本次请求是最后一条用户消息
+                        *history_messages(context.history),
+                        HumanMessage(content=json.dumps(payload, ensure_ascii=False)),
+                    ]
+                },
+                context=QAContext(data=DataContext(tuple(context.data_files))),
+                stream_mode="messages",
+            ):
+                # 工具结果、工具调用和中间件插入的消息都不推送
+                if metadata.get("langgraph_node") != "model" or not isinstance(
+                    chunk, AIMessageChunk
+                ):
+                    continue
+                if not (delta := parser.invoke(chunk)):
+                    continue
+                if parts and chunk.id != last_message_id:
+                    delta = PARAGRAPH_BREAK + delta
+                last_message_id = chunk.id
                 parts.append(delta)
                 if writer is not None:
                     writer(
@@ -49,6 +78,15 @@ class QANode(BaseNode):
                             "delta": delta,
                         }
                     )
+        except ModelCallLimitExceededError:
+            self.logger.warning("问答 agent 超出模型调用上限（MAX_QA_STEPS）")
+            updated_state["status"] = "failed"
+            updated_state["answer"] = (
+                "查询步骤超出上限，这次没有完成，可以缩小范围或换个问法再试"
+            )
+            return updated_state
+
+        # 推送了什么就保存什么，会话历史和前端看到的回答一致
         answer = "".join(parts).strip()
         # 没有有效文本时报错，避免将空回答标记为处理完成
         if not answer:

app/processor/summary.py

@@ -55,12 +55,12 @@ def request_summary(context: InputContext) -> JsonObject:
             for item in context.attachments
             if item.get("kind") != "data_file"
         ],
-        "data_files": [_data_file(item) for item in context.data_files],
+        "data_files": [data_file_summary(item) for item in context.data_files],
     }
 
 
-def _data_file(item: RequestDataFile) -> JsonObject:
-    """每个文件只给表名、label、行数和列名；列概况由 describe_data 按需查看"""
+def data_file_summary(item: RequestDataFile) -> JsonObject:
+    """每个文件只给表名、label、行数和列名；列概况由 describe_data 按需查看。问答节点也用它"""
     tables: list[JsonObject] = []
     for table in item.tables:
         summary: JsonObject = {

app/prompt/qa.py

@@ -18,6 +18,16 @@ QA_SYSTEM_PROMPT = f"""## 任务
 
 ## 输出与边界
 直接给出面向用户的回答，可使用 Markdown，不生成动作或将建议视为修改授权，也不承诺稍后会执行修改
-本节点只分析已提供的信息，不声称已修改大屏、运行代码、请求接口或读取外部文件
+本节点只分析已提供的信息和工具返回的结果，不声称已修改大屏、运行代码或请求接口；没有通过工具读到的文件，不声称读过
 {INPUT_AS_DATA}
 """
+
+# 会话里有数据文件时追加在系统提示末尾，同时挂上 describe_data、query_data
+QA_DATA_RULES = """## 数据文件
+请求里的 data_files 是用户在本会话上传的数据文件，每张表有表名（t1、t2……）、行数和列名
+回答和数据有关的问题时，先用 describe_data 看要用的表，再用 query_data 查询，不要只凭列名或样例下结论
+需要查询时直接调用工具，查完再写回答，不要先输出"我来查一下"之类的过渡语
+回答里的数字必须来自查询结果，不要心算或估算；说明用了哪张表、什么筛选条件
+列的含义或单位不明确时写明你的假设；查询结果被截断（truncated 为 true）时改用聚合，不要把前几行当成全部数据
+用户说的文件或表在 data_files 里找不到时，说明这一点；可能已过期或被删除，请用户重新上传
+"""

app/schemas/actions/add.py

@@ -15,10 +15,12 @@ class AddAction(ActionModel):
     node: ScreenNode = Field(
         description="新组件。id、type、layout 必填，其余只写需要覆盖物料默认配置的字段；"
         "后端按 type 找到物料，用默认配置补全：对象逐层覆盖，数组整体替换，null 明确覆盖。"
-        "id 不能与现有组件重复，同一批里先 delete 的 ID 可以再用；type 用 list_components 返回的 type"
+        "id 不能与现有组件重复，同一批里先 delete 的 ID 可以再用；type 用 list_components 返回的 type。"
+        '任何字段值都可以写成 {"$query": "SELECT ..."}，后端替换成上传数据的查询结果（对象数组，键是结果的列名）'
     )
     sample_data: bool = Field(
         default=False,
         description="true 表示组件显示的不是真实业务数据（物料示例数据，或自己编写的数据）。不写时按物料设置；"
-        "只有展示的数据全部来自大屏上已有的组件时写 false，例如更换类型时迁移了原组件的数据",
+        "只有展示的数据全部来自大屏上已有的组件或上传的数据文件时写 false，例如更换类型时迁移了原组件的数据；"
+        "node 里用了 $query 时不写也会设为 false",
     )

app/schemas/actions/update.py

@@ -21,7 +21,10 @@ class UpdateAction(BaseAction):
     path: list[PathSegment] = Field(
         description='字段路径第一段之后的各段，字符串为对象键，整数为数组下标；name 没有后续路径，传空列表。例如字段 props.option.title.text 写成 section=props、path=["option", "title", "text"]'
     )
-    value: JsonValue = Field(description="替换该字段的值，必须符合字段定义的类型和取值")
+    value: JsonValue = Field(
+        description="替换该字段的值，必须符合字段定义的类型和取值；"
+        '可以写成 {"$query": "SELECT ..."}，后端替换成上传数据的查询结果（对象数组，键是结果的列名）'
+    )
 
     @model_validator(mode="after")
     def path_matches_section(self) -> Self: