|
|
@@ -0,0 +1,151 @@
|
|
|
+"""
|
|
|
+LangGraph 自动打标签 (auto-tag) 功能测试
|
|
|
+=============================================
|
|
|
+
|
|
|
+背景:
|
|
|
+ Java 后端在知识中心发布/更新文章时, 会调用 LangGraph 的 POST /api/v1/chat
|
|
|
+ 端点, 传入"提取文章标签"的 prompt, 让 LLM 返回 3-5 个中文关键词标签。
|
|
|
+
|
|
|
+ Java 端构造的 prompt (见 ArticleService.autoGenerateTags()):
|
|
|
+ "你是一个家庭教育内容分析助手。请根据以下文章内容,提取 3 到 5 个
|
|
|
+ 最能代表文章主题的中文关键词作为标签。
|
|
|
+ 要求:1) 关键词为中文短语或名词,2-6 个字;2) 用英文逗号分隔;
|
|
|
+ 3) 不要包含编号、引号或解释文字;4) 只返回标签列表本身。
|
|
|
+ 文章内容:{plainText}"
|
|
|
+
|
|
|
+ Java 端解析逻辑 (与下方 parse_tags 保持一致):
|
|
|
+ - 用英文逗号 split
|
|
|
+ - 去掉行首的数字/点/横线/空白/顿号句号逗号
|
|
|
+ - 去掉行尾空白/顿号句号逗号
|
|
|
+ - 标签非空、长度 <= 20、最多取 5 个
|
|
|
+
|
|
|
+本测试: 复用 ASGI 内嵌调用, 验证 LangGraph 返回的标签质量与格式,
|
|
|
+ 并验证其能被 Java 端的解析逻辑正确解析。
|
|
|
+"""
|
|
|
+
|
|
|
+import re
|
|
|
+
|
|
|
+import pytest
|
|
|
+from httpx import AsyncClient, ASGITransport
|
|
|
+
|
|
|
+from app.main import app
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# 测试数据
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+# 真实文章内容 (来自 articles/D14*, 肠道菌群/过敏主题)
|
|
|
+ARTICLE_GUT = (
|
|
|
+ "儿童过敏反复不好,根源可能在肠道。人体约70%的免疫细胞分布在肠道,"
|
|
|
+ "肠道是人体最大的免疫器官。当肠道菌群处于健康平衡状态时,免疫系统能够准确区分"
|
|
|
+ "朋友和敌人。但当菌群失衡——有益菌减少、有害菌过度繁殖——肠道屏障功能就会受损。"
|
|
|
+ "科学调理四步走:第一步消除破坏因素,减少精制糖和加工食品;第二步重建菌群平衡,"
|
|
|
+ "增加膳食纤维摄入;第三步修复肠道屏障,补充谷氨酰胺、锌、维生素D;第四步维持健康"
|
|
|
+ "生活方式。每个孩子的菌群状况都是独特的,这正是菌群检测的核心价值。"
|
|
|
+)
|
|
|
+
|
|
|
+# 短内容
|
|
|
+ARTICLE_SHORT = "孩子睡眠质量差,作息不规律怎么办?规律作息对成长很重要。"
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# 工具函数
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+# 与 Java 端 ArticleService.autoGenerateTags() 的解析逻辑保持一致
|
|
|
+TAG_CLEAN_START = re.compile(r"^[\d\.\-\s、。,]+")
|
|
|
+TAG_CLEAN_END = re.compile(r"[\s、。,]+$")
|
|
|
+
|
|
|
+
|
|
|
+def parse_tags(ai_response: str) -> list:
|
|
|
+ """按 Java 端解析规则解析 LLM 返回的标签列表"""
|
|
|
+ if not ai_response:
|
|
|
+ return []
|
|
|
+ tags = []
|
|
|
+ for raw in ai_response.split(","):
|
|
|
+ tag = TAG_CLEAN_START.sub("", raw.strip().strip())
|
|
|
+ tag = TAG_CLEAN_END.sub("", tag).strip()
|
|
|
+ if tag and len(tag) <= 20 and len(tags) < 5:
|
|
|
+ tags.append(tag)
|
|
|
+ return tags
|
|
|
+
|
|
|
+
|
|
|
+def build_tag_prompt(plain_text: str) -> str:
|
|
|
+ """构造与 Java 端完全一致的标签提取 prompt"""
|
|
|
+ return (
|
|
|
+ "你是一个家庭教育内容分析助手。请根据以下文章内容,"
|
|
|
+ "提取 3 到 5 个最能代表文章主题的中文关键词作为标签。"
|
|
|
+ "要求:1) 关键词为中文短语或名词,2-6 个字;2) 用英文逗号分隔;"
|
|
|
+ "3) 不要包含编号、引号或解释文字;4) 只返回标签列表本身。"
|
|
|
+ f"\n\n文章内容:{plain_text}"
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+async def call_chat(query: str, user_id: int = 0) -> dict:
|
|
|
+ """调用 LangGraph /api/v1/chat 端点"""
|
|
|
+ transport = ASGITransport(app=app)
|
|
|
+ async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
|
+ resp = await client.post("/api/v1/chat", json={
|
|
|
+ "query": query,
|
|
|
+ "user_id": user_id,
|
|
|
+ "conversation_id": "",
|
|
|
+ })
|
|
|
+ assert resp.status_code == 200, f"chat 端点返回 {resp.status_code}"
|
|
|
+ return resp.json()
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# 测试用例
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_auto_tag_endpoint_responds():
|
|
|
+ """发送标签 prompt, 端点应返回 200, 且含 answer 与 tasks 数组"""
|
|
|
+ data = await call_chat(build_tag_prompt(ARTICLE_GUT))
|
|
|
+ assert "answer" in data, "响应缺少 answer 字段"
|
|
|
+ assert isinstance(data.get("answer"), str)
|
|
|
+ assert isinstance(data.get("tasks"), list)
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_auto_tag_valid_format():
|
|
|
+ """返回的标签应能被 Java 解析逻辑解析为 3-5 个中文标签"""
|
|
|
+ data = await call_chat(build_tag_prompt(ARTICLE_GUT))
|
|
|
+ tags = parse_tags(data.get("answer", ""))
|
|
|
+ assert 3 <= len(tags) <= 5, f"标签数量应为 3-5 个, 实际 {len(tags)}: {tags}"
|
|
|
+ for tag in tags:
|
|
|
+ # 2-6 个中文字
|
|
|
+ chinese_chars = re.findall(r"[\u4e00-\u9fff]", tag)
|
|
|
+ assert 2 <= len(chinese_chars) <= 6, f"标签应为 2-6 个中文字: '{tag}'"
|
|
|
+ # 不应包含编号/引号/英文逗号
|
|
|
+ assert not re.search(r"[\d\"',。]", tag), f"标签不应包含数字/引号/标点: '{tag}'"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_auto_tag_content_relevance():
|
|
|
+ """标签应反映文章核心主题 (肠道/过敏/免疫 等)"""
|
|
|
+ data = await call_chat(build_tag_prompt(ARTICLE_GUT))
|
|
|
+ tags = parse_tags(data.get("answer", ""))
|
|
|
+ assert tags, "未解析出任何标签"
|
|
|
+ # 核心主题词中至少命中一个
|
|
|
+ core_keywords = ["肠胃", "过敏", "免疫", "肠道", "菌群", "营养", "健康"]
|
|
|
+ hit = [kw for kw in core_keywords if any(kw in t for t in tags)]
|
|
|
+ assert hit, f"标签未命中核心主题词, 实际标签: {tags}"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_auto_tag_short_content():
|
|
|
+ """短内容也应正常返回标签, 不崩溃"""
|
|
|
+ data = await call_chat(build_tag_prompt(ARTICLE_SHORT))
|
|
|
+ tags = parse_tags(data.get("answer", ""))
|
|
|
+ assert tags, "短内容也应提取出标签"
|
|
|
+ assert len(tags) <= 5
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_auto_tag_empty_content():
|
|
|
+ """空内容应安全返回空标签, 不崩溃"""
|
|
|
+ data = await call_chat(build_tag_prompt(""))
|
|
|
+ # 空内容不应抛异常, answer 可为空字符串
|
|
|
+ assert "answer" in data
|
|
|
+ assert isinstance(data.get("answer"), str)
|