geo/backend/app/agent_framework/tools/schema_tools.py

"""Schema 业务工具 - 将Schema建议服务注册为 FunctionTool"""

import copy
import json
import logging
from typing import Any

from agentkit.tools.function_tool import FunctionTool
from agentkit.tools.registry import ToolRegistry

logger = logging.getLogger(__name__)

SCHEMA_TEMPLATES = {
    "Organization": {
        "@context": "https://schema.org",
        "@type": "Organization",
        "name": "",
        "description": "",
        "url": "",
        "logo": "",
        "sameAs": [],
        "contactPoint": {
            "@type": "ContactPoint",
            "contactType": "customer service",
            "telephone": "",
        },
    },
    "Product": {
        "@context": "https://schema.org",
        "@type": "Product",
        "name": "",
        "description": "",
        "brand": {"@type": "Brand", "name": ""},
        "offers": {
            "@type": "Offer",
            "priceCurrency": "CNY",
            "availability": "https://schema.org/InStock",
        },
    },
    "FAQPage": {
        "@context": "https://schema.org",
        "@type": "FAQPage",
        "mainEntity": [
            {
                "@type": "Question",
                "name": "",
                "acceptedAnswer": {"@type": "Answer", "text": ""},
            }
        ],
    },
    "Article": {
        "@context": "https://schema.org",
        "@type": "Article",
        "headline": "",
        "description": "",
        "author": {"@type": "Organization", "name": ""},
        "datePublished": "",
        "image": "",
    },
    "LocalBusiness": {
        "@context": "https://schema.org",
        "@type": "LocalBusiness",
        "name": "",
        "address": {
            "@type": "PostalAddress",
            "streetAddress": "",
            "addressLocality": "",
            "addressRegion": "",
            "postalCode": "",
            "addressCountry": "CN",
        },
        "geo": {"@type": "GeoCoordinates", "latitude": "", "longitude": ""},
        "telephone": "",
        "openingHours": "",
    },
}

DIMENSION_SCHEMA_MAP = {
    "schema_marketing": ["Organization", "LocalBusiness"],
    "entity_clarity": ["Organization", "Product"],
    "citation_readiness": ["FAQPage", "Article"],
    "brand_visibility": ["Organization", "Product"],
    "local_seo": ["LocalBusiness"],
}

PRIORITY_THRESHOLD = {"high": 30.0, "medium": 60.0}
DIFFICULTY_MAP = {
    "Organization": "easy",
    "Product": "medium",
    "FAQPage": "medium",
    "Article": "easy",
    "LocalBusiness": "hard",
}


async def fill_schema_with_llm(
    schema_type: str,
    brand_info: dict | None = None,
    diagnosis_dimensions: dict | None = None,
) -> dict:
    """使用LLM填充Schema模板"""
    from app.services.llm import LLMFactory
    from app.agent_framework.prompts.schema_advisor import SCHEMA_ADVISOR_TEMPLATE
    from app.utils.json_extractor import extract_json

    brand_info = brand_info or {}
    diagnosis_dimensions = diagnosis_dimensions or {}

    template = SCHEMA_TEMPLATES.get(schema_type)
    if not template:
        return {"schema_type": schema_type, "json_ld_filled": None, "error": "Unknown schema type"}

    provider = LLMFactory.get_default()
    variables = {
        "brand_name": brand_info.get("name", ""),
        "brand_website": brand_info.get("website", ""),
        "brand_industry": brand_info.get("industry", ""),
        "schema_type": schema_type,
        "diagnosis_data": json.dumps(diagnosis_dimensions, ensure_ascii=False),
        "existing_schemas": "无",
    }
    messages = SCHEMA_ADVISOR_TEMPLATE.render(variables)

    try:
        response = await provider.chat(messages, temperature=0.3, max_tokens=2048)
        filled = json.loads(extract_json(response.content))
        return {"schema_type": schema_type, "json_ld_filled": filled}
    except Exception as e:
        logger.warning(f"LLM填充Schema {schema_type} 失败: {e}")
        return {"schema_type": schema_type, "json_ld_filled": None, "error": str(e)}


async def identify_missing_dimensions(
    diagnosis_data: dict,
    focus_dimensions: list[str] | None = None,
) -> dict:
    """识别Schema缺失维度"""
    dimensions = []
    dimension_scores = diagnosis_data.get("dimensions", {})
    for dim_name, dim_info in dimension_scores.items():
        if dim_name not in DIMENSION_SCHEMA_MAP:
            continue
        if focus_dimensions and dim_name not in focus_dimensions:
            continue
        score = dim_info.get("score", 0) if isinstance(dim_info, dict) else dim_info
        max_score = dim_info.get("max_score", 100) if isinstance(dim_info, dict) else 100
        percentage = (score / max_score * 100) if max_score > 0 else 0
        if percentage < 80:
            dimensions.append({
                "dimension": dim_name,
                "current_score": round(score, 2),
                "max_score": max_score,
                "percentage": round(percentage, 2),
            })
    return {"missing_dimensions": dimensions}


def register_schema_tools(registry: ToolRegistry) -> None:
    """注册所有Schema建议相关工具"""
    registry.register(
        FunctionTool(
            name="fill_schema_with_llm",
            description="使用LLM填充Schema JSON-LD模板",
            func=fill_schema_with_llm,
            tags=["schema", "llm"],
        )
    )
    registry.register(
        FunctionTool(
            name="identify_missing_dimensions",
            description="识别诊断数据中的Schema缺失维度",
            func=identify_missing_dimensions,
            tags=["schema", "diagnosis"],
        )
    )
    logger.info("Schema tools registered")