fix(upload): 上传截断显式告知+generic补文件处理

pipeline_core/upload_tools.py

@@ -0,0 +1,122 @@
+# upload_tools.py - 会话上传文件处理统一入口（agent_chat / agent_chat_generic 共用）
+#
+# 根治两个真实 bug：
+# 1. 静默截断：_extract_text 硬截 15000 字符，LLM 把截断内容当完整文件回答
+#    （实测：30031 字符文件，LLM 只看到前 ~15000，答"最后一行"答错且不自知）。
+#    修复：截断必须显式告知——注入预览 + 总字符数 + 全文获取手段（read_file 路径）。
+# 2. generic 入口完全忽略 file 字段：上传静默丢弃。修复：两个入口走同一处理。
+import os
+import re
+import zipfile
+
+PREVIEW_LIMIT = 15000   # 注入 prompt 的预览上限（read_file 单次可读 30000，全文可取）
+TEXT_EXTS = ('.txt', '.md', '.json', '.csv', '.py', '.log', '.yaml', '.yml', '.xml', '.html', '.ini')
+
+
+def extract_text(path, name):
+    """提取文件文本。返回 (preview_text, total_chars, truncated)。
+
+    二进制/无法解析 → ('', 0, False)。绝不静默截断：调用方拿得到总长与截断标志。
+    """
+    ext = os.path.splitext(name)[1].lower()
+    try:
+        if ext in TEXT_EXTS or ext == '':
+            with open(path, 'r', encoding='utf-8', errors='ignore') as f:
+                full = f.read()
+            return full[:PREVIEW_LIMIT], len(full), len(full) > PREVIEW_LIMIT
+        if ext == '.docx':
+            with zipfile.ZipFile(path) as z:
+                xml = z.read('word/document.xml').decode('utf-8', errors='ignore')
+            full = '\n'.join(re.findall(r'<w:t[^>]*>(.*?)</w:t>', xml))
+            return full[:PREVIEW_LIMIT], len(full), len(full) > PREVIEW_LIMIT
+    except Exception:
+        pass
+    return '', 0, False
+
+
+async def resolve_upload_dir(sor, uid, session_id=''):
+    """上传文件落盘目录：有当前项目 → 项目根目录；无项目 → workspace 会话目录。
+
+    两个位置都在会话 agent 的 read_file 根内（项目根 / WORKSPACE_BASE），
+    保证「上传放的位置 = agent 读的位置」。
+    返回 (dir, rel_prefix)：rel_prefix 是相对 read_file 根的路径前缀（项目根为 ''）。
+    """
+    try:
+        from pipeline_service.workspace import (
+            get_session_project_id, get_project_dir_by_id, WORKSPACE_BASE)
+    except Exception:
+        return '', ''
+    try:
+        pid = await get_session_project_id(sor, uid, session_id or '')
+        if pid:
+            pdir, _ = await get_project_dir_by_id(sor, pid)
+            if pdir:
+                return pdir, ''
+    except Exception:
+        pass
+    sess = (session_id or uid or 'default').replace('/', '_').replace('..', '_')
+    return os.path.join(WORKSPACE_BASE, '_uploads', sess), '_uploads/' + sess + '/'
+
+
+def save_uploads(target_dir, uploads):
+    """复制上传文件到目标目录，同名自动加后缀防覆盖。返回 [(保存名, 绝对路径)]。"""
+    import shutil
+    saved = []
+    if not target_dir:
+        return saved
+    try:
+        os.makedirs(target_dir, exist_ok=True)
+    except Exception:
+        return saved
+    for src, name in uploads:
+        name = os.path.basename((name or '').strip())
+        if not name or not os.path.isfile(src):
+            continue
+        target = os.path.join(target_dir, name)
+        if os.path.exists(target):
+            stem, ext = os.path.splitext(name)
+            i = 1
+            while os.path.exists(os.path.join(target_dir, f"{stem}_{i}{ext}")):
+                i += 1
+            target = os.path.join(target_dir, f"{stem}_{i}{ext}")
+        try:
+            shutil.copyfile(src, target)
+            saved.append((os.path.basename(target), target))
+        except Exception:
+            pass
+    return saved
+
+
+def build_file_context(items):
+    """items: [(filename, relpath, preview, total, truncated, is_binary)]
+
+    生成注入 prompt 的上下文。截断时显式告知预览性质与全文获取手段——
+    LLM 绝不能把预览当完整文件。
+    """
+    if not items:
+        return ''
+    parts = ["用户本次上传了以下文件："]
+    for fname, rel, preview, total, truncated, is_binary in items:
+        parts.append(f"  - {fname}" + (f"（read_file 路径：{rel}）" if rel else ""))
+    parts.append("")
+    for fname, rel, preview, total, truncated, is_binary in items:
+        if is_binary:
+            parts.append(f"【文件 {fname}】二进制文件，无法直接读取文本。")
+            if rel:
+                parts.append(f"如需处理，用 run_command 或 read_file 提示用户传文本版本。")
+            parts.append("")
+            continue
+        if truncated:
+            parts.append(
+                f"【文件 {fname} 内容预览：前 {len(preview)} 字符 / 全文共 {total} 字符，"
+                f"其余未展示——这不是完整文件】")
+            parts.append(preview)
+            if rel:
+                parts.append(
+                    f"⚠️ 需要完整内容或后文时，用 read_file 读取 \"{rel}\""
+                    f"（单次最多 30000 字符，可分段读）；不要基于预览臆断被截断部分。")
+        else:
+            parts.append(f"【文件 {fname} 完整内容（{total} 字符）】")
+            parts.append(preview)
+        parts.append("")
+    return "\n".join(parts)

wwwroot/api/agent_chat.dspy

@@ -10,20 +10,10 @@ from ahserver.filestorage import FileStorage
 
 
 def _extract_text(path, name):
-    """提取文件文本内容（docx/txt/md等），返回文本或空字符串。二进制/无法解析返回空。"""
-    ext = os.path.splitext(name)[1].lower()
-    try:
-        if ext in ('.txt', '.md', '.json', '.csv', '.py', '.log', '.yaml', '.yml', '.xml', '.html', '.ini'):
-            with open(path, 'r', encoding='utf-8', errors='ignore') as f:
-                return f.read()[:15000]
-        if ext == '.docx':
-            with zipfile.ZipFile(path) as z:
-                xml = z.read('word/document.xml').decode('utf-8', errors='ignore')
-                texts = re.findall(r'<w:t[^>]*>(.*?)</w:t>', xml)
-                return '\n'.join(texts)[:15000]
-    except Exception:
-        pass
-    return ''
+    """[已废弃] 保留签名兼容；真实逻辑在 pipeline_core.upload_tools.extract_text。"""
+    from pipeline_core.upload_tools import extract_text
+    preview, _total, _trunc = extract_text(path, name)
+    return preview
 
 
 action = (params_kw or {}).get('action', 'send_message')
@@ -46,9 +36,9 @@ if action == 'send_message':
     prompt = prompt.strip()
 
     # 处理用户上传的文件（multipart file 字段 → web_path）：
-    # 1. 抽取文本作为中性上下文注入 prompt
-    # 2. 收集 (绝对路径, 文件名)，稍后复制进项目根目录，让会话 agent 的 read_file 能找到
-    file_ctx = ''
+    # 统一走 pipeline_core.upload_tools——截断显式告知、落盘位置=agent 可读位置。
+    from pipeline_core.upload_tools import extract_text, resolve_upload_dir, save_uploads, build_file_context
+    _items = []            # build_file_context 入参
     _uploads = []          # [(src_abs, filename)]
     _fval = (params_kw or {}).get('file')
     _fpaths = _fval if isinstance(_fval, list) else ([_fval] if _fval else [])
@@ -57,11 +47,6 @@ if action == 'send_message':
             _abs = FileStorage().realPath(_fp)
             _name = os.path.basename(_abs)
             _uploads.append((_abs, _name))
-            _txt = _extract_text(_abs, _name)
-            if _txt:
-                file_ctx += f"【文件 {_name} 内容】\n{_txt}\n\n"
-            else:
-                file_ctx += f"【文件 {_name}】二进制文件，无法直接读取文本。\n\n"
         except Exception:
             pass
 
@@ -81,24 +66,24 @@ if action == 'send_message':
     # ── 产线默认能力（入口指定，如 bidding_general）：无当前项目时装载该产线能力 ──
     pipeline_id = (params_kw or {}).get('pipeline_id', '') or ''
 
-    # ── 上传文件落盘到项目根目录（会话 agent 的 read_file 以项目目录为根，
-    #    不复制进去则 agent 永远找不到；无当前项目时仅注入文本）──
-    _saved_files = []
+    # ── 上传文件落盘（有项目→项目根；无项目→workspace 会话目录）+ 生成显式截断上下文 ──
+    file_ctx = ''
     if _uploads:
         try:
             async with DBPools().sqlorContext(dbname) as sor:
-                _pid = await get_session_project_id(sor, uid, session_id)
-                _pdir, _ = await get_project_dir_by_id(sor, _pid)
-            _saved_files = copy_uploads_to_project(_pdir, _uploads)
+                _udir, _rel = await resolve_upload_dir(sor, uid, session_id)
+            _saved = save_uploads(_udir, _uploads)
+            _byname = {os.path.basename(src): n for (src, _n), (n, _p) in zip(_uploads, _saved)}
+            for _src, _name in _uploads:
+                _preview, _total, _trunc = extract_text(_src, _name)
+                _isbin = (not _preview and _total == 0)
+                _relp = (_rel + _byname.get(_name, _name)) if _name in _byname else ''
+                _items.append((_name, _relp, _preview, _total, _trunc, _isbin))
+            file_ctx = build_file_context(_items)
         except Exception:
-            _saved_files = []
-
-    if file_ctx or _saved_files:
-        _loc = ''
-        if _saved_files:
-            _loc = "文件已保存到项目根目录（" + _pdir + "），可用 read_file 直接读取：\n"
-            _loc += "\n".join("  - " + n for n, _ in _saved_files) + "\n\n"
-        prompt = "用户本次上传了以下文件：\n" + _loc + file_ctx + "\n用户指令：" + prompt
+            file_ctx = ''
+    if file_ctx:
+        prompt = file_ctx + "\n用户指令：" + prompt
 
     # ── 走 gateway 统一入口（Web AgentIO 通道）──
     # model_id：前端模型下拉选中的模型 → gateway 校验后持久化到项目（项目模型一经设置

wwwroot/api/agent_chat_generic.dspy

@@ -30,6 +30,40 @@ if action == 'send_message':
     # 默认模型名与 llm 表不匹配时直接"无可用模型"卡死。
     model_id = (params_kw or {}).get('model_id', '') or ''
 
+    # 上传文件：之前完全忽略 file 字段（上传静默丢弃）。与 agent_chat 走同一套处理：
+    # 落盘到 workspace 会话目录（generic 无项目，read_file 根=WORKSPACE_BASE，相对路径可读）。
+    import os
+    from ahserver.filestorage import FileStorage
+    from sqlor.dbpools import DBPools
+    from pipeline_core.upload_tools import extract_text, resolve_upload_dir, save_uploads, build_file_context
+    _uploads = []
+    _fval = (params_kw or {}).get('file')
+    _fpaths = _fval if isinstance(_fval, list) else ([_fval] if _fval else [])
+    for _fp in _fpaths:
+        try:
+            _abs = FileStorage().realPath(_fp)
+            _uploads.append((_abs, os.path.basename(_abs)))
+        except Exception:
+            pass
+    if _uploads:
+        try:
+            _dbname = get_module_dbname('pipeline_core')
+            async with DBPools().sqlorContext(_dbname) as sor:
+                _udir, _rel = await resolve_upload_dir(sor, uid, '')
+            _saved = save_uploads(_udir, _uploads)
+            _byname = {os.path.basename(src): n for (src, _n), (n, _p) in zip(_uploads, _saved)}
+            _items = []
+            for _src, _name in _uploads:
+                _preview, _total, _trunc = extract_text(_src, _name)
+                _isbin = (not _preview and _total == 0)
+                _relp = (_rel + _byname.get(_name, _name)) if _name in _byname else ''
+                _items.append((_name, _relp, _preview, _total, _trunc, _isbin))
+            _fctx = build_file_context(_items)
+            if _fctx:
+                prompt = _fctx + "\n用户指令：" + prompt
+        except Exception:
+            pass
+
     async def agent_stream():
         async for chunk in gateway.run_message("web", uid, prompt, generic=True, model_id=model_id):
             data = json.loads(chunk)