xiaoli-cli/core/parser.py

"""JSON解析 — 保底方案"""
import json, re, uuid
from .types import ToolCall, Plan


def parse_prompt_json(text):
    """解析prompt-JSON，返回 (text_content, tool_calls_or_None)"""
    text = text.strip()
    json_str = _extract_json(text)
    if json_str is None:
        return text, None
    try:
        data = json.loads(json_str)
    except json.JSONDecodeError:
        return text, None
    idx = text.find(json_str)
    prefix = text[:idx].strip() if idx > 0 else ""
    if isinstance(data, list):
        calls = [c for c in (_make(c) for c in data) if c]
        return prefix, calls if calls else None
    if isinstance(data, dict):
        c = _make(data)
        if c:
            return prefix, [c]
        return prefix, data
    return text, None


def _extract_json(text):
    """平衡括号扫描"""
    depth = 0
    start = -1
    in_str = False
    esc = False
    for i, c in enumerate(text):
        if esc:
            esc = False
            continue
        if c == '\\' and in_str:
            esc = True
            continue
        if c == '"' and not esc:
            in_str = not in_str
            continue
        if in_str:
            continue
        if c == '{':
            if depth == 0:
                start = i
            depth += 1
        elif c == '}':
            depth -= 1
            if depth == 0 and start != -1:
                candidate = text[start:i+1]
                if '"action"' in candidate or '"tool"' in candidate:
                    return candidate
                start = -1
        elif c == '[' and depth == 0:
            bd = 1
            j = i + 1
            in_s = False
            e2 = False
            while j < len(text) and bd > 0:
                cj = text[j]
                if e2:
                    e2 = False
                elif cj == '\\' and in_s:
                    e2 = True
                elif cj == '"' and not e2:
                    in_s = not in_s
                elif not in_s:
                    if cj == '[': bd += 1
                    elif cj == ']': bd -= 1
                j += 1
            if bd == 0:
                candidate = text[i:j]
                if '"action"' in candidate or '"tool"' in candidate:
                    return candidate
    return None


def _make(data):
    """dict → ToolCall"""
    if not isinstance(data, dict):
        return None
    cid = data.get("id", str(uuid.uuid4())[:8])
    name = ""
    args = ""
    arguments = None
    if data.get("action") == "use_tool":
        name = data.get("tool", data.get("name", ""))
        ra = data.get("args", data.get("arguments", ""))
        if isinstance(ra, str): args = ra
        elif isinstance(ra, dict):
            arguments = ra
            args = json.dumps(ra, ensure_ascii=False)
        else: args = str(ra)
    elif "name" in data and "arguments" in data:
        name = data["name"]
        ra = data["arguments"]
        if isinstance(ra, str):
            args = ra
            try: arguments = json.loads(ra)
            except: pass
        elif isinstance(ra, dict):
            arguments = ra
            args = json.dumps(ra, ensure_ascii=False)
    elif "tool" in data and "args" in data:
        name = data["tool"]
        ra = data["args"]
        if isinstance(ra, str): args = ra
        elif isinstance(ra, dict):
            arguments = ra
            args = json.dumps(ra, ensure_ascii=False)
    else:
        return None
    if not name:
        return None
    return ToolCall(id=cid, name=name, args=args, arguments=arguments, raw=data)


def interpret(resp, fc_mode):
    """EngineResponse → Plan"""
    # FC模式
    if fc_mode and resp.tool_calls:
        return Plan(action="use_tool", text=resp.text or "", tool_calls=resp.tool_calls)
    # prompt-JSON
    text, parsed = parse_prompt_json(resp.text)
    if parsed is None:
        return Plan(action="final", text=resp.text)
    if isinstance(parsed, list):
        return Plan(action="use_tool", text=text, tool_calls=parsed)
    if isinstance(parsed, dict):
        action = parsed.get("action", "")
        if action == "use_tool":
            c = _make(parsed)
            if c:
                return Plan(action="use_tool", text=text, tool_calls=[c])
            return Plan(action="final", text=resp.text)
        if action == "continue" or parsed.get("continue"):
            msg = parsed.get("content", parsed.get("message", ""))
            return Plan(action="continue", text=msg or text)
        if "message" in parsed:
            return Plan(action="final", text=parsed["message"])
        return Plan(action="final", text=text or resp.text)
    return Plan(action="final", text=resp.text)