🚀 10분 만에 끝내는 에이전트 거버넌스 시작 가이드¶

from agent_os.policies import PolicyEvaluator
from agent_os.policies.schema import (
    PolicyDocument, PolicyRule, PolicyCondition,
    PolicyAction, PolicyOperator, PolicyDefaults,
)

# --- 1단계: 에이전트 도구(tool) 정의 ---

def web_search(query: str) -> str:
    """웹 검색 도구 시뮬레이션"""
    return f"검색 결과: {query}"

def delete_file(path: str) -> str:
    """위험한 도구 — 정책에 의해 차단되어야 함"""
    return f"삭제됨: {path}"

TOOLS = {
    "web_search": web_search,
    "delete_file": delete_file,
}

# --- 2단계: 거버넌스 정책 정의 ---

policy = PolicyDocument(
    name="agent-safety",
    version="1.0",
    description="연구용 에이전트를 위한 안전 정책",
    defaults=PolicyDefaults(action=PolicyAction.ALLOW),
    rules=[
        PolicyRule(
            name="block-dangerous-tools",
            condition=PolicyCondition(
                field="tool_name",
                operator=PolicyOperator.IN,
                value=["delete_file", "shell_exec", "execute_code"],
            ),
            action=PolicyAction.DENY,
            message="안전 정책에 의해 도구가 차단되었습니다.",
            priority=100,
        ),
        PolicyRule(
            name="block-ssn-patterns",
            condition=PolicyCondition(
                field="input_text",
                operator=PolicyOperator.MATCHES,
                value=r"\b\d{3}-\d{2}-\d{4}\b",
            ),
            action=PolicyAction.DENY,
            message="개인정보에 해당하는 사회보장번호(SSN) 패턴 감지 — 차단됨",
            priority=90,
        ),
    ],
)

evaluator = PolicyEvaluator(policies=[policy])

# --- 3단계: 거버넌스 에이전트 구축 ---

class GovernedAgent:
    """모든 도구(Tool) 호출 전에 정책을 확인하는 간단한 에이전트"""

    def __init__(self, name, tools, evaluator):
        self.name = name
        self.tools = tools
        self.evaluator = evaluator

    def call_tool(self, tool_name: str, params: dict) -> str:
        # 실행관련 사전 정책 확인 (Pre-execution check)
        decision = self.evaluator.evaluate({
            "tool_name": tool_name,
            "input_text": str(params),
            "agent_id": self.name,
        })

        if not decision.allowed:
            print(f"  ✗ 차단됨: {decision.reason}")
            return f"[차단됨] {decision.reason}"

        # 도구 실행
        print(f"  ✓ 허용됨: {tool_name}")
        tool_fn = self.tools[tool_name]
        return tool_fn(**params)

# --- 4단계: 실행 ---

agent = GovernedAgent("research-agent", TOOLS, evaluator)

print("에이전트: 웹 검색 중...")
result = agent.call_tool("web_search", {"query": "최신 AI 거버넌스 뉴스"})
print(f"  결과: {result}\n")

print("에이전트: 파일 삭제 시도 중...")
result = agent.call_tool("delete_file", {"path": "/etc/passwd"})
print(f"  결과: {result}\n")

print("에이전트: 쿼리에 SSN을 포함하여 검색 중...")
result = agent.call_tool("web_search", {"query": "lookup 123-45-6789"})
print(f"  결과: {result}")

from agent_os.policies import PolicyEvaluator

evaluator = PolicyEvaluator()
evaluator.load_policies("policies/")   # 모든 *.yaml 파일을 로드합니다.

result = evaluator.evaluate({"tool_name": "web_search", "agent_id": "analyst-1"})
print(f"허용 여부: {result.allowed}")

from agent_os.policies import PolicyEvaluator

# 거버넌스 정책 로드
evaluator = PolicyEvaluator()
evaluator.load_policies("policies/")

# 프레임워크의 모든 도구 호출 전 평가 수행
decision = evaluator.evaluate({
    "agent_id": "langchain-agent-1",
    "tool_name": "web_search",
    "action": "tool_call",
})

if decision.allowed:
    # LangChain 도구 호출 진행
    result = your_langchain_agent.run(...)
else:
    print(f"차단됨: {decision.reason}")