# ============================================================ # 多轮对话上下文（moduleId: core-context, 可重生 ✅） # 把「是的 / 好 / 那个」等短回复接到上一轮；识别文件夹解析意图。 # ============================================================ from __future__ import annotations import os import re from typing import Any HistoryItem = dict[str, str] # {role: user|agent, text: str} def normalize_history(raw: list[Any] | None, *, limit: int = 12) -> list[HistoryItem]: out: list[HistoryItem] = [] for m in raw or []: if not isinstance(m, dict): continue role = str(m.get("role") or "") if role in ("assistant", "bot", "ai"): role = "agent" if role not in ("user", "agent"): continue text = str(m.get("text") or m.get("content") or "").strip() if not text: continue out.append({"role": role, "text": text}) return out[-limit:] def last_agent_text(history: list[HistoryItem]) -> str: for m in reversed(history): if m["role"] == "agent": return m["text"] return "" def last_user_text(history: list[HistoryItem]) -> str: for m in reversed(history): if m["role"] == "user": return m["text"] return "" def format_history_for_llm(history: list[HistoryItem], *, max_chars: int = 3500) -> str: lines: list[str] = [] for m in history: who = "用户" if m["role"] == "user" else "助手" chunk = m["text"] if len(chunk) > 500: chunk = chunk[:500] + "…" lines.append(f"{who}：{chunk}") blob = "\n".join(lines) if len(blob) > max_chars: blob = "…\n" + blob[-max_chars:] return blob _AFFIRM = re.compile( r"^(是的?|对|好|好的|行|可以|嗯+|恩+|哦+|OK|ok|yes|Yes|确认|同意|就这样|开始吧|来吧|搞起)[\s!！.。？?]*$", re.I, ) _DENY = re.compile(r"^(不|不要|别|算了|取消|不用了|否)[\s!！.。？?]*$", re.I) def _extract_quoted_commands(agent_text: str) -> list[str]: """从上一轮助手回复里抠「带我排一版」这类可执行口令。""" cmds = re.findall(r"[「『\"“]([^」』\"”]{2,24})[」』\"”]", agent_text or "") # 常见无引号建议 for pat in ( r"带我排一版", r"直接排一版", r"帮我看看现在能不能排", r"分析项目数据", r"跑一版柔性排产", r"我要排产", ): if pat in (agent_text or "") and pat not in cmds: cmds.append(pat) return cmds def resolve_followup(text: str, history: list[HistoryItem]) -> dict[str, Any]: """短回复 → 改写为明确意图；返回 {text, note, kind}。""" t = (text or "").strip() prev_agent = last_agent_text(history) prev_user = last_user_text(history) if _DENY.match(t): return { "text": t, "note": "用户拒绝了上一轮建议", "kind": "deny", "reply": "已取消本次操作，当前数据未修改。需要继续时，请重新描述目标。", } if _AFFIRM.match(t) and prev_agent: cmds = _extract_quoted_commands(prev_agent) # 上一轮在问「你更想先做哪一步」或给了具体建议 if cmds: chosen = cmds[0] return { "text": chosen, "note": f"用户用「{t}」确认上一轮建议 → 执行「{chosen}」", "kind": "affirm", } # 上一轮是分析/盘点后说「直接排一版」 if re.search(r"直接排一版|带我排一版|马上给你出", prev_agent): return { "text": "直接排一版", "note": f"用户确认「{t}」→ 直接排一版", "kind": "affirm", } # 上一轮在问要不要看交期/天气耽误 if re.search(r"单号|交期|耽误", prev_agent) and prev_user: return { "text": f"继续：{prev_user}", "note": "用户确认继续上一话题", "kind": "affirm", } return { "text": "帮我看看现在能不能排", "note": f"用户说「{t}」，上一轮无明确口令 → 默认检查能不能排", "kind": "affirm", } # 「那个 / 上面说的 / 刚才那个」 if re.search(r"^(那个|上面|刚才|继续|再说一下|展开说说)[\s!！.。？?]*$", t) and prev_user: return { "text": prev_user, "note": "用户指代上一问", "kind": "refer", } return {"text": t, "note": "", "kind": "plain"} def wants_folder_listing(text: str) -> bool: return bool(re.search( r"文件夹|目录|项目(里|里面|内)?(的)?(文件|数据|资料)|" r"解析.{0,8}(文件夹|目录|文件)|把.{0,6}(文件夹|目录|文件).{0,8}(展现|展示|列出来|看看)|" r"workDir|工作目录|工程目录", text or "", re.I, )) def list_project_folder(session_id: str | None) -> str: """列出当前会话所属项目的登记文件 + workDir 磁盘文件。""" try: from server.state.projects import get_project_store ps = get_project_store() snap = ps.snapshot(include_messages=False) except Exception as exc: return f"项目信息暂时读不到：{exc}" sessions = {s["id"]: s for s in (snap.get("sessions") or []) if isinstance(s, dict)} projects = {p["id"]: p for p in (snap.get("projects") or []) if isinstance(p, dict)} sess = sessions.get(session_id or "") or {} pid = sess.get("projectId") proj = projects.get(pid) if pid else None lines = ["**当前项目里的资料：**", ""] if not proj: lines.append("这会儿话题不在某个工程项目下（独立任务）。") lines.append("你可以先在左侧选一个项目，或把 Excel 用输入框左边 **+** 附上来，我帮你解析。") else: lines.append(f"- 项目名称：**{proj.get('name') or '未命名'}**") work = (proj.get("workDir") or "").strip() if work: lines.append(f"- 工程目录：`{work}`") else: lines.append("- 工程目录：还没设定（建项目时可选文件夹）") files = [f for f in (snap.get("files") or []) if isinstance(f, dict) and f.get("projectId") == pid] lines.append("") lines.append("**项目面板已登记的文件：**") if not files: lines.append("- （还没有登记文件）") else: for f in files[:30]: lines.append( f"- {f.get('name')}（{f.get('kind') or 'file'}）" + (f"：{f.get('note')}" if f.get("note") else "") ) if len(files) > 30: lines.append(f"- ……另有 {len(files) - 30} 个") if work and os.path.isdir(work): lines.append("") lines.append("**工程目录里实际看到的文件：**") try: names = sorted(os.listdir(work)) except OSError as exc: lines.append(f"- 读目录失败：{exc}") names = [] shown = 0 for name in names: if name.startswith("."): continue path = os.path.join(work, name) tag = "文件夹" if os.path.isdir(path) else "文件" size = "" if os.path.isfile(path): try: size = f" · {os.path.getsize(path)} 字节" except OSError: size = "" lines.append(f"- [{tag}] {name}{size}") shown += 1 if shown >= 40: lines.append(f"- ……目录里还有更多，我先列前 {shown} 个") break if shown == 0: lines.append("- （目录是空的，或没有可读文件）") # 顺带提示可解析的表格 tables = [n for n in names if re.search(r"\.(xlsx|xlsm|csv|txt)$", n, re.I)] if tables: lines.append("") lines.append( f"里面有 **{len(tables)}** 个表格类文件。" "要我按内容解析的话，用输入框 **+** 附上其中一个，或者说「分析这个文件」。" ) elif work: lines.append("") lines.append(f"工程目录路径我记下了，但本机访问不到：`{work}`（可能没插盘或路径变了）。") lines.append("") lines.append( "如果要看的是**排产主数据**（订单/物料/工艺），跟我说「帮我分析一下当前项目里面的数据」。" ) return "\n".join(lines)