Agentic-AI-Risk-Mitigation/Janus
GitHub: Agentic-AI-Risk-Mitigation/Janus
Janus 通过对 LLM Agent 的工具调用实施细粒度策略执行，提供系统级安全防护以抵御间接提示注入等风险。
Stars: 13 | Forks: 2
# Janus
**通过对工具调用实施细粒度策略执行，为 LLM agent 提供系统级安全性。**
Janus 会拦截 LLM agent 发出的每一个工具调用，并在执行前根据安全策略对其进行验证——遵循最小权限原则。策略以 JSON 格式定义（或由 LLM 自动生成），并在运行时使用 JSON Schema 限制进行验证。
**[文档](https://agentic-ai-risk-mitigation.github.io/Janus/)** | **[GitHub](https://github.com/Agentic-AI-Risk-Mitigation/Janus)**
## 目录
- [Janus](#janus)
- [目录](#table-of-contents)
- [功能](#features)
- [安装](#installation)
- [快速开始](#quick-start)
- [`JanusAgent` 参数](#janusagent-parameters)
- [`JanusAgent` 方法](#janusagent-methods)
- [策略格式](#policy-format)
- [完整格式](#full-format)
- [简写格式（仅条件）](#shorthand-format-conditions-only)
- [策略规则字段](#policy-rule-fields)
- [条件 Schema](#condition-schemas)
- [评估逻辑](#evaluation-logic)
- [LLM 提供商](#llm-providers)
- [内置工具](#built-in-tools)
- [文件工具](#file-tools)
- [命令工具](#command-tools)
- [自定义工具](#custom-tools)
- [`ToolParam` 字段](#toolparam-fields)
- [LLM 生成的策略](#llm-generated-policies)
- [独立策略生成](#standalone-policy-generation)
- [策略细化](#policy-refinement)
- [框架适配器](#framework-adapters)
- [LangChain](#langchain)
- [深度 1 — 将 `ToolDef` 列表转换为受保护的 `StructuredTool` 列表](#depth-1--convert-tooldef-list-to-secured-structuredtool-list)
- [深度 2 — 包装现有的 LangChain 工具](#depth-2--wrap-existing-langchain-tools)
- [深度 3 — `JanusLangChainAgent`（一站式）](#depth-3--januslangchainagent-turnkey)
- [Google ADK (Gemini)](#google-adk-gemini)
- [深度 1 — 将 `ToolDef` 列表转换为 ADK 原生类型](#depth-1--convert-tooldef-list-to-adk-native-types)
- [深度 2 — `JanusADKAgent`（一站式）](#depth-2--janusadkagent-turnkey)
- [Claude Agent SDK (Claude Code)](#claude-agent-sdk-claude-code)
- [主要切入点 — `PreToolUse` 钩子](#primary-seam--pretooluse-hook)
- [替代切入点 — `can_use_tool` 回调](#alternative-seam--can_use_tool-callback)
- [双重保险 — `guard_tool_body()`](#belt-and-braces--guard_tool_body)
- [独立策略执行](#standalone-policy-enforcement)
- [运行时策略管理](#runtime-policy-management)
- [错误处理](#error-handling)
- [项目结构](#project-structure)
- [许可证](#license)
## 功能
- **细粒度策略执行** — 基于参数级别的 JSON Schema 条件允许或拒绝工具调用
- **最小权限原则** — 策略将 agent 限制在执行特定任务所需的最低权限内
- **多种策略来源** — 从 JSON 文件、Python 字典加载，或由 LLM 自动生成
- **LLM 生成的策略** — 从用户的查询中自动推断最小权限策略
- **策略细化** — 随着 agent 在任务执行中发现的信息，逐步收紧策略
- **基于图的能力呈现** — 通过 `janus/policy/pde/` 中集成的 PDE 引擎，实现由 SpiceDB 支持的 ReBAC 和运行时污点追踪（IPI 防御）
- **内置工具** — 开箱即用的文件系统和命令执行工具，并提供工作区沙箱隔离
- **自定义工具** — 使用 `ToolDef` / `ToolParam` 定义您自己的工具；Janus 会自动为它们提供保护
- **10 多个 LLM 提供商** — OpenAI, Anthropic, Google Gemini, Azure OpenAI, AWS Bedrock, Ollama, vLLM, Together AI, OpenRouter
- **框架适配器** — 将 Janus 的执行机制插入 LangChain、Google ADK 和 Claude Agent SDK (Claude Code) agent 中
- **独立执行器** — 在任何 agent 框架中独立使用 `PolicyEnforcer`
- **三种回退操作** — 抛出 `PolicyViolation`、调用 `sys.exit` 或以交互方式提示用户
- **工作区隔离** — 文件工具被限制在指定目录内；拒绝路径遍历尝试
## 安装
要求 Python ≥ 3.10。推荐使用 [uv](https://docs.astral.sh/uv/) 作为包管理器。
**核心库（独立的 `PolicyEnforcer` + OpenAI / OpenAI 兼容的提供商）：**
```
uv add janus-guard
```
核心安装不依赖 SpiceDB/`authzed` 或 Web 服务器，因此
`from janus.policy import PolicyEnforcer` 可以直接单独使用，
作为一个轻量级的工具调用拦截门。
**安装可选的提供商/附加功能：**
```
uv add "janus-guard[pde]" # SpiceDB-backed ReBAC + taint tracking (PDEEnforcer)
uv add "janus-guard[server]" # Example web demo UI (FastAPI + uvicorn)
uv add "janus-guard[anthropic]" # Anthropic Claude
uv add "janus-guard[google]" # Google Gemini
uv add "janus-guard[bedrock]" # AWS Bedrock
uv add "janus-guard[langchain]" # LangChain adapter
uv add "janus-guard[adk]" # Google ADK adapter
uv add "janus-guard[claude]" # Claude Agent SDK (Claude Code) adapter
uv add "janus-guard[all]" # Everything
```
**用于开发环境：**
```
uv add "janus-guard[dev]" # pytest, ruff, mypy
```
**从源码安装：**
```
git clone https://github.com/Agentic-AI-Risk-Mitigation/Janus.git
cd janus
uv pip install -e .
```
## 快速开始
```
from janus import JanusAgent
agent = JanusAgent(
model="openai/gpt-4o",
api_key="sk-...", # or set OPENAI_API_KEY env var
use_builtin_tools=True,
policy="policies.json",
system_prompt="You are a helpful coding assistant.",
)
response = agent.run("List the Python files in the project.")
print(response)
```
### `JanusAgent` 参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| `model` | `str` | — | 模型字符串：`"/"`（例如 `"openai/gpt-4o"`） |
| `system_prompt` | `str` | `"You are a helpful assistant."` | LLM 的系统消息 |
| `tools` | `list[ToolDef] \| None` | `None` | 要注册的自定义工具定义 |
| `use_builtin_tools` | `bool` | `True` | 注册内置的文件和命令工具 |
| `policy` | `str \| Path \| dict \| None` | `None` | 策略来源（文件路径、字典、`"generate"` 或 `None`） |
| `policy_model` | `str \| None` | `"gpt-4o-2024-08-06"` | 用于基于 LLM 生成策略的模型 |
| `policy_engine`| `str` | `"janus"` | 要使用的执行引擎（`"janus"` 或 `"pde"`） |
| `agent_id` | `str` | `"coding_agent"` | 用于 `pde` 污点和 ACL 检查的 agent 身份标识 |
| `api_key` | `str \| None` | `None` | API 密钥（默认回退到提供商的环境变量） |
| `workspace` | `str \| Path \| None` | `cwd` | 文件系统工具的根目录 |
| `max_tool_iterations` | `int` | `10` | 每次 `run()` 调用的最大工具调用循环次数 |
| `temperature` | `float` | `0.1` | 采样温度 |
| `log_level` | `str \| None` | `"INFO"` | 日志级别（`"DEBUG"`, `"INFO"`, `"WARNING"`） |
### `JanusAgent` 方法
| 方法 | 描述 |
|---|---|
| `run(user_input)` | 运行 agent 并返回最终的文本响应 |
| `clear_history()` | 重置对话历史记录（保留策略和工具） |
| `set_policy(policy)` | 在运行时加载或替换安全策略 |
| `get_policy()` | 返回当前策略字典 |
| `save_policy(path)` | 将当前策略持久化到 JSON 文件中 |
| `allow_tools(tools)` | 无条件允许列出的工具（最高优先级） |
| `block_tools(tools)` | 无条件阻止列出的工具（最高优先级） |
| `add_tool(tool)` | 在运行时注册附加工具 |
| `remove_tool(name)` | 按名称注销工具 |
| `list_tools()` | 返回所有已注册工具的名称 |
| `update_taint(risk)` | 单调更新会话污点风险（仅适用于 PDE 引擎） |
## 策略格式
策略是 JSON 文档，将工具名称映射到规则列表。每条规则指定是允许还是拒绝工具调用，并可以包含参数级别的限制。
### 完整格式
```
{
"read_file": [
{
"priority": 1,
"effect": 0,
"conditions": {
"file_path": {
"type": "string",
"pattern": "^reports/.*\\.csv$"
}
},
"fallback": 0
}
],
"run_command": [
{
"priority": 1,
"effect": 1,
"conditions": {},
"fallback": 0
}
]
}
```
### 简写格式（仅条件）
当您只需要限制参数值时，简写格式会跳过 `priority`、`effect` 和 `fallback`（默认为允许，优先级为 1，违规时抛出异常）：
```
{
"read_file": {
"file_path": { "type": "string", "pattern": "^data/.*" }
}
}
```
### 策略规则字段
| 字段 | 类型 | 描述 |
|---|---|---|
| `priority` | `int` | 评估顺序 — 值越小越先执行 |
| `effect` | `int` | `0` = 允许，`1` = 拒绝 |
| `conditions` | `dict` | 以参数名称为键的 JSON Schema 限制 |
| `fallback` | `int` | `0` = 抛出 `PolicyViolation`，`1` = `sys.exit(1)`，`2` = 询问用户 |
### 条件 Schema
条件遵循 JSON Schema 语法。常见模式：
```
{ "type": "string", "pattern": "^/safe/path/.*" }
{ "type": "string", "enum": ["ls", "pwd", "cat"] }
{ "type": "integer", "minimum": 0, "maximum": 100 }
{ "type": "array", "items": { "type": "string" } }
```
### 评估逻辑
1. 工具的规则按 `priority` 升序评估。
2. **允许规则**（`effect=0`）：如果所有条件都通过 → 立即允许使用该工具。
3. **拒绝规则**（`effect=1`）：如果所有条件都匹配 → 使用配置的 `fallback` 阻止该工具。
4. 如果没有规则匹配 → **默认阻止**该工具。
5. 加载策略后，未在策略中列出的工具将被阻止。
## LLM 提供商
为 `model` 参数使用 `"/"` 格式：
| 提供商 | 模型字符串示例 | 环境变量 |
|---|---|---|
| **OpenAI** | `openai/gpt-4o`, `openai/gpt-4o-mini` | `OPENAI_API_KEY` |
| **Anthropic** | `anthropic/claude-3-5-sonnet-20241022` | `ANTHROPIC_API_KEY` |
| **Google Gemini** | `google/gemini-2.0-flash`, `gemini/gemini-1.5-pro` | `GOOGLE_API_KEY` / `GEMINI_API_KEY` |
| **Azure OpenAI** | `azure/` | `AZURE_OPENAI_API_KEY` |
| **AWS Bedrock** | `bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0` | AWS 凭证 |
| **Ollama**（本地） | `ollama/llama3.2`, `ollama/mistral` | — |
| **vLLM**（本地） | `vllm/meta-llama/Llama-3.3-70B-Instruct` | `VLLM_BASE_URL` |
| **Together AI** | `together/meta-llama/Llama-3-70b-chat-hf` | `TOGETHER_API_KEY` |
| **OpenRouter** | `openrouter/anthropic/claude-3.5-sonnet` | `OPENROUTER_API_KEY` |
**示例 — Anthropic：**
```
agent = JanusAgent(
model="anthropic/claude-3-5-sonnet-20241022",
policy="policies.json",
)
```
**示例 — Ollama（本地）：**
```
agent = JanusAgent(
model="ollama/llama3.2",
policy="policies.json",
)
```
**示例 — Azure OpenAI：**
```
agent = JanusAgent(
model="azure/my-deployment",
api_key="...",
base_url="https://my-resource.openai.azure.com/",
api_version="2024-02-01",
policy="policies.json",
)
```
## 内置工具
当 `use_builtin_tools=True`（默认值）时，Janus 会自动注册以下工具：
### 文件工具
| 工具 | 描述 | 参数 |
|---|---|---|
| `read_file` | 读取文件的全部内容 | `file_path: str` |
| `write_file` | 创建或覆盖文件 | `file_path: str`, `content: str` |
| `edit_file` | 替换文件中的唯一字符串 | `file_path: str`, `old_string: str`, `new_string: str` |
| `list_directory` | 列出目录内容 | `path: str`（默认：工作区根目录） |
所有文件工具都被限制在 `workspace` 目录内 — 尝试访问工作区外的路径将被拒绝。
### 命令工具
| 工具 | 描述 | 参数 |
|---|---|---|
| `run_command` | 在工作区中执行 shell 命令 | `command: str`, `timeout: int`（默认：60） |
| `fetch_url` | 通过 HTTP GET 从 URL 获取内容 | `url: str` |
**设置工作区：**
```
agent = JanusAgent(
model="openai/gpt-4o",
workspace="./my_project", # file tools are sandboxed here
policy="policies.json",
)
```
## 自定义工具
使用 `ToolDef` 和 `ToolParam` 定义您自己的工具：
```
from janus import JanusAgent, ToolDef, ToolParam
def search_database(query: str, limit: int = 10) -> str:
# your implementation
return f"Results for '{query}' (limit={limit})"
agent = JanusAgent(
model="openai/gpt-4o",
use_builtin_tools=False, # disable built-ins if not needed
tools=[
ToolDef(
name="search_database",
description="Search the internal database for records matching a query.",
params=[
ToolParam("query", "string", "Search query string"),
ToolParam("limit", "integer", "Maximum number of results", required=False, default=10),
],
handler=search_database,
)
],
policy={
"search_database": [
{
"priority": 1,
"effect": 0,
"conditions": {
"limit": {"type": "integer", "maximum": 50}
},
"fallback": 0,
}
]
},
)
response = agent.run("Find records about renewable energy.")
```
### `ToolParam` 字段
| 字段 | 类型 | 描述 |
|---|---|---|
| `name` | `str` | 参数名称（必须与处理程序的 kwarg匹配） |
| `type` | `str` | JSON Schema 类型：`"string"`, `"integer"`, `"number"`, `"boolean"`, `"array"`, `"object"` |
| `description` | `str` | 向 LLM 展示的描述 |
| `required` | `bool` | 是否必须提供该参数（默认：`True`） |
| `default` | `Any` | 当 `required=False` 时的默认值 |
| `enum` | `list \| None` | 限制为固定的允许值集合 |
## LLM 生成的策略
Janus 可以通过让 LLM 推断特定用户查询需要哪些工具和限制，自动生成最小权限策略：
```
agent = JanusAgent(
model="openai/gpt-4o",
policy="generate", # generate on first run()
policy_model="openai/gpt-4o-2024-08-06", # model used for generation
)
# 策略在首次调用时基于此 query 生成
response = agent.run("Read the file sales_2024.csv and summarize the totals.")
# 检查并保存生成的 policy
print(agent.get_policy())
agent.save_policy("generated_policy.json")
```
### 独立策略生成
无需完整 agent，直接使用 `generate_policy`：
```
from janus import generate_policy, save_policy
tools = [
{"name": "read_file", "description": "Read a file", "args": {...}},
{"name": "run_command", "description": "Run a shell command", "args": {...}},
]
policy = generate_policy(
query="Read the quarterly report and list the top 5 expenses.",
tools=tools,
model="gpt-4o-2024-08-06",
manual_confirm=True, # ask before applying
)
save_policy(policy, "policies.json")
```
### 策略细化
在执行信息收集类的工具调用后，利用发现的值收紧策略：
```
from janus import refine_policy
updated = refine_policy(
query="Send the invoice to the customer.",
tools=tools,
tool_call_params={"file_path": "invoices/inv_001.txt"},
tool_call_result="Customer email: alice@example.com",
current_policy=current_policy,
model="gpt-4o-2024-08-06",
manual_confirm=True,
)
```
## 框架适配器
### LangChain
提供三种集成深度。
**安装：**
```
uv add "janus-guard[langchain]"
```
#### 深度 1 — 将 `ToolDef` 列表转换为受保护的 `StructuredTool` 列表
适用于您构建自己的 LangChain agent，但希望使用 Janus 保护工具的情况：
```
from janus.adapters.langchain import secure_langchain_tools
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_openai import ChatOpenAI
lc_tools = secure_langchain_tools(my_janus_tools, "policies.json")
llm = ChatOpenAI(model="gpt-4o")
agent = create_tool_calling_agent(llm, lc_tools, prompt)
executor = AgentExecutor(agent=agent, tools=lc_tools)
```
#### 深度 2 — 包装现有的 LangChain 工具
适用于您已有 LangChain 代码库，并希望加装 Janus 执行机制的情况：
```
from janus.adapters.langchain import wrap_langchain_tools
# existing_tools 是你现有的 LangChain BaseTool 对象列表
existing_tools = wrap_langchain_tools(existing_tools, "policies.json")
# 像往常一样传递给你现有的 AgentExecutor
```
#### 深度 3 — `JanusLangChainAgent`（一站式）
```
from janus.adapters.langchain import JanusLangChainAgent
agent = JanusLangChainAgent(
model="openai/gpt-4o",
tools=my_janus_tools,
policy="policies.json",
system_prompt="You are a helpful assistant.",
max_iterations=10,
)
response = agent.run("Summarize the quarterly results.")
agent.clear_history()
```
### Google ADK (Gemini)
提供两种集成深度。
**安装：**
```
uv add "janus-guard[adk]"
```
#### 深度 1 — 将 `ToolDef` 列表转换为 ADK 原生类型
适用于您自行管理 Gemini 聊天循环的情况：
```
from janus.adapters.adk import secure_adk_tools
from google import genai
from google.genai import types
declarations, handlers = secure_adk_tools(my_janus_tools, "policies.json")
client = genai.Client(api_key="...")
config = types.GenerateContentConfig(
tools=[types.Tool(function_declarations=declarations)],
system_instruction="You are helpful.",
automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True),
)
chat = client.chats.create(model="gemini-2.0-flash", config=config)
response = chat.send_message("List the files in the project.")
while response.function_calls:
fc = response.function_calls[0]
result = handlers[fc.name](**dict(fc.args))
response = chat.send_message(
types.Part.from_function_response(fc.name, {"result": result})
)
print(response.text)
```
#### 深度 2 — `JanusADKAgent`（一站式）
```
from janus.adapters.adk import JanusADKAgent
agent = JanusADKAgent(
model="gemini-2.0-flash",
tools=my_janus_tools,
policy="policies.json",
system_prompt="You are a helpful assistant.",
max_tool_iterations=10,
)
response = agent.run("What Python files are in the workspace?")
agent.clear_history()
```
### Claude Agent SDK (Claude Code)
[Claude Agent SDK](https://github.com/anthropics/claude-agent-sdk-python) (`claude-agent-sdk`) 在 `claude` CLI 子进程中运行工具循环，因此 Janus 无法像控制自身循环时那样直接介入调用路径。相反，适配器将 `PolicyEnforcer` 插入到 SDK 的预执行切入点中，从而确保在工具执行*之前*依然会进行策略决策。
**安装**（还需要确保 `PATH` 中包含 `claude` CLI）：
```
uv add "janus-guard[claude]"
```
#### 主要切入点 — `PreToolUse` 钩子
使用 `janus_hooks()` 获取现成的 `hooks=` 字典，供 `ClaudeAgentOptions` 使用。`PreToolUse` 钩子会在**每一次**工具调用时触发，甚至包括 `allowed_tools` 中的工具以及处于 `permission_mode="dontAsk"` 模式下的调用 —— 因此策略不会被绕过：
```
from claude_agent_sdk import ClaudeAgentOptions, query, create_sdk_mcp_server
from janus.adapters.claude_agent_sdk import janus_hooks
# Policy 以纯 tool 名称为键；adapter 会去除 mcp____ 前缀。
TOOL_POLICY = {
"web_search": [(1, 0, {"query": {"type": "string", "maxLength": 400}}, 0)],
"fetch_page": [(1, 0, {"url": my_ssrf_check}, 0)], # callable condition
}
options = ClaudeAgentOptions(
mcp_servers={"research": create_sdk_mcp_server(name="research", tools=[...])},
allowed_tools=["mcp__research__web_search", "mcp__research__fetch_page"],
permission_mode="dontAsk",
output_format={"type": "json_schema", "schema": SCHEMA},
# required_args: belt-and-braces presence check (non-empty) on top of the
# core enforcer's strict condition semantics.
hooks=janus_hooks(TOOL_POLICY, required_args={"fetch_page": ["url"]}),
)
```
当策略拒绝时，钩子会返回一个 `permissionDecision: "deny"`，其拒绝原因会反馈给模型，以便 agent 能够自行调整而不是崩溃。适配器为您处理了以下两种 SDK 特定的行为：
- **`StructuredOutput` 透传** — 当您设置 `output_format` 时，SDK 会通过一个内部 `StructuredOutput` 工具调用来交付最终结果，该调用也会被钩子拦截。默认拒绝的策略会阻止它并导致结果为空。此类 SDK 内部工具会被直接放行（可通过 `passthrough_tools` 配置）。
- **名称映射** — 进程内服务器名为 `research` 上的 `@tool("fetch", …)` 会被当作 `mcp__research__fetch` 调用。适配器会将其映射回原始的策略键名（`fetch`）。如果您的策略使用的是带有完整前缀的名称，请传入 `resolve_name=lambda n: n`。
#### 替代切入点 — `can_use_tool` 回调
`make_can_use_tool()` 构建一个 `can_use_tool` 回调。**它可被绕过**：在回调运行之前，SDK 会自动批准任何整工具级别的 `allowed_tools` 条目（以及处于 `permission_mode="bypassPermissions"` 下的所有内容），并发出 `CanUseToolShadowedWarning`。建议优先使用钩子；仅在没有其他内容遮蔽它的情况下才使用此回调。
#### 双重保险 — `guard_tool_body()`
包装进程内的 `@tool` 主体，使得策略执行也会在执行阶段生效，独立于任何 SDK 的权限切入点：
```
from janus.adapters.claude_agent_sdk import guard_tool_body
guarded = guard_tool_body("fetch_page", my_async_body, TOOL_POLICY,
required_args={"fetch_page": ["url"]})
```
## 独立策略执行
`PolicyEnforcer` 可以在任何 agent 框架中独立使用 — 只需在执行任何工具之前调用 `enforce()` 即可：
```
from janus.policy import PolicyEnforcer
from janus import PolicyViolation
enforcer = PolicyEnforcer()
enforcer.load("policies.json")
# 在执行 tool 之前：
try:
enforcer.enforce("read_file", {"file_path": "data/report.csv"})
result = read_file("data/report.csv") # proceeds normally
except PolicyViolation as exc:
print(f"Blocked: {exc.reason}")
# 通过编程方式更新 policy：
enforcer.allow_tools(["list_directory"]) # unconditional allow
enforcer.block_tools(["run_command"]) # unconditional deny
enforcer.update({"write_file": [(1, 0, {}, 0)]}) # merge additional rules
```
## 运行时策略管理
在创建 agent 之后，可以对策略进行检查和修改：
```
# 加载新的 policy
agent.set_policy("new_policies.json")
agent.set_policy({"read_file": [{"priority": 1, "effect": 0, "conditions": {}, "fallback": 0}]})
# 检查当前的 policy
print(agent.get_policy())
# 保存到磁盘
agent.save_policy("saved_policy.json")
# 在运行时允许 / 阻止特定的 tool
agent.allow_tools(["list_directory", "read_file"])
agent.block_tools(["run_command"])
# 管理 tool
agent.add_tool(my_new_tool)
agent.remove_tool("old_tool")
print(agent.list_tools())
```
## 错误处理
所有 Janus 异常都继承自 `JanusError`：
```
from janus import (
JanusError,
PolicyViolation, # tool call blocked by policy
ArgumentValidationError, # argument failed JSON Schema check
PolicyLoadError, # policy file not found or invalid JSON
PolicyGenerationError, # LLM-based generation failed
ToolNotFoundError, # tool name not registered
ProviderError, # LLM provider error
)
try:
response = agent.run("Delete all log files.")
except PolicyViolation as exc:
print(f"Tool '{exc.tool_name}' was blocked.")
print(f"Reason: {exc.reason}")
print(f"Arguments: {exc.arguments}")
except JanusError as exc:
print(f"Janus error: {exc}")
```
## 项目结构
```
janus/
├── agent.py # JanusAgent — main entry point
├── exceptions.py # Custom exception classes
├── logger.py # Structured logging utilities
├── __init__.py # Public API re-exports
│
├── llm/
│ ├── base.py # BaseLLMProvider interface
│ ├── runner.py # LLMRunner — conversation loop
│ ├── response_types.py # Provider response types
│ └── providers/
│ ├── openai_provider.py
│ ├── anthropic_provider.py
│ ├── google_provider.py
│ ├── azure_provider.py
│ ├── bedrock_provider.py
│ ├── ollama_provider.py
│ ├── vllm_provider.py
│ ├── together_provider.py
│ └── openrouter_provider.py
│
├── policy/
│ ├── enforcer.py # PolicyEnforcer — JSON Schema rule engine
│ ├── pde_enforcer.py # PDEEnforcer — adapter for SpiceDB/taint engine
│ ├── pde/ # SpiceDB-backed ReBAC + taint (PDE)
│ │ ├── config.py # SCHEMA, TOOL_TAINT_LIMIT, RISK_TO_TAINT
│ │ ├── interceptor.py # GraphInterceptor — taint gate + SpiceDB ACL
│ │ ├── discovery.py # GraphDiscoveryEngine
│ │ └── bootstrap.py # make_client, bootstrap, Session, allow_tool
│ ├── generator.py # LLM-based policy generation & refinement
│ ├── loader.py # JSON parsing and policy persistence
│ └── validator.py # JSON Schema argument validation
│
├── tools/
│ ├── base.py # ToolDef, ToolParam dataclasses
│ ├── registry.py # ToolRegistry — manages registered tools
│ └── builtin/
│ ├── file_tools.py # read_file, write_file, edit_file, list_directory
│ └── command_tools.py # run_command, fetch_url
│
└── adapters/
├── _base.py # Shared adapter utilities
├── langchain.py # LangChain integration
├── adk.py # Google ADK (Gemini) integration
└── claude_agent_sdk.py # Claude Agent SDK (Claude Code) integration
examples/ # Demo scenario framework + FastAPI web app + docker-compose.yml for SpiceDB
```
## 运行演示 Web 应用
演示 Web UI 位于 `examples/app.py` 中。如果您希望通过一条命令启动 FastAPI 应用，而无需记住 uvicorn 的导入路径，可以使用仓库根目录下的辅助脚本。
```
# 从 repo checkout 安装 demo/example 依赖
uv sync --extra langchain --extra dev
# 或者：uv sync --extra all --extra dev
# 在 http://127.0.0.1:8000 启动 web UI
./scripts/run_demo_webapp.sh
# 启动 web UI 并为 PDE 场景运行 SpiceDB
./scripts/run_demo_webapp.sh --with-spicedb
```
该脚本还支持 `--host`、`--port` 和 `--no-reload` 参数。
## 运行 PDE (SpiceDB) 演示
PDE 引擎需要运行中的 SpiceDB 实例。从源码检出版本时，请先安装演示依赖；示例运行程序会导入 LangChain。然后使用演示应用的 Docker 设置：
**前提条件：** 已安装并运行 Docker。
```
# 从 repo checkout 安装 demo/example 依赖
uv sync --extra langchain --extra dev
# 或者：uv sync --extra all --extra dev
# 不使用 SpiceDB 的可选 smoke test
uv run python -m examples.run coding_agent_poisoned_readme --protected
# 启动 SpiceDB（从项目根目录）
cd examples && docker compose up -d && cd ..
# 运行带有 PDE 的 coding-agent taint cascade — 需要 SpiceDB
uv run python -m examples.run coding_agent_taint_cascade --protected
# 完成后停止 SpiceDB
cd examples && docker compose stop && cd ..
```
**Coding-agent 污点级联测试涉及的内容：**
- 由 ACL 授权的工具（只读、开发者角色）在低污点级别下可通过
- 当 `current_taint > TOOL_TAINT_LIMIT[tool]` 时，Python 污点门会阻止工具
- 在执行 `fetch_url`（中等风险）后，污点级别上升；`git_push`（限制为 20）会被阻止
- 完整的 IPI 场景：agent 读取外部内容 → 污点增加 → 危险工具被阻止
## 许可证
MIT
标签：LangChain, Lerna, LLM Agent, Streamlit, 人工智能, 安全防护, 提示词注入防御, 用户模式Hook绕过, 策略执行, 访问控制, 轻量级, 逆向工具