forked from phoenix-oss/llama-stack-mirror
c1f7ba3aed
Splits the meta-reference safety implementation into three distinct providers: - inline::llama-guard - inline::prompt-guard - inline::code-scanner Note that this PR is a backward incompatible change to the llama stack server. I have added deprecation_error field to ProviderSpec -- the server reads it and immediately barfs. This is used to direct the user with a specific message on what action to perform. An automagical "config upgrade" is a bit too much work to implement right now :/ (Note that we will be gradually prefixing all inline providers with inline:: -- I am only doing this for this set of new providers because otherwise existing configuration files will break even more badly.)
58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
|
# All rights reserved.
|
|
#
|
|
# This source code is licensed under the terms described in the LICENSE file in
|
|
# the root directory of this source tree.
|
|
|
|
from typing import Any, Dict, List
|
|
|
|
from llama_models.llama3.api.datatypes import interleaved_text_media_as_str, Message
|
|
from termcolor import cprint
|
|
|
|
from .config import CodeScannerConfig
|
|
|
|
from llama_stack.apis.safety import * # noqa: F403
|
|
|
|
|
|
class MetaReferenceCodeScannerSafetyImpl(Safety):
|
|
def __init__(self, config: CodeScannerConfig, deps) -> None:
|
|
self.config = config
|
|
|
|
async def initialize(self) -> None:
|
|
pass
|
|
|
|
async def shutdown(self) -> None:
|
|
pass
|
|
|
|
async def register_shield(self, shield: Shield) -> None:
|
|
if shield.shield_type != ShieldType.code_scanner:
|
|
raise ValueError(f"Unsupported safety shield type: {shield.shield_type}")
|
|
|
|
async def run_shield(
|
|
self,
|
|
shield_id: str,
|
|
messages: List[Message],
|
|
params: Dict[str, Any] = None,
|
|
) -> RunShieldResponse:
|
|
shield = await self.shield_store.get_shield(shield_id)
|
|
if not shield:
|
|
raise ValueError(f"Shield {shield_id} not found")
|
|
|
|
from codeshield.cs import CodeShield
|
|
|
|
text = "\n".join([interleaved_text_media_as_str(m.content) for m in messages])
|
|
cprint(f"Running CodeScannerShield on {text[50:]}", color="magenta")
|
|
result = await CodeShield.scan_code(text)
|
|
|
|
violation = None
|
|
if result.is_insecure:
|
|
violation = SafetyViolation(
|
|
violation_level=(ViolationLevel.ERROR),
|
|
user_message="Sorry, I found security concerns in the code.",
|
|
metadata={
|
|
"violation_type": ",".join(
|
|
[issue.pattern_id for issue in result.issues_found]
|
|
)
|
|
},
|
|
)
|
|
return RunShieldResponse(violation=violation)
|