diff --git a/pipeline_core/upload_tools.py b/pipeline_core/upload_tools.py new file mode 100644 index 0000000..4fe39da --- /dev/null +++ b/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']*>(.*?)', 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) diff --git a/wwwroot/api/agent_chat.dspy b/wwwroot/api/agent_chat.dspy index 8b1bd2d..ff4c49f 100644 --- a/wwwroot/api/agent_chat.dspy +++ b/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']*>(.*?)', 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 校验后持久化到项目(项目模型一经设置 diff --git a/wwwroot/api/agent_chat_generic.dspy b/wwwroot/api/agent_chat_generic.dspy index 943b6bf..425f6c3 100644 --- a/wwwroot/api/agent_chat_generic.dspy +++ b/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)