跳转至

analysis

feishu.attachments.analysis

附件分析:安全提取附件内容后,调用 OpenAI 兼容多模态模型返回结构化摘要。

analyze_attachment async

Python
analyze_attachment(data: bytes, file_metadata: Mapping[str, Any], *, extractor: AttachmentExtractor, openai_client: Any, model: str, prompt: str = DEFAULT_ATTACHMENT_ANALYSIS_PROMPT, instruction: str = DEFAULT_ATTACHMENT_ANALYSIS_INSTRUCTION, text_label: str = '附件文本') -> dict[str, Any]

安全提取附件内容,并调用 OpenAI 兼容多模态模型生成稳定结构化摘要。

模型调用或解析失败时,函数返回 {"ok": False, "kind": ..., "error": ...},而不是把异常抛给工具层。 extractor.extract 的超限或底层提取异常仍按提取器自己的语义处理。

源代码位于: feishu/attachments/analysis.py
Python
async def analyze_attachment(
    data: bytes,
    file_metadata: Mapping[str, Any],
    *,
    extractor: AttachmentExtractor,
    openai_client: Any,
    model: str,
    prompt: str = DEFAULT_ATTACHMENT_ANALYSIS_PROMPT,
    instruction: str = DEFAULT_ATTACHMENT_ANALYSIS_INSTRUCTION,
    text_label: str = "附件文本",
) -> dict[str, Any]:
    r"""
    安全提取附件内容,并调用 OpenAI 兼容多模态模型生成稳定结构化摘要。

    模型调用或解析失败时,函数返回 `{"ok": False, "kind": ..., "error": ...}`,而不是把异常抛给工具层。
    `extractor.extract` 的超限或底层提取异常仍按提取器自己的语义处理。
    """

    content = await extractor.extract(data, file_metadata=dict(file_metadata))
    parts = to_openai_content_parts(content, prompt=prompt, text_label=text_label)
    try:
        response = await openai_client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": instruction},
                {"role": "user", "content": parts},
            ],
            response_format={"type": "json_object"},
        )
    except Exception as exc:  # noqa: BLE001 - tool results should surface model-call failures in-band
        return {"ok": False, "kind": content.kind, "error": f"附件识别调用失败:{type(exc).__name__}"}
    raw = response.choices[0].message.content if response.choices else ""
    if not raw:
        return {"ok": False, "kind": "unknown", "error": "附件识别没有返回内容。"}
    try:
        parsed = json.loads(raw)
    except (ValueError, TypeError):
        return {"ok": False, "kind": "unknown", "error": "附件识别返回的不是合法 JSON。"}
    if not isinstance(parsed, dict):
        return {"ok": False, "kind": "unknown", "error": "附件识别返回了非对象 JSON。"}
    return normalize_attachment_analysis(parsed)

normalize_attachment_analysis

Python
normalize_attachment_analysis(result: dict[str, Any]) -> dict[str, Any]

校验并补齐附件分析模型返回的 JSON,使调用方拿到稳定字段。

源代码位于: feishu/attachments/analysis.py
Python
def normalize_attachment_analysis(result: dict[str, Any]) -> dict[str, Any]:
    r"""校验并补齐附件分析模型返回的 JSON,使调用方拿到稳定字段。"""

    if not isinstance(result.get("ok"), bool):
        return {
            "ok": False,
            "kind": str(result.get("kind") or "unknown").lower(),
            "error": "附件识别返回缺少布尔 ok 字段。",
        }
    if result["ok"] is False:
        result.setdefault("kind", "unknown")
        result.setdefault("error", "附件识别没有成功。")
        return result
    result["kind"] = str(result.get("kind") or "unknown").lower()
    fields = result.get("structured_fields")
    result["structured_fields"] = fields if isinstance(fields, dict) else {}
    if not isinstance(result.get("key_points"), list):
        result["key_points"] = []
    if not _has_useful_analysis(result):
        return {"ok": False, "kind": result["kind"], "error": "附件识别返回内容为空。"}
    if not result.get("summary"):
        result["summary"] = result.get("title") or "我已识别这个附件。"
    return result