``` ███████████ █████ ░░███░░░░░███ ░░███ ░███ ░███ ████████ ██████ █████████████ ████████ ███████ ░██████████ ░░███░░███ ███░░███░░███░░███░░███ ░░███░░███░░░███░ ░███░░░░░███ ░███ ░░░ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ░███ ███ ███████████ █████ ░░██████ █████░███ █████ ░███████ ░░█████ ░░░░░░░░░░░ ░░░░░ ░░░░░░ ░░░░░ ░░░ ░░░░░ ░███░░░ ░░░░░ ░███ █████ ░░░░░ ```

目录

为什么选择 Brompt？ · 架构 · 安全 · 功能 · 布局 · 配置 · 快速开始 · Providers · API · 高级功能 · CI/CD · 生产环境 · 许可证

## 为什么选择 Brompt？ **Brompt 是合规级别的 LLM 中间件 —— 而不是另一个 provider 抽象层。** LangChain 和 LiteLLM 在广度上竞争（100 多个 provider，庞大的社区）。Brompt 并不这么做。Brompt 在信任度上竞争：**签名的执行回执、哈希链审计跟踪，以及针对受监管行业（法律、金融、政府）的 policy-as-code 治理**。 它们负责路由。Brompt 负责证明。 | 为什么不选 LangChain / LiteLLM | Brompt 的回答 | |---|---| | 没有不可否认性 —— 响应事后无法验证 | **签名的执行回执**（每个 `ExecutionResult` 使用 HMAC/哈希链）—— 每个响应都被证明是真实的且防篡改 | | 审计只是事后的补充（文本日志） | **哈希链审计日志** —— `AuditLog.verify()` 会重放链并检测任何篡改 | | 没有确定性重放 —— 更改模型会悄悄改变行为 | `PromptClient.replay(id, model=X)` —— 在不同的模型上重新运行相同的 prompt，并对结果进行 diff 以检测 **prompt drift** | | 安全过滤器要么缺失，要么不透明 | **纵深防御**：regex 黑名单 + LLM 语义分类器 + 输出脱敏，全部记录在审计链中 | | 策略被硬编码或根本不存在 | **Policy-as-code**：在 prompt 到达任何 provider 之前，评估每个租户的 YAML 策略（基于 `caller_id` 允许/拒绝） | | 灰色地带没有人工干预 | **可配置的置信度阈值** —— 低于 0.4 = 通过，0.4-0.7 = 暂停等待人工审查，高于 0.7 = 阻止 | | 供应商锁定 —— 切换模型意味着重写 prompt | **与 provider 无关的 pipeline** —— 在 7 个 provider 之间直接替换；签名的回执证明了提交的内容，而不管使用什么模型 | ### Brompt 的工作原理 Brompt 位于 provider 层 **之上**。你甚至可以在 Brompt 下面使用 LiteLLM 作为 provider —— 合规层保持不变。 ``` Client App │ ▼ ┌─────────────────────────────────────────┐ │ BROMPT COMPLIANCE LAYER │ │ ├ Rate Limiter ├ Security Engine │ │ ├ Policy-as-Code ├ Classifier (LLM) │ │ ├ Circuit Breaker ├ Model Router │ │ ├ Token Optimizer ├ Memory Manager │ │ └ Audit Log (hash-chained, signed) │ └─────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────┐ │ PROVIDER (Anthropic / OpenAI / …) │ └─────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────┐ │ Execution Receipt (signed + logged) │ └─────────────────────────────────────────┘ ``` ## 1. 系统架构概览 **Brompt 引擎** 解决了现代 LLM 智能体的根本限制：非确定性执行路径和线性上下文漂移（`O(N)` token 增长）。它作为一个合规中间件，位于宿主应用程序环境和上游 model endpoint 之间。 **性能提示：** 安全 pipeline 在输入长度上是 `O(N)` 的，但由于有 64KB 的上限，最坏情况下的运行时间是有限的。在实践中，LLM provider 的调用（1–10秒）占据了 2-3 个数量级的端到端延迟，因此 pipeline 中输入大小的差异可以忽略不计。 **关于模式匹配安全层的提示：** `SecurityEngine.sanitize` 是一个 regex 黑名单。它能捕获简单的、字面意义的注入 尝试（包括阿拉伯语变体），但对于被改写、编码或以其他方式混淆的 prompt 注入来说，它不是一个强大的防御手段 — 没有哪个黑名单能做到这一点。将其视为一个低成本的第一道过滤器，而不是保证，并在 应用程序端将其与输出净化器和最小权限工具/权限 设计配合使用。 ### 请求流 ``` graph TD A["Client App"] --> B["Rate Limiter"] B --> C["Security Engine"] C --> D{"Classifier"} D -->|opt-in| E["LLM Semantic"] D -->|skip| F["Memory Manager"] E --> F F --> G["Circuit Breaker"] G --> H["Model Router"] H --> I["Provider"] I --> K["Output Sanitizer"] K --> L["Audit Log"] L --> M["Client App"] G -->|open| N["Fallback"] N --> L ``` ### 核心架构支柱 | 支柱 | 描述 | |---|---| | **签名的执行回执** | 每个 `ExecutionResult` 都通过阶段标志 + 时间戳进行 HMAC 签名 —— 这是响应通过了完整合规 pipeline 的不可抵赖证明 | | **哈希链审计日志** | 带有 `AuditLog.verify()` 的 SHA-256 仅追加链 —— 事后检测任何篡改。每个安全事件都记录有一个不可变的链接 | | **确定性重放** | `PromptClient.replay(id, model=X)` 在不同的模型上重新运行相同的消息，并对输出进行 diff —— 在升级模型时捕获 **prompt drift** | | **Policy-as-Code** | 在 prompt 到达任何 provider 之前评估每个租户的 YAML 策略（基于 `caller_id` 允许/拒绝）。无需为每个客户更改代码 | | **Human-in-the-Loop** | 针对灰色地带输入的可配置置信度阈值（通过 / 暂停审查 / 阻止）—— 保证在敏感部署中有人工监督 | | **纵深防御安全** | 多层防御：输入规范化（NFKC、零宽字符、base64、leetspeak）、regex 黑名单、LLM 语义分类器、输出脱敏 | | **有界状态管理** | 线程安全的 `deque(maxlen=max_turns)` 对话历史记录 —— 跨轮次没有原始消息累积 | | **结构化类型契约** | Pydantic v2 schema 验证保证为下游工具提供类型化的、程序化的输出 | | **可插拔 Provider 系统** | 7 个 LLM provider：Anthropic、OpenAI、Ollama、Gemini、Mistral、Azure OpenAI、LM Studio —— 同步 + 异步 | | **Hooks/Middleware** | Pipeline hooks（日志、计时、验证、审计、RateLimit、安全），在执行前后运行 | | **断路器** | CLOSED/OPEN/HALF_OPEN 状态机，保护 provider 免受级联故障影响；支持 fallback | | **模型路由器** | 启发式复杂度分类器（字数、代码/数学标记、分析关键词），具有 4 种策略：CHEAPEST、FASTEST、BEST_QUALITY、FALLBACK | | **定价与优化** | 每个模型的成本估算、基于压缩的 token 优化、节省跟踪 | | **CLI (Typer)** | 8 个命令：`chat`、`run`、`history`、`audit`、`status`、`templates`、`config`、`clear` | | **Web UI** | 基于 Streamlit 的界面，带有聊天面板、指标仪表板、审计查看器和模板浏览器 | | **Tkinter GUI** | 置顶浮动小部件，包含文档、实时、图表、聊天、设置选项卡 | ## 2. 安全架构 Brompt 实现了多层纵深防御安全 pipeline，而不仅仅是单一的 regex 黑名单。 ### 第 1 层：输入规范化 - **NFKC Unicode 规范化** —— 中和同形字攻击 - **零宽字符剥离** —— 移除不可见的混淆（`\u200b`、`\u200c`、`\ufeff` 等） - **Base64 payload 检测** —— 对编码的 payload 进行启发式评分 + 验证 - **Leetspeak 规范化** —— 中和 `0→o`、`3→e`、`4→a`、`@→a`、`$→s` 等 ### 第 2 层：模式黑名单 涵盖 4 种语言（英语、阿拉伯语、意大利语、德语）的 14 种 regex 模式，用于： - 直接指令覆盖 - System prompt 泄露尝试 - Guardrail 绕过尝试 - 越狱角色切换 - 角色扮演绕过 ### 第 3 层：语义分类器 (`classifier.py`) 可选的基于 LLM 的第二道防线，它推理意图，而不是表面文本： - 捕获改写、翻译或混淆的攻击 - 返回结构化 JSON：`{is_injection, confidence, reasoning}` - 可配置的置信度阈值（默认为 0.7） - 默认关闭 —— 每个请求需要额外的模型调用成本 ### 第 4 层：输出净化 在内容到达调用者之前，对类似秘密的内容进行脱敏： - Anthropic API 密钥（`sk-ant-...`） - OpenAI 风格的密钥（`sk-...`） - AWS 访问密钥（`AKIA...`） - 私钥块 - GitHub / Slack token ### 第 5 层：审计级可观测性 - SHA-256 哈希链，仅追加的审计日志 - 通过 `AuditLog.verify()` 实现防篡改 - 记录的每个安全事件都带有不可变链 **注意：** 没有哪个黑名单是绝对保证。分类器层显著提高了门槛，但纵深防御设计（最小权限工具、输出脱敏、审计跟踪）是必不可少的。 ## 3. 核心功能 ### 断路器 (`brompt.circuit_breaker`) 标准的 CLOSED / OPEN / HALF_OPEN 状态机： - 可配置的失败阈值 + 恢复超时 - 使用 `threading.Lock()` 实现线程安全 - 支持同步（`call_sync`）和异步（`call`）路径 - 内置 fallback 支持 ``` from brompt.circuit_breaker import CircuitBreaker cb = CircuitBreaker(failure_threshold=5, recovery_timeout=30.0) result = await cb.call(provider.generate(messages), fallback="Service unavailable") ``` ### 模型路由器 (`brompt.router`) 根据策略路由请求： - **Cheapest** —— 最小化成本 - **Fastest** —— 最小化延迟 - **Best Quality** —— 最大化输出质量 - **Fallback** —— 失败时级联 - 复杂度感知 —— 将简单的查询路由到较便宜的模型 ``` from brompt.router import ModelRouter, RoutingStrategy router = ModelRouter() router.register_provider("cheap", ollama_provider) router.register_provider("fast", openai_provider) router.register_provider("quality", anthropic_provider) route = await router.route(query, strategy=RoutingStrategy.CHEAPEST) ``` ### Prompt 优化器 (`brompt.optimizer`) 内置的 prompt 优化工具，用于提高 token 效率和输出质量。 ``` from brompt.optimizer import TokenOptimizer optimizer = TokenOptimizer() compressed = optimizer.compress_context(messages) ``` ### 成本跟踪 (`brompt.pricing`) 跨 provider 的每请求成本估算和跟踪。 ``` from brompt.pricing import estimate_cost cost = estimate_cost("anthropic", prompt_tokens=150, completion_tokens=50) # → 0.0012 (USD) ``` ## 4. 仓库布局 ``` Brompt/ ├── .github/ │ └── workflows/ │ └── ci.yml # GitHub Actions CI/CD Pipeline ├── assets/ │ ├── dark.png # Dark mode banner │ └── light.png # Light mode banner ├── src/ │ └── brompt/ │ ├── __init__.py # Package root — exports all public API │ ├── providers_core.py # Core provider integrations (sync) │ ├── schema.py # Data Models & System Schemas │ ├── security.py # Ingress filtering + output redaction │ ├── memory.py # Bounded turn history + session state (Thread-Safe) │ ├── ratelimit.py # Per-caller sliding-window rate limiter │ ├── audit.py # Hash-chained, tamper-evident audit log │ ├── config.py # Dataclass configs (WidgetConfig, ProviderConfig, etc.) │ ├── session.py # Session management (Session, SessionManager, Message) │ ├── widget.py # PromptClient — unified client entry point │ ├── hooks.py # Hooks/middleware system (Logging, Timing, Validation, etc.) │ ├── observability.py # Tracing, metrics (Prometheus), alert management │ ├── circuit_breaker.py # CLOSED/OPEN/HALF_OPEN state machine with fallback │ ├── router.py # ModelRouter — heuristic complexity classification, 4 strategies │ ├── classifier.py # LLM-based semantic injection classifier (opt-in) │ ├── pricing.py # Cost estimation per provider/model │ ├── optimizer.py # Token optimization and compression │ ├── core/ │ │ ├── __init__.py # Re-exports BromptEngine │ │ ├── engine.py # Main Execution Runtime Engine │ │ └── template_engine.py # Template engine (variables, filters, control flow, 6 built-in templates) │ ├── providers/ │ │ ├── __init__.py # Provider registry with all 7 providers │ │ ├── base.py # Abstract LLMProvider base class │ │ ├── factory.py # ProviderFactory + ProviderRegistry │ │ ├── openai_provider.py # OpenAI / ChatGPT / GPT-4o │ │ ├── anthropic_provider.py # Anthropic / Claude │ │ ├── google_provider.py # Google Gemini │ │ ├── mistral_provider.py # Mistral AI │ │ ├── ollama_provider.py # Ollama (local) │ │ └── azure_provider.py # Azure OpenAI │ ├── cli/ │ │ ├── __init__.py # CLI package │ │ └── main.py # Typer-based CLI (8 commands) │ ├── guiapp/ # Tkinter GUI application │ │ ├── __init__.py # BromptWidget — floating always-on-top panel (GUI, not backend) │ │ ├── ui.py # Tab bar, title bar, resize grip, tooltip, keyboard bindings │ │ ├── chart.py # ChartEngine — 5 chart types (bar, line, area, stacked, donut) │ │ ├── theme.py # Design tokens (colors, fonts, spacing) │ │ └── badge.py # System tray / Toplevel badge for minimize │ ├── api/ # FastAPI REST API server │ └── feedback/ # Feedback loop system ├── webui/ │ └── streamlit_app.py # Streamlit web UI ├── tests/ │ ├── test_core.py # Core Runtime Unit Tests │ ├── test_security.py # Security Filter Unit Tests │ ├── test_memory.py # Memory Engine Unit Tests │ ├── test_providers.py # Provider Abstraction Unit Tests │ ├── test_ratelimit.py # Rate Limiter Unit Tests │ ├── test_audit.py # Audit Log Unit Tests │ ├── test_api.py # API endpoint tests │ └── ... # Feedback, retry, classifier, etc. ├── agent.brompt.yaml # Declarative Runtime Manifest ├── pyproject.toml # Package Configuration & Dependencies └── README.md # Technical Specification ``` ## 5. 配置清单 运行时行为由 `agent.brompt.yaml` 控制： ``` metadata: name: "ProductionAgentEngine" version: "2.0.0" environment: "production" security_policy: isolation_level: "ZERO_TRUST" sanitize_inputs: true max_payload_size_kb: 64 memory_strategy: paging_mode: "VIRTUAL_STATE_O1" max_history_turns: 3 rate_limit: max_requests: 30 window_seconds: 60 schema_validation: strict_mode: true ``` ## 6. 快速开始 ``` # 克隆 git clone https://github.com/sh0-dax/Brompt.git cd Brompt # 安装 python -m venv venv source venv/bin/activate # 使用特定 provider 安装（选择一个） pip install -e ".[anthropic]" # Anthropic / Claude pip install -e ".[openai]" # OpenAI / ChatGPT / GPT-4o pip install -e ".[ollama]" # Ollama (local) pip install -e ".[gemini]" # Google Gemini pip install -e ".[mistral]" # Mistral pip install -e ".[azure]" # Azure OpenAI pip install -e ".[lmstudio]" # LM Studio (local) pip install -e ".[all]" # All providers + web UI # 运行测试 pytest -v # 启动 CLI（交互式聊天） brompt chat # 运行单个 prompt brompt run "What is the capital of France?" # 列出可用 templates brompt templates # 启动 web UI（需要 pip install -e ".[webui]"） streamlit run webui/streamlit_app.py ``` ## 7. Providers Brompt 通过可插拔的 provider 系统支持 **7 个 LLM provider**。每个都是可选的依赖项 —— 你只需要安装你需要的。 ### Provider 矩阵 | Provider | 包 | 环境变量 | 默认模型 | 类型 | |---|---|---|---|---| | **Anthropic** | `anthropic` | `ANTHROPIC_API_KEY` | `claude-sonnet-4-5` | Cloud | | **OpenAI** | `openai` | `OPENAI_API_KEY` | `gpt-4o` | Cloud | | **Ollama** | `ollama` | `OLLAMA_HOST` | `llama3.2` | Local | | **Gemini** | `google-genai` | `GEMINI_API_KEY` | `gemini-2.5-flash` | Cloud | | **Mistral** | `mistralai` | `MISTRAL_API_KEY` | `mistral-large-latest` | Cloud | | **Azure OpenAI** | `openai` | `AZURE_OPENAI_API_KEY`, `AZURE_OPENAI_ENDPOINT`, `AZURE_OPENAI_DEPLOYMENT` | — | Cloud | | **LM Studio** | `openai` | `LM_STUDIO_HOST` | `default` | Local | ### 环境变量 ``` # Cloud providers export ANTHROPIC_API_KEY=sk-ant-... export OPENAI_API_KEY=sk-... export GEMINI_API_KEY=... export MISTRAL_API_KEY=... export AZURE_OPENAI_API_KEY=... export AZURE_OPENAI_ENDPOINT=https://myinstance.openai.azure.com export AZURE_OPENAI_DEPLOYMENT=gpt-4o-deployment # Local providers export OLLAMA_HOST=http://localhost:11434 # default export LM_STUDIO_HOST=http://localhost:1234/v1 # default ``` ### 自动检测 当没有显式注入 provider 时，`build_provider_from_env()` 会按此优先级顺序检查环境变量： 1. `ANTHROPIC_API_KEY` → Anthropic 2. `OPENAI_API_KEY` → OpenAI 3. `GEMINI_API_KEY` → Gemini 4. `MISTRAL_API_KEY` → Mistral 5. `AZURE_OPENAI_API_KEY` → Azure OpenAI 6. `OLLAMA_HOST` → Ollama 7. `LM_STUDIO_HOST` → LM Studio 如果未设置任何环境变量，引擎将以 **dry-run / 仅验证模式** 运行 —— 输入将被净化，状态将被管理，但不会调用任何 LLM。 ## 8. API 参考 ### `BromptEngine(config_path, provider=None, async_provider=None, rate_limiter=None, injection_classifier=None, circuit_breaker=None)` 核心运行时入口点。加载 YAML 清单并初始化子系统。 | 参数 | 类型 | 默认值 | 描述 | |---|---|---|---| | `config_path` | `str` | `"agent.brompt.yaml"` | 运行时清单的路径 | | `provider` | `LLMProvider \| None` | `None` | 同步 provider（如果为 `None` 则从环境变量自动检测） | | `async_provider` | `LLMProvider \| None` | `None` | 用于 `execute_async()` 的异步 provider | | `audit_log_path` | `str \| None` | `None` | 自定义审计日志路径 | | `rate_limiter` | `RateLimiterBackend \| None` | `None` | 自定义 rate limiter 实例 | | `injection_classifier` | `InjectionClassifier \| None` | `None` | 可选的基于 LLM 的注入分类器 | | `circuit_breaker` | `CircuitBreaker \| None` | `None` | 用于 provider 调用的可选断路器 | **方法：** - `execute(user_query, context=None, caller_id="default", system_prompt=None) → ExecutionResult` —— 同步 pipeline。 - `execute_async(user_query, context=None, caller_id="default", system_prompt=None) → ExecutionResult` —— 相同的 pipeline，可 await。 ### `LLMProvider` (ABC) 所有 provider 的基类。实现 `generate()` 用于同步，可选实现 `agenerate()` 用于异步。 | 方法 | 返回值 | 描述 | |---|---|---| | `generate(messages, system=None)` | `str` | 使用有界对话历史调用 LLM | | `agenerate(messages, system=None)` | `str` | 异步对应方法（可选） | ### `SecurityEngine.sanitize(text)` 根据对抗性模式验证输入。如果违规则引发 `SecurityViolationError` 或 `ValueError`。 ### `SecurityEngine.sanitize_output(text)` 对模型输出中的类似秘密的内容（API 密钥、token）进行脱敏。 ### `MemoryManager(max_turns)` 线程安全的有界状态管理器。 | 方法 | 返回值 | 描述 | |---|---|---| | `update_state(key, value)` | `None` | 线程安全的状态更新 | | `get_state()` | `dict[str, Any]` | 当前状态的快照副本 | | `add_turn(role, content)` | `None` | 追加一轮对话 | | `get_history()` | `list[dict[str, str]]` | 有界对话历史 | | `clear()` | `None` | 线程安全的状态 + 历史刷新 | ### `RateLimiter(max_requests, window_seconds)` 基于调用者的滑动窗口 rate limiter。 | 方法 | 引发 | 描述 | |---|---|---| | `check(identifier)` | `RateLimitExceededError` | 记录一次命中；如果预算耗尽则引发异常 | ### `AuditLog(path)` SHA-256 哈希链，仅追加的审计日志。 | 方法 | 返回值 | 描述 | |---|---|---| | `record(event, state_id, is_secure, detail=None)` | `dict` | 追加防篡改记录 | | `verify()` | `bool` | 重放链；如果被篡改则为 `False` | | `read_all()` | `list[dict]` | 读取所有条目 | ### `CircuitBreaker(failure_threshold=5, recovery_timeout=30.0, half_open_max_calls=3)` 使用 CLOSED/OPEN/HALF_OPEN 状态机保护 provider 免受级联故障影响。 | 参数 | 类型 | 默认值 | 描述 | |---|---|---|---| | `failure_threshold` | `int` | `5` | 断开电路前的连续失败次数 | | `recovery_timeout` | `float` | `30.0` | 转换到 HALF_OPEN 之前的秒数 | | `half_open_max_calls` | `int` | `3` | HALF_OPEN 状态下允许的探测请求数 | | 方法 | 返回值 | 描述 | |---|---|---| | `call(coro, fallback=None)` | `Any` | 异步调用；如果断开则引发 `CircuitBreakerOpenError` | | `call_sync(fn, args=(), kwargs={}, fallback=None)` | `Any` | 同步对应方法 | ### `ModelRouter(profiles=None, strategy=RoutingStrategy.CHEAPEST)` 根据复杂度分类将 prompt 路由到最佳 provider。 | 方法 | 返回值 | 描述 | |---|---|---| | `classify_complexity(text)` | `ComplexityLevel` | 启发式：字数、代码/数学标记、分析关键词 | | `score_providers(complexity)` | `list[Route]` | 排序后的（provider、模型、得分、估算成本）列表 | | `route(text, strategy=None)` | `Route \| None` | 按策略选择 provider（CHEAPEST / FASTEST / BEST_QUALITY / FALLBACK） | | `register_provider(name, provider)` | `None` | 注册一个 provider 实例用于路由 | | `register_providers(providers)` | `None` | 通过字典批量注册 provider | ### `LLMInjectionClassifier(provider, confidence_threshold=0.7)` 可选的基于 LLM 的语义注入检测器 —— 捕获改写的攻击。 | 方法 | 返回值 | 描述 | |---|---|---| | `classify(text)` | `ClassificationResult` | 结构化 JSON：`{is_injection, confidence, reasoning}` | | `is_blocked(text)` | `ClassificationResult \| None` | 如果安全则为 `None`；如果高于阈值被阻止则为结果 | ### CLI 命令 ``` Usage: brompt [OPTIONS] COMMAND [ARGS]... Commands: chat Start an interactive chat session run Execute a single prompt history Show conversation history audit Show audit log entries status Show engine status and configuration templates List or render prompt templates config Show or validate a Brompt config file clear Clear engine memory and history ``` ### `PromptClient(config=None)` (别名：`BromptWidget`) 统一的高级入口点，结合了 engine、session 和小部件配置。 ``` from brompt import PromptClient client = PromptClient() result = client.execute("Hello!") print(result.data) ``` ### 模板引擎 ``` from brompt import Template, template_registry # 带有 filters 的简单 template t = Template("Hello {{ name|upper }}!", name="greeting") print(t.render(name="world")) # "Hello WORLD!" # 内置 templates rendered = template_registry.render("chat", user_message="What's new?", system_prompt="You are a helpful assistant.", messages=[], ) # 注册自定义 template from brompt.core.template_engine import create_builtin_templates reg = create_builtin_templates() reg.render("code_review", language="python", code="print('hello')") ``` **内置模板：** `chat`、`code_review`、`summarize`、`translate`、`analysis`、`debug` **过滤器：** `upper`、`lower`、`capitalize`、`title`、`trim`、`json`、`now` ### Hooks / Middleware ``` from brompt import HooksManager, LoggingHook, TimingHook, ValidationHook hooks = HooksManager() hooks.register(LoggingHook()) hooks.register(TimingHook()) query, ctx = hooks.before_execute("Hello", None) result = engine.execute(query, ctx) result = hooks.after_execute(result) ``` **内置 hooks：** `LoggingHook`、`TimingHook`、`ValidationHook`、`AuditHook`、`RateLimitHook`、`SecurityHook` ### 可观测性 ``` from brompt import tracer, metrics, AlertManager, AlertRule # Tracing span = tracer.start_span("llm_call", attributes={"model": "gpt-4"}) span.finish() # Metrics metrics.inc("api_calls") metrics.observe("latency_ms", 320.5) print(metrics.export_prometheus()) # Alerts am = AlertManager() am.add_rule(AlertRule( name="high_error_rate", condition=lambda ctx: ctx.get("errors", 0) > 10, message="Error rate exceeds threshold", )) am.evaluate({"errors": 15}) ``` ### Web UI ``` pip install -e ".[webui]" streamlit run webui/streamlit_app.py ``` 打开基于浏览器的界面，带有聊天面板、指标仪表板、审计日志查看器和模板浏览器。 ### Tkinter GUI（合规仪表板） ``` python -m brompt.guiapp [--live] ``` 置顶浮动小部件，具有 5 个选项卡（文档、实时状态、图表、聊天、设置）、系统托盘最小化以及实时引擎监控。GUI 的 `BromptWidget` 类是 **前端** —— 它在内部使用 `PromptClient`（后端客户端）。 ## 9. 高级功能 ### 语义注入分类 在安全性至关重要的生产部署中启用 LLM 分类器： ``` from brompt.classifier import LLMInjectionClassifier from brompt.providers_core import build_provider_from_env provider = build_provider_from_env() classifier = LLMInjectionClassifier(provider, confidence_threshold=0.7) result = classifier.is_blocked(user_input) if result: print(f"Blocked: {result.reasoning}") ``` ### 带 Fallback 的断路器 ``` from brompt.circuit_breaker import CircuitBreaker cb = CircuitBreaker(failure_threshold=3, recovery_timeout=60.0) try: response = await cb.call( primary_provider.agenerate(messages), fallback=await backup_provider.agenerate(messages) ) except CircuitBreakerOpenError: response = "All providers unavailable. Please try again later." ``` ### 模型路由 ``` from brompt.router import ModelRouter, RoutingStrategy router = ModelRouter() router.register_provider("cheap", ollama_provider) router.register_provider("fast", openai_provider) router.register_provider("quality", anthropic_provider) route = await router.route(query, strategy=RoutingStrategy.CHEAPEST) response = await route.provider.agenerate(messages) ``` ### Prompt 优化 ``` from brompt.optimizer import TokenOptimizer optimizer = TokenOptimizer() tokens = optimizer.estimate_tokens("Hello world") # ~3 compressed = optimizer.compress_context(messages) summary = optimizer.summarize_history(message_history) ``` ### 成本估算 ``` from brompt.pricing import estimate_cost cost = estimate_cost("gpt-4o", prompt_tokens=200, completion_tokens=100) print(f"Request cost: ${cost:.6f}") ``` ## 10. CI/CD Pipeline GitHub Actions 会在每次 push/PR 时跨 Python 3.10–3.13 运行测试： ``` matrix: python-version: ["3.10", "3.11", "3.12", "3.13"] ``` ## 11. 生产就绪情况 **当前状态：生产就绪 (v2)** —— Brompt 专为合规性、不可否认性和可审计性是必需条件而非锦上添花的受监管行业而设计。 ### 合规与治理 - ✅ **哈希链审计日志** —— SHA-256 仅追加链；`AuditLog.verify()` 检测篡改 - ✅ **签名的执行回执** —— 带有阶段标志和时间戳的 HMAC 哈希 `ExecutionResult`（通过 pipeline 的法律级证明） - ✅ **Policy-as-code** —— 在 provider 执行之前评估每个租户的 YAML 策略 - ✅ **Human-in-the-loop** —— 用于敏感部署的可配置灰色地带阈值 - ✅ **安全防护栏** —— 多层：规范化、regex 黑名单、LLM 分类器、输出脱敏 - ✅ **Rate limiting**（进程内滑动窗口；分布式变体需要 Redis） - ✅ **输出净化** —— 在到达调用者之前对类似秘密的字符串进行脱敏 ### Provider 与性能 - ✅ 7 个 LLM provider（Anthropic、OpenAI、Ollama、Gemini、Mistral、Azure OpenAI、LM Studio） - ✅ 异步执行路径（`execute_async`，为同步 provider 提供线程卸载） - ✅ 断路器（带有 fallback 的 CLOSED/OPEN/HALF_OPEN 状态机） - ✅ 模型路由器（启发式复杂度分类，4 种路由策略） - ✅ 带有压缩和缓存的 token 优化 - ✅ 基于模型的成本估算和节省跟踪 ### 接口 - ✅ Typer CLI（8 个命令） - ✅ Streamlit Web UI - ✅ Tkinter 浮动小部件 GUI - ✅ FastAPI REST API - ✅ 带有过滤器、条件、循环的模板引擎 - ✅ 可观测性（链路追踪、Prometheus 指标、告警规则） - ✅ Redis 缓存（带进程内 LRU/SmartCache fallback 的键值存储） ### 路线图 - ⚠️ **签名执行回执序列化** —— 为外部审计生成独立的 `.receipt` 文件 - ⚠️ **确定性重放 CLI** —— `brompt replay --model=X`，带 diff 输出 - ⚠️ **分布式 rate limiting** —— 多实例 Redis 支持的 rate limiter ## 12. 许可证 MIT 许可证 —— 有关详细信息，请参阅 [LICENSE](LICENSE)。

⬆ 返回顶部

_{由 ❤️ SH ÂZZOUZ 构建 —— sh0-dax}

标签：AI 中间件, API集成, Kubernetes, LLM 运行时, Python, 可观测性, 执行控制, 无后门, 系统安全加固, 逆向工具