新知识库访问工具

.gitignore

@@ -5,6 +5,7 @@
 
 # 个人调试文件
 terminal_demo.py
+test_polling_performance.py
 
 # Python编译文件
 __pycache__/

core/agent.py

@@ -43,7 +43,7 @@ def create_system_prompt(
 {"- API用户的认证令牌: " + token if token else ""}
 时间：{datetime.datetime.now().strftime("%m-%d %H:%M")}
 
-库存及销量查询结果尽量以 Markdown 表格格式输出，格式如下：
+数据查询结果尽量以 Markdown 表格格式输出，格式如下：
 | 列名1 | 列名2 | 列名3 |
 | :--- | :--- | :--- |
 | 数据1 | 数据2 | 数据3 |

tools/knowledge_tools.py

@@ -7,20 +7,127 @@ from .base_tool import html_to_text, get_unique_match_count
 from config.settings import settings
 
 
+# @tool
+# def get_knowledge_list(filter_words: List[str], match_limit: int = 3) -> str:
+#     """根据关键词筛选知识库文章列表
+
+#     Args:
+#         filter_words: 关键词列表，匹配任一关键词即返回
+#         match_limit: 最小匹配数(默认3)，无结果时可减少重试(最小1)
+
+#     Returns:
+#         文章列表，格式：每行"DocID:DocName|keyword"，用于后续获取内容
+#     """
+#     print(f"正在查询知识库列表，筛选关键词：{filter_words} 匹配下限：{match_limit}")
+
+#     kms_list_url = settings.KMS_LIST_URL  # os.getenv("KMS_LIST_URL")
+#     payload = {
+#         "categorycodeList": [],
+#         "ignoreTypeSub": False,
+#         "ignoreStandardByTopic": True,
+#     }
+
+#     headers = {
+#         "Accept": "application/json, text/plain, */*",
+#         "Content-Type": "application/json",
+#         "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+#     }
+
+#     try:
+#         response = requests.post(
+#             kms_list_url, headers=headers, json=payload, timeout=10
+#         )
+#         if response.status_code == 200:
+#             data = response.json()
+#             matched_lines = ""
+#             for doc in data["docList"]:
+#                 doc_id = doc["DocID"]
+#                 doc_name = doc["DocName"]
+#                 doc_keywords = doc["keyword"]
+#                 search_text = f"{doc_name} {doc_keywords}".lower()
+
+#                 if not filter_words:
+#                     line = (
+#                         f"{doc_id}:{doc_name}|{doc_keywords}"
+#                         if doc_keywords
+#                         else f"{doc_id}:{doc_name}"
+#                     )
+#                     matched_lines += line + "\n"
+#                 else:
+#                     match_count = get_unique_match_count(search_text, filter_words)
+#                     if match_count >= match_limit:
+#                         line = (
+#                             f"{doc_id}:{doc_name}|{doc_keywords}"
+#                             if doc_keywords
+#                             else f"{doc_id}:{doc_name}"
+#                         )
+#                         matched_lines += line + "\n"
+
+#             return matched_lines
+#         else:
+#             return f"请求失败，状态码: {response.status_code}"
+#     except Exception as e:
+#         return f"请求异常: {e}"
+
+
+# @tool
+# def get_knowledge_content(docid: str) -> str:
+#     """获取知识库文章内容
+
+#     Args:
+#         docid: 知识库文章的DocID
+
+#     Returns:
+#         知识库文章内容
+#     """
+#     print(f"正在获取知识库文章内容，DocID: {docid}")
+
+#     kms_view_url = settings.KMS_VIEW_URL  # os.getenv("KMS_VIEW_URL")
+#     headers = {
+#         "Accept": "application/json, text/plain, */*",
+#         "Content-Type": "application/json",
+#         "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+#     }
+
+#     try:
+#         payload = {"docid": docid}
+#         response = requests.post(
+#             kms_view_url, headers=headers, json=payload, timeout=10
+#         )
+#         if response.status_code == 200:
+#             data = response.json()
+#             doc_html = data.get("DocHtml", "")
+#             plain_text = html_to_text(doc_html)
+#             print(f"已获取到ID: {docid}的文章内容，长度{len(plain_text)}")
+#             return plain_text
+#         else:
+#             return f"请求失败，状态码: {response.status_code}"
+#     except Exception as e:
+#         return f"请求异常: {e}"
+
+
 @tool
-def get_knowledge_list(filter_words: List[str], match_limit: int = 3) -> str:
-    """根据关键词筛选知识库文章列表
+def search_and_retrieve_knowledge(keywords: List[str], max_matches: int = 5) -> str:
+    """搜索并获取最相关的知识库文章内容
 
+    此工具会：
+    1. 自动搜索知识库文章列表
+    2. 根据关键词匹配度排序，选择匹配最多关键词的文章
+    3. 自动获取并返回该文章的完整内容
+    拆分关键词核心原则:
+    最小化原则：将用户问题拆分为最小单位的关键词,最好2个字一个关键词
+    例如："销售订单终止数量失败" → ["销售", "订单", "终止", "数量", "失败"],"销售订单提示没有新建权限怎么办" → ["销售", "订单", "新建", "权限"]
     Args:
-        filter_words: 关键词列表，匹配任一关键词即返回
-        match_limit: 最小匹配数(默认3)，无结果时可减少重试(最小1)
+        keywords: 关键词列表，会尽可能细化匹配
+        max_matches: 最多考虑的匹配文章数(默认5)
 
     Returns:
-        文章列表，格式：每行"DocID:DocName|keyword"，用于后续获取内容
+        最相关文章的完整内容
     """
-    print(f"正在查询知识库列表，筛选关键词：{filter_words} 匹配下限：{match_limit}")
+    print(f"正在搜索知识库，关键词：{keywords}")
 
-    kms_list_url = settings.KMS_LIST_URL  # os.getenv("KMS_LIST_URL")
+    # 1. 获取文章列表
+    kms_list_url = settings.KMS_LIST_URL
     payload = {
         "categorycodeList": [],
         "ignoreTypeSub": False,
@@ -34,73 +141,76 @@ def get_knowledge_list(filter_words: List[str], match_limit: int = 3) -> str:
     }
 
     try:
+        # 获取所有文章
         response = requests.post(
             kms_list_url, headers=headers, json=payload, timeout=10
         )
-        if response.status_code == 200:
-            data = response.json()
-            matched_lines = ""
-            for doc in data["docList"]:
-                doc_id = doc["DocID"]
-                doc_name = doc["DocName"]
-                doc_keywords = doc["keyword"]
-                search_text = f"{doc_name} {doc_keywords}".lower()
-
-                if not filter_words:
-                    line = (
-                        f"{doc_id}:{doc_name}|{doc_keywords}"
-                        if doc_keywords
-                        else f"{doc_id}:{doc_name}"
-                    )
-                    matched_lines += line + "\n"
-                else:
-                    match_count = get_unique_match_count(search_text, filter_words)
-                    if match_count >= match_limit:
-                        line = (
-                            f"{doc_id}:{doc_name}|{doc_keywords}"
-                            if doc_keywords
-                            else f"{doc_id}:{doc_name}"
-                        )
-                        matched_lines += line + "\n"
-
-            return matched_lines
-        else:
-            return f"请求失败，状态码: {response.status_code}"
-    except Exception as e:
-        return f"请求异常: {e}"
+        if response.status_code != 200:
+            return f"获取文章列表失败，状态码: {response.status_code}"
 
+        data = response.json()
+        doc_list = data.get("docList", [])
 
-@tool
-def get_knowledge_content(docid: str) -> str:
-    """获取知识库文章内容
+        if not doc_list:
+            return "知识库中没有找到文章"
 
-    Args:
-        docid: 知识库文章的DocID
+        # 2. 计算匹配度并排序
+        matched_docs = []
+        for doc in doc_list:
+            doc_id = doc["DocID"]
+            doc_name = doc["DocName"]
+            doc_keywords = doc["keyword"]
+            search_text = f"{doc_name} {doc_keywords}".lower()
 
-    Returns:
-        知识库文章内容
-    """
-    print(f"正在获取知识库文章内容，DocID: {docid}")
+            # 计算匹配的关键词数量
+            match_count = get_unique_match_count(search_text, keywords)
+            if match_count > 0:
+                matched_docs.append({
+                    "doc_id": doc_id,
+                    "doc_name": doc_name,
+                    "match_count": match_count,
+                    "keywords": doc_keywords
+                })
 
-    kms_view_url = settings.KMS_VIEW_URL  # os.getenv("KMS_VIEW_URL")
-    headers = {
-        "Accept": "application/json, text/plain, */*",
-        "Content-Type": "application/json",
-        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
-    }
+        if not matched_docs:
+            return "没有找到与关键词相关的文章"
 
-    try:
-        payload = {"docid": docid}
-        response = requests.post(
-            kms_view_url, headers=headers, json=payload, timeout=10
+        # 按匹配数量降序排序，选择前max_matches个
+        matched_docs.sort(key=lambda x: x["match_count"], reverse=True)
+        top_docs = matched_docs[:max_matches]
+
+        print(f"找到 {len(matched_docs)} 篇相关文章，前{len(top_docs)}篇匹配度最高")
+        for i, doc in enumerate(top_docs):
+            print(f"  {i+1}. {doc['doc_name']} (匹配{doc['match_count']}个关键词)")
+
+        # 3. 获取匹配度最高的文章内容
+        best_doc = top_docs[0]
+        print(f"选择最相关的文章: {best_doc['doc_name']} (DocID: {best_doc['doc_id']})")
+
+        # 4. 获取文章内容
+        kms_view_url = settings.KMS_VIEW_URL
+        content_payload = {"docid": best_doc['doc_id']}
+        content_response = requests.post(
+            kms_view_url, headers=headers, json=content_payload, timeout=10
         )
-        if response.status_code == 200:
-            data = response.json()
-            doc_html = data.get("DocHtml", "")
-            plain_text = html_to_text(doc_html)
-            print(f"已获取到ID: {docid}的文章内容，长度{len(plain_text)}")
-            return plain_text
-        else:
-            return f"请求失败，状态码: {response.status_code}"
+
+        if content_response.status_code != 200:
+            return f"获取文章内容失败，状态码: {content_response.status_code}"
+
+        content_data = content_response.json()
+        doc_html = content_data.get("DocHtml", "")
+        plain_text = html_to_text(doc_html)
+
+        # 5. 构建返回结果
+        result = f"【知识库文章】\n"
+        result += f"标题: {best_doc['doc_name']}\n"
+        result += f"匹配关键词数量: {best_doc['match_count']}\n"
+        if best_doc['keywords']:
+            result += f"文章关键词: {best_doc['keywords']}\n"
+        result += f"内容: {plain_text}\n"
+
+        print(f"已获取文章内容，长度: {len(plain_text)} 字符")
+        return result
+
     except Exception as e:
-        return f"请求异常: {e}"
+        return f"搜索和获取知识库文章时出错: {e}"