154 lines
6.3 KiB
Python
154 lines
6.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""LLM 接入集成测试：规划/标题/意图/禁止降级/长文档两级规划。
|
||
|
||
用法：python tests/e2e_llm.py
|
||
"""
|
||
import os
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT))
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
os.environ["PLAN_CHUNK_CHARS"] = "2500" # 压低阈值，确保长文档用例真正走分段通读
|
||
|
||
from src.agent import llm # noqa: E402
|
||
from src.agent.perception import GenerateConfig # noqa: E402
|
||
from src.agent import planning as P # noqa: E402
|
||
from src.agent.planning import build_plan, PlanningError # noqa: E402
|
||
from src.agent.chat import parse_intent_llm, parse_intent # noqa: E402
|
||
|
||
SAMPLE = """# 项目背景
|
||
为提升知识周转效率启动平台建设，前期调研两个月
|
||
# 本期成果
|
||
部署完成基础服务，导入文档120份
|
||
混合检索准确率较基线提升明显
|
||
完成三个部门试点接入
|
||
# 下一步
|
||
扩大试点范围，建立运营机制
|
||
"""
|
||
|
||
failures = []
|
||
|
||
|
||
def check(name, cond, extra=""):
|
||
print(f" [{'OK' if cond else 'NG'}] {name} {extra}")
|
||
if not cond:
|
||
failures.append(name)
|
||
|
||
|
||
def make_long_doc(sections=40):
|
||
"""构造约 1.5-2 万字符的多节长文档（含数字，验证分段通读与页数硬约束）。"""
|
||
parts = []
|
||
for i in range(1, sections + 1):
|
||
parts.append(f"# 第{i}章节 阶段性工作汇报")
|
||
parts.append(f"本阶段完成任务{i}的设计与开发，投入人力3人，周期2周")
|
||
parts.append(f"完成度达到{i*2}%，质量抽检合格率98%")
|
||
parts.append("- 关键交付物已通过评审")
|
||
parts.append("- 遗留问题已登记跟踪")
|
||
parts.append("下一步将推进与业务系统的对接联调")
|
||
return "\n".join(parts)
|
||
|
||
|
||
def main():
|
||
print("== 0. 禁止降级：LLM 失败必须报错而非照搬 ==")
|
||
cfg_nf = GenerateConfig(title="降级测试", user_id=1, content=SAMPLE,
|
||
page_min=5, page_max=8)
|
||
orig_chat = llm.chat_json
|
||
try:
|
||
llm.chat_json = lambda *a, **k: None
|
||
try:
|
||
build_plan(cfg_nf)
|
||
check("no silent fallback", False, "未抛出 PlanningError")
|
||
except PlanningError as e:
|
||
check("no silent fallback", True, str(e)[:50])
|
||
finally:
|
||
llm.chat_json = orig_chat
|
||
|
||
ok = llm.ensure_backend()
|
||
check("llm backend", ok)
|
||
|
||
print("== 1. LLM 内容规划（短文档直通） ==")
|
||
t = time.time()
|
||
cfg = GenerateConfig(title="知识平台周报", user_id=1, content=SAMPLE,
|
||
scene="report", language="zh", page_min=6, page_max=9)
|
||
r = build_plan(cfg)
|
||
plan = r.plan
|
||
slides = plan["slides"]
|
||
print(f" mode={r.mode} pages={len(slides)} 耗时={time.time()-t:.0f}s")
|
||
check("llm mode used", r.mode == "llm")
|
||
check("page count in range", 6 <= len(slides) <= 9, f"n={len(slides)}")
|
||
check("cover first / end last", slides[0]["type"] == "cover" and slides[-1]["type"] == "end")
|
||
notes_n = sum(1 for s in slides if s.get("notes"))
|
||
check("per-slide notes", notes_n >= len(slides) - 2, f"notes={notes_n}")
|
||
generic = sum(1 for s in slides if s["title"] in ("",) or s["title"].startswith("要点"))
|
||
check("no generic titles", generic == 0)
|
||
|
||
def avg_len(sl):
|
||
bl = [len(b) for x in sl if x["type"] == "content" for b in x.get("content", [])]
|
||
return sum(bl) / max(len(bl), 1)
|
||
check("full-sentence bullets (short doc)", avg_len(slides) >= 12,
|
||
f"avg={avg_len(slides):.0f}")
|
||
for s in slides:
|
||
print(f" - [{s['type']:8s}] {s['title'][:24]} notes={'Y' if s.get('notes') else '-'}")
|
||
|
||
print("== 2. 长文档两级规划（分段通读→汇总大纲） ==")
|
||
long_doc = make_long_doc()
|
||
t = time.time()
|
||
cfg2 = GenerateConfig(title="季度工作总结", user_id=1, content=long_doc,
|
||
scene="report", language="zh", page_min=10, page_max=15)
|
||
r2 = build_plan(cfg2)
|
||
slides2 = r2.plan["slides"]
|
||
n2 = len(slides2)
|
||
print(f" mode={r2.mode} pages={n2} 源={len(long_doc)}字 耗时={time.time()-t:.0f}s")
|
||
check("long doc page hard cap", n2 <= 15 and n2 >= 4, f"n={n2}")
|
||
notes2 = sum(1 for s in slides2 if s.get("notes"))
|
||
check("long doc has notes", notes2 >= n2 - 2, f"notes={notes2}")
|
||
covered = sum(1 for i in range(1, 41) if str(i) in "".join(
|
||
s["title"] + "".join(s["content"]) for s in slides2))
|
||
check("late sections represented", covered >= 3, f"covered_sections={covered}")
|
||
check("full-sentence bullets (long doc)", avg_len(slides2) >= 12,
|
||
f"avg={avg_len(slides2):.0f}")
|
||
|
||
print("== 3. 智能标题 ==")
|
||
t = time.time()
|
||
data = llm.chat_json(
|
||
"根据以下 PPT 内容拟 3 个标题（每个不超过 20 字）。"
|
||
'输出 JSON {"titles": ["..."]}。\n\n' + SAMPLE)
|
||
titles = [str(x).strip() for x in (data or {}).get("titles", []) if str(x).strip()]
|
||
print(f" titles={titles} 耗时={time.time()-t:.0f}s")
|
||
check("title suggest", len(titles) >= 1)
|
||
|
||
print("== 4. LLM 意图解析（规则无法处理的句式） ==")
|
||
msg = '第2页末尾添加一条要点：风险与依赖已同步全组'
|
||
rule = parse_intent(msg, {})
|
||
llm_it = parse_intent_llm(msg, plan)
|
||
check("rule cannot parse (no quotes)", rule["type"] == "unknown", str(rule))
|
||
check("llm parses add", llm_it and llm_it.get("type") == "edit_page"
|
||
and llm_it.get("op") == "add" and llm_it.get("page_no") == 2,
|
||
str(llm_it))
|
||
msg2 = "把标题改成 AI 知识平台建设汇报"
|
||
it2 = parse_intent_llm(msg2, plan)
|
||
check("llm parses title", it2 and it2.get("type") == "change_title"
|
||
and "知识平台" in it2.get("title", ""), str(it2))
|
||
|
||
print("== 5. 渲染 LLM plan（含 notes） ==")
|
||
import json
|
||
import tempfile
|
||
from src.engine_bridge import render_plan, verify_output
|
||
tmp = Path(tempfile.mkdtemp(prefix="llm_plan_"))
|
||
p = tmp / "plan.json"
|
||
p.write_text(json.dumps(plan, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
out = tmp / "out.pptx"
|
||
render_plan(p, out)
|
||
vok, detail = verify_output(out, p)
|
||
check("render+verify llm plan", vok, detail[:60])
|
||
|
||
print("\n" + ("LLM E2E ALL PASSED" if not failures else "FAILURES: " + "; ".join(failures)))
|
||
sys.exit(1 if failures else 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|