What causes "MockAuditAPI.input_media is not implemented yet" in iflytek/astron-agent?

Avoid calling input_media with MockAuditAPI; either skip media auditing in that environment or wire a backend that implements it.

iflytek/astron-agent · warning · NotImplementedError

MockAuditAPI.input_media is not implemented yet

Error message

MockAuditAPI.input_media is not implemented yet

What it means

MockAuditAPI.input_media is an unimplemented stub for auditing user-submitted media (/audit/v3/aichat/inputMedia). The method only contains the commented endpoint path and a TODO, so every call raises NotImplementedError. It exists to fulfill the audit API interface; the mock backend never actually moderates input media.

Solutions

Avoid calling input_media with MockAuditAPI; either skip media auditing in that environment or wire a backend that implements it.
Implement the stub: POST the payload to /audit/v3/aichat/inputMedia and raise CustomException(CodeEnum.AUDIT_INPUT_ERROR) when data.action != ActionEnum.NONE, mirroring input_text.
Add a capability flag on audit API implementations (e.g. supports_media_audit) and have the pipeline check it before calling media audit methods.
For a pure mock, make it a no-op return instead of raising so dev/test multimodal flows run end to end.

Example fix

# before
async def input_media(self, text: str, **kwargs: Any) -> None:
    # path = f"/audit/v3/aichat/inputMedia"
    # TODO: To be implemented
    raise NotImplementedError("MockAuditAPI.input_media is not implemented yet")

# after
async def input_media(self, text: str, **kwargs: Any) -> None:
    payload = {"intention": "dialog", "content": text, **kwargs}
    resp = await self._post("/audit/v3/aichat/inputMedia", payload, kwargs.get("chat_app_id", ""), kwargs.get("uid", ""))
    if resp.get("data", {}).get("action") != ActionEnum.NONE:
        raise CustomException(CodeEnum.AUDIT_INPUT_ERROR, cause_error=f"Audit result abnormal: {resp}")

Defensive patterns

Strategy: fallback

Validate before calling

if audit_backend == "mock" and content_type == "media":
    logger.warning("MockAuditAPI does not implement input_media; skipping media audit")
    return

Type guard

def implements_input_media(api) -> bool:
    import inspect
    return "raise NotImplementedError" not in inspect.getsource(type(api).input_media)

Try / catch

try:
    await audit_api.input_media(media_content)
except NotImplementedError:
    logger.warning("input_media unimplemented for %s; media passed unaudited", type(audit_api).__name__)
except CustomException as e:
    handle_audit_rejection(e)

Prevention

In dev/test, treat mock media audit as pass-through rather than routing it through stub methods
Check implementation status of audit methods before enabling multimodal features
Prefer implementing stubs as explicit no-ops with warnings in mock backends so tests do not crash
Document which audit interface methods each backend implements and validate configuration against it at startup

When it happens

Trigger: Any call to MockAuditAPI.input_media(text, **kwargs) — e.g. a workflow stage auditing user-uploaded images/videos/documents — regardless of arguments.

Common situations: A multimodal chat flow (user uploads an image) routes the input through the mock audit implementation in dev/test, hitting the stub; developers assume the mock covers all interface methods and only discover the gap at runtime; an interface sweep/conformance test enumerates audit methods and triggers it.

Related errors

AI-assisted analysis of iflytek/astron-agent@5e758547a8 (2026-09-12). Data as JSON: /api/errors/f2e636e49569f460. Report an issue: GitHub.

Appendix: source

Thrown at core/workflow/infra/audit_system/audit_api/mock/mock_audit_api.py:248

        logging.info(f"\nMockAuditAPI.output_text resp: {resp}")
        if resp.get("data", {}).get("action") != ActionEnum.NONE:
            raise CustomException(
                CodeEnum.AUDIT_OUTPUT_ERROR,
                cause_error=f"Audit result abnormal: {resp}",
            )

    async def input_media(self, text: str, **kwargs: Any) -> None:
        """
        In LLM content security scenarios, filter, detect and identify user input text,
        images, videos, documents, etc., and process and respond accordingly based on security policies.
        :param text: Text content to be processed
        :param kwargs: Additional keyword arguments
        :return: None
        """
        # path = f"/audit/v3/aichat/inputMedia"

        # TODO: To be implemented
        raise NotImplementedError("MockAuditAPI.input_media is not implemented yet")

    async def output_media(self, text: str, **kwargs: Any) -> None:
        """
        In LLM content security scenarios, filter, detect and identify LLM output images,
        videos, audio, etc., and process and respond accordingly based on security policies.
        :param text: Text content to be processed
        :param kwargs: Additional keyword arguments
        :return: None
        """
        # path = f"/audit/v3/aichat/outputMedia"
        # TODO: To be implemented
        raise NotImplementedError("MockAuditAPI.output_media is not implemented yet")

    async def know_ref(self, text: str, **kwargs: Any) -> None:
        """
        In LLM content security scenarios, filter, detect and identify websites, knowledge bases
        and other data referenced during LLM responses, and process and respond accordingly based on security policies.
        :param text: Text content to be processed

View on GitHub (pinned to 5e758547a8)