feat: 切换Agent自主知识工具检索链路

ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+import asyncio
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import StaticPool
+
+from app.models import Base
+from app.models.knowledge import (
+    Knowledge,
+    KnowledgeChunk,
+    KnowledgeManifest,
+    KnowledgeSection,
+    KnowledgeSourceSnapshot,
+    KnowledgeVersion,
+)
+from app.services.knowledge_agent_service import KnowledgeAgentService
+
+
+def _database() -> Session:
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    Base.metadata.create_all(engine)
+    return Session(engine)
+
+
+def _add_published_knowledge(db: Session, *, knowledge_id: int, name: str, open_status: int = 1) -> Knowledge:
+    knowledge = Knowledge(
+        id=knowledge_id,
+        name=name,
+        feishu_space_id="space",
+        feishu_node_id=f"node-{knowledge_id}",
+        status=open_status,
+        source_status="normal",
+        manifest_confirmed=1,
+        knowledge_type="course",
+        review_mode="manual",
+    )
+    db.add(knowledge)
+    db.flush()
+    snapshot = KnowledgeSourceSnapshot(
+        knowledge_id=knowledge.id,
+        source_title=name,
+        source_content="家长沟通需要先稳定自己的情绪，再倾听孩子的真实困难。",
+        content_hash=f"hash-{knowledge_id}",
+        source_identifier=f"space/node-{knowledge_id}",
+    )
+    db.add(snapshot)
+    db.flush()
+    version = KnowledgeVersion(
+        knowledge_id=knowledge.id,
+        snapshot_id=snapshot.id,
+        version_no=1,
+        status="published",
+        quality_status="passed",
+        processing_rule_version="test-v1",
+    )
+    db.add(version)
+    db.flush()
+    knowledge.current_version_id = version.id
+    db.add(
+        KnowledgeManifest(
+            knowledge_id=knowledge.id,
+            version_id=version.id,
+            purpose="解答家长与孩子沟通、学习动力相关课程问题",
+            applicable_questions="孩子学习动力、亲子沟通",
+            inapplicable_questions="天气和交通",
+            core_topics="家长情绪、倾听孩子、学习动力",
+            boundaries="不输出完整课程资料",
+            content_hash=f"manifest-{knowledge_id}",
+            confirmed=1,
+        )
+    )
+    section = KnowledgeSection(
+        knowledge_id=knowledge.id,
+        version_id=version.id,
+        section_key="S0001",
+        title="家长沟通的第一步",
+        content="家长和学习动力不足的孩子沟通时，第一步是先稳定自己的焦虑，再倾听孩子遇到的具体困难。",
+        source_start=0,
+        source_end=44,
+        sort_order=1,
+        content_hash=f"section-{knowledge_id}",
+    )
+    db.add(section)
+    db.flush()
+    db.add(
+        KnowledgeChunk(
+            knowledge_id=knowledge.id,
+            version_id=version.id,
+            section_id=section.id,
+            title=section.title,
+            content=section.content,
+            normalized_text="家长 沟通 学习 动力 孩子 焦虑 倾听",
+            keywords='["家长","沟通","学习动力"]',
+            synonyms='["父母","交流"]',
+            source_start=0,
+            source_end=44,
+            sort_order=1,
+            content_hash=f"chunk-{knowledge_id}",
+        )
+    )
+    db.commit()
+    return knowledge
+
+
+def test_catalog_only_contains_open_published_knowledge():
+    with _database() as db:
+        _add_published_knowledge(db, knowledge_id=1, name="开放课程", open_status=1)
+        _add_published_knowledge(db, knowledge_id=2, name="关闭课程", open_status=0)
+        catalog = KnowledgeAgentService.get_knowledge_catalog(db)
+        assert [item["knowledgeId"] for item in catalog] == [1]
+
+
+def test_common_question_skips_knowledge_tools():
+    with _database() as db:
+        _add_published_knowledge(db, knowledge_id=1, name="亲子课程")
+        result = asyncio.run(KnowledgeAgentService.build_result(db, question="今天北京天气怎么样？"))
+        assert result.allow_general_knowledge is True
+        assert result.chunks == []
+        assert result.tool_trace[1]["needKnowledge"] is False
+
+
+def test_course_question_without_catalog_does_not_fall_back_to_general_knowledge():
+    with _database() as db:
+        result = asyncio.run(KnowledgeAgentService.build_result(db, question="卢慧老师课程里怎么解释亲子沟通？"))
+        assert result.allow_general_knowledge is False
+        assert result.chunks == []
+
+
+def test_course_question_searches_chunk_and_reads_parent_section():
+    with _database() as db:
+        _add_published_knowledge(db, knowledge_id=1, name="亲子课程")
+        result = asyncio.run(KnowledgeAgentService.build_result(db, question="课程里家长怎么和学习动力不足的孩子沟通？"))
+        assert result.allow_general_knowledge is False
+        assert len(result.chunks) == 1
+        assert "先稳定自己的焦虑" in result.chunks[0].content
+        assert any(item["tool"] == "read_knowledge_sections" for item in result.tool_trace)