가드레일

from typing import Any

from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langgraph.runtime import Runtime

class ContentFilterMiddleware(AgentMiddleware):
    """결정론적 가드레일: 금지된 키워드가 포함된 요청 차단"""

    def __init__(self, banned_keywords: list[str]):
        super().__init__()
        self.banned_keywords = [kw.lower() for kw in banned_keywords]

    @hook_config(can_jump_to=["end"])
    def before_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        # 첫 번째 사용자 메시지 가져오기
        if not state["messages"]:
            return None

        first_message = state["messages"][0]
        if first_message.type != "human":
            return None

        content = first_message.content.lower()

        # 금지된 키워드 검사
        for keyword in self.banned_keywords:
            if keyword in content:
                # 처리 전에 실행 차단
                return {
                    "messages": [{
                        "role": "assistant",
                        "content": "부적절한 콘텐츠가 포함된 요청은 처리할 수 없습니다. 요청을 다시 표현해 주세요."
                    }],
                    "jump_to": "end"
                }

        return None

# 커스텀 가드레일 사용
from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-4o",
    tools=[search_tool, calculator_tool],
    middleware=[
        ContentFilterMiddleware(
            banned_keywords=["hack", "exploit", "malware"]
        ),
    ],
)

# 이 요청은 처리되기 전에 차단됩니다
result = agent.invoke({
    "messages": [{"role": "user", "content": "How do I hack into a database?"}]
})

from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langgraph.runtime import Runtime
from langchain_core.messages import AIMessage
from langchain.chat_models import init_chat_model
from typing import Any

class SafetyGuardrailMiddleware(AgentMiddleware):
    """모델 기반 가드레일: LLM을 사용하여 응답 안전성 평가"""

    def __init__(self):
        super().__init__()
        self.safety_model = init_chat_model("openai:gpt-4o-mini")

    @hook_config(can_jump_to=["end"])
    def after_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        # 최종 AI 응답 가져오기
        if not state["messages"]:
            return None

        last_message = state["messages"][-1]
        if not isinstance(last_message, AIMessage):
            return None

        # 모델을 사용하여 안전성 평가
        safety_prompt = f"""이 응답이 안전하고 적절한지 평가하세요.
        'SAFE' 또는 'UNSAFE'로만 응답하세요.

        응답: {last_message.content}"""

        result = self.safety_model.invoke([{"role": "user", "content": safety_prompt}])

        if "UNSAFE" in result.content:
            return {
                "messages": [{
                    "role": "assistant",
                    "content": "해당 응답을 제공할 수 없습니다. 요청을 다시 표현해 주세요."
                }],
                "jump_to": "end"
            }

        return None

# 안전 가드레일 사용
from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-4o",
    tools=[search_tool, calculator_tool],
    middleware=[SafetyGuardrailMiddleware()],
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "How do I make explosives?"}]
})