mirror of
https://github.com/countbot-ai/CountBot.git
synced 2026-09-14 20:46:47 +08:00
w
14d4d4e2a0
功能(核心): 会话级配置系统与前端体验全面升级 核心功能：- 会话级配置系统，轻松打造你的AI团队（每个会话独立 API、模型、提示词） - 新增微博、企业微信、小智 AI 三个 IM 渠道 - 实现企业微信、飞书两个IM渠道的消息流式输出 - 全面优化多智能体协作系统 - 全面优化项目代码兼容性 - 新增 /help 命令（通过 IM 渠道输入即可体验） - 集成 Mermaid 图表渲染引擎 - 全面兼容 OpenClaw Skills 技能生态 - Heartbeat 主动问候系统重构 - WEB聊天页面支持自定义用户头像 - 支持自定义输出语言，比如中文、英文、西语等 - 修复大量已知问题
131 lines
3.9 KiB
Python
131 lines
3.9 KiB
Python
"""工具调用文本解析器"""
|
|
|
|
import json
|
|
import re
|
|
from typing import Any, Dict, Optional
|
|
from loguru import logger
|
|
|
|
|
|
class ToolCallParser:
|
|
|
|
JSON_PATTERN = re.compile(
|
|
r'\{[\s\n]*"name"[\s\n]*:[\s\n]*"([^"]+)"[\s\n]*,[\s\n]*"arguments"[\s\n]*:[\s\n]*(\{[^}]*\})[\s\n]*\}',
|
|
re.DOTALL
|
|
)
|
|
|
|
SIMPLE_PATTERN = re.compile(
|
|
r'^([a-z_]+)\n((?:[a-z_]+:\s*.+\n?)+)',
|
|
re.MULTILINE
|
|
)
|
|
|
|
@classmethod
|
|
def parse(cls, text: str) -> Optional[Dict[str, Any]]:
|
|
if not text or not isinstance(text, str):
|
|
return None
|
|
|
|
text = text.strip()
|
|
|
|
result = cls._parse_json(text)
|
|
if result:
|
|
logger.debug(f"Parsed tool call (JSON): {result['name']}")
|
|
return result
|
|
|
|
result = cls._parse_simple(text)
|
|
if result:
|
|
logger.debug(f"Parsed tool call (simple): {result['name']}")
|
|
return result
|
|
|
|
result = cls._parse_pure_json(text)
|
|
if result:
|
|
logger.debug(f"Parsed tool call (pure JSON): {result['name']}")
|
|
return result
|
|
|
|
return None
|
|
|
|
@classmethod
|
|
def _parse_json(cls, text: str) -> Optional[Dict[str, Any]]:
|
|
match = cls.JSON_PATTERN.search(text)
|
|
if not match:
|
|
return None
|
|
|
|
try:
|
|
name = match.group(1)
|
|
arguments_str = match.group(2)
|
|
arguments = json.loads(arguments_str)
|
|
|
|
return {
|
|
"name": name,
|
|
"arguments": arguments
|
|
}
|
|
except (json.JSONDecodeError, IndexError) as e:
|
|
logger.warning(f"Failed to parse JSON tool call: {e}")
|
|
return None
|
|
|
|
@classmethod
|
|
def _parse_pure_json(cls, text: str) -> Optional[Dict[str, Any]]:
|
|
try:
|
|
data = json.loads(text)
|
|
|
|
if isinstance(data, dict) and "name" in data:
|
|
name = data["name"]
|
|
arguments = data.get("arguments", {})
|
|
|
|
if not isinstance(arguments, dict):
|
|
if isinstance(arguments, str):
|
|
try:
|
|
arguments = json.loads(arguments)
|
|
except json.JSONDecodeError:
|
|
arguments = {"value": arguments}
|
|
else:
|
|
arguments = {"value": arguments}
|
|
|
|
return {
|
|
"name": name,
|
|
"arguments": arguments
|
|
}
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
return None
|
|
|
|
@classmethod
|
|
def _parse_simple(cls, text: str) -> Optional[Dict[str, Any]]:
|
|
match = cls.SIMPLE_PATTERN.search(text)
|
|
if not match:
|
|
return None
|
|
|
|
try:
|
|
name = match.group(1)
|
|
params_text = match.group(2)
|
|
|
|
arguments = {}
|
|
for line in params_text.strip().split('\n'):
|
|
if ':' in line:
|
|
key, value = line.split(':', 1)
|
|
key = key.strip()
|
|
value = value.strip().strip('"\'')
|
|
arguments[key] = value
|
|
|
|
return {
|
|
"name": name,
|
|
"arguments": arguments
|
|
}
|
|
except (IndexError, ValueError) as e:
|
|
logger.warning(f"Failed to parse simple tool call: {e}")
|
|
return None
|
|
|
|
@classmethod
|
|
def is_tool_call_text(cls, text: str) -> bool:
|
|
if not text or not isinstance(text, str):
|
|
return False
|
|
|
|
text = text.strip()
|
|
|
|
indicators = [
|
|
'"name"' in text and '"arguments"' in text,
|
|
text.startswith('{') and text.endswith('}'),
|
|
'\n' in text and ':' in text,
|
|
]
|
|
|
|
return any(indicators)
|