vgovern-sdk 0.1.0
生产级、低侵入的 AI Agent 治理与可观测性 SDK
License: MIT
Requires Python: >=3.10
Keywords: ai, agent, governance, observability, monitoring, llm, langchain, langgraph

# vgovern SDK

生产级、低侵入的 AI Agent 治理与可观测性 SDK

[![Python Version](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)
[![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)
[![Tests](https://img.shields.io/badge/tests-passing-brightgreen.svg)](#测试)
[![Coverage](https://img.shields.io/badge/coverage-82%25-success.svg)](#测试)

## 简介

vgovern SDK 为 AI Agent 提供生产级的治理和可观测性能力，帮助你：

- **监控 Agent 执行**：自动追踪 LLM 调用、工具调用、成本和延迟
- **执行治理策略**：基于成本、风险、频率等维度控制 Agent 行为
- **保持业务连续性**：Fail-safe 设计确保 SDK 故障不影响 Agent 运行
- **支持主流框架**：原生集成 LangChain、LangGraph、AutoGen 等

## 特性

### 核心能力

| 特性 | 说明 |
|------|------|
| **Trace 追踪** | 完整的 Agent 执行生命周期追踪，支持树形步骤结构 |
| **策略检查** | 实时成本控制、频率限制、风险评估，支持 ALLOW/HALT 决策 |
| **数据脱敏** | 自动识别并脱敏敏感数据（API Key、Token、SSN、邮箱等） |
| **成本计算** | 内置主流 LLM 模型成本计算（GPT-4、Claude、Gemini 等） |
| **异步上报** | 非阻塞事件上报，不影响 Agent 性能 |
| **Fail-Safe** | 网络故障、后端超时等情况下 Agent 继续执行 |

### 框架集成

- ✅ OpenAI API
- ✅ LangChain
- ✅ LangGraph
- ✅ AutoGen
- ✅ 自定义 Agent

## 快速开始

### 安装

```bash
pip install vgovern-sdk
```

或安装特定框架集成：

```bash
# 仅安装基础包
pip install vgovern-sdk

# 安装 LangChain 集成
pip install "vgovern-sdk[langchain]"

# 安装 LangGraph 集成
pip install "vgovern-sdk[langgraph]"

# 安装 AutoGen 集成
pip install "vgovern-sdk[autogen]"

# 安装所有集成
pip install "vgovern-sdk[all]"
```

### 基本使用

```python
from vgovern_sdk import create_guard, StepType, GovernanceHaltError

# 创建 Guard 实例
guard = create_guard(
    backend_url="https://governance.example.com",
    api_key="your-api-key",
    agent_id="my-agent",
    environment="production",
)

# 治理整个 Agent 执行
with guard.govern() as trace_id:
    try:
        # 你的 Agent 代码
        result = my_agent_function(trace_id=trace_id)
        print(f"Agent 完成: {result}")
    except GovernanceHaltError as e:
        print(f"策略阻止执行: {e.reason}")
        # 处理中止情况
        return {"error": "blocked_by_policy", "reason": e.reason}
```

### 装饰器方式

```python
from vgovern_sdk import create_guard, StepType

guard = create_guard(
    backend_url="https://governance.example.com",
    api_key="your-api-key",
    agent_id="my-agent",
)

@guard.with_governance(
    step_type=StepType.LLM_CALL,
    model="gpt-4",
    check_policy=True,
)
def call_llm(prompt: str, trace_id: str) -> str:
    # 这里调用实际的 LLM API
    # 例如：openai.ChatCompletion.create(...)
    return f"LLM 回答: {prompt}"

# 使用
with guard.govern() as trace_id:
    result = call_llm("你好，世界！", trace_id=trace_id)
    print(result)
```

### LangChain 集成

```python
from vgovern_sdk import create_guard, StepType, TraceMetadata

guard = create_guard(
    backend_url="https://governance.example.com",
    api_key="your-api-key",
    agent_id="langchain-agent",
)

metadata = TraceMetadata(
    agent_id="langchain-agent",
    session_id="session-123",
    environment="production",
    user_id="user-456",
    tags={"framework": "langchain", "model": "gpt-4"},
)

@guard.with_governance(
    step_type=StepType.LLM_CALL,
    model="gpt-4",
    check_policy=True,
)
def call_langchain(prompt: str, trace_id: str) -> str:
    # 这里调用 LangChain LLM
    # from langchain.chat_models import ChatOpenAI
    # from langchain.schema import HumanMessage
    # llm = ChatOpenAI(model="gpt-4")
    # messages = [HumanMessage(content=prompt)]
    # return llm(messages).content
    return f"LangChain 回答: {prompt}"

with guard.govern(metadata=metadata) as trace_id:
    response = call_langchain("解释量子计算", trace_id=trace_id)
    print(response)
```

更多集成示例请查看 [examples/](examples/) 目录。

## 配置选项

```python
from vgovern_sdk import SDKConfig

config = SDKConfig(
    backend_url="https://governance.example.com",
    api_key="your-api-key",
    agent_id="my-agent",
    environment="production",
    timeout_ms=1000,              # 请求超时时间
    max_retries=2,                # 失败重试次数
    enable_policy_check=True,     # 启用策略检查
    enable_async_reporting=True, # 异步上报
    redact_sensitive_data=True,   # 自动脱敏敏感数据
    fail_safe_mode=True,          # Fail-safe 模式
)

guard = create_guard(**config.__dict__)
```

## Fail-Safe 行为

SDK 设计为 Fail-safe，确保 Agent 的业务连续性：

| 场景 | SDK 行为 | Agent 行为 |
|------|----------|------------|
| 网络故障 | 记录警告日志 | 继续执行 |
| 后端超时 | 记录警告日志 | 继续执行 |
| SDK 内部错误 | 记录警告日志 | 继续执行 |
| 策略 HALT 决策 | 抛出 `GovernanceHaltError` | 停止执行（需处理异常） |

```python
# 即使后端不可用，Agent 仍能正常运行
guard = create_guard(
    backend_url="https://unreachable.example.com",
    api_key="test",
    agent_id="my-agent",
    fail_safe_mode=True,
)

with guard.govern() as trace_id:
    # 后端不可用，但 Agent 继续执行
    result = agent_function(trace_id=trace_id)
```

## API 文档

### 核心类型

- `SDKConfig`: SDK 配置类
- `TraceMetadata`: Trace 元数据
- `TraceInfo`: Trace 信息
- `StepData`: 步骤数据
- `StepNode`: 步骤树节点
- `Trace`: 完整追踪记录
- `TokenUsage`: Token 使用统计
- `GovernanceResponse`: 策略检查响应
- `GovernanceHaltError`: 策略中止异常

### 枚举类型

- `StepType`: 步骤类型（LLM_CALL、TOOL_CALL、TOOL_RESULT、OBSERVATION、FINAL_RESPONSE、UNKNOWN）
- `TraceStatus`: Trace 状态（RUNNING、HALTED、COMPLETED、ERROR）
- `GovernanceDecision`: 策略决策（ALLOW、HALT）

### 客户端

- `GovernanceClient`: 异步 HTTP 客户端
- `SyncGovernanceClient`: 同步 HTTP 客户端

### 主要 API

#### 高层 API
- `create_guard(**kwargs)`: 创建 Guard 实例的工厂函数
- `GovernanceGuard.govern()`: Trace 上下文管理器
- `GovernanceGuard.with_governance()`: 函数装饰器
- `withGovernance`: 上下文管理器式 API

#### Tracer API
- `Tracer.start_trace()`: 启动 Trace
- `Tracer.end_trace()`: 结束 Trace
- `Tracer.record_step()`: 记录步骤
- `Tracer.check_policy()`: 检查策略
- `Tracer.trace_context()`: Trace 上下文管理器
- `Tracer.step_context()`: 步骤上下文管理器
- `Tracer.start_step()`: 启动步骤
- `Tracer.end_step()`: 结束步骤
- `Tracer.add_policy_hit()`: 添加策略命中记录

#### 工具函数
- `TraceLogger`: 带 trace_id 的日志记录器
- `get_logger()`: 获取日志记录器
- `redact_sensitive_data()`: 敏感数据脱敏
- `safe_truncate()`: 安全截断文本
- `generate_step_id()`: 生成步骤 ID
- `calculate_cost_usd()`: 计算成本
- `mask_api_key()`: 掩码 API Key

详细文档请查看 [使用示例](USAGE_EXAMPLES.md)。

## 测试

### 运行测试

```bash
# 运行所有测试
python -m unittest discover tests

# 运行单元测试
python -m unittest tests.test_sdk_types tests.test_utils tests.test_client tests.test_guard tests.test_tracer

# 运行集成测试
python -m unittest tests.test_integration

# 运行性能测试
python -m unittest tests.test_performance
```

### 测试覆盖率

```bash
# 生成覆盖率报告
python -m coverage run -m unittest discover tests
python -m coverage report
python -m coverage html
```

当前覆盖率：**82%**

- 单元测试：90 个测试用例，100% 通过
- 集成测试：14 个测试用例，覆盖主要场景
- 性能测试：7 个基准测试

## 文档

- [使用示例](USAGE_EXAMPLES.md): 详细的使用示例
- [故障排查](TROUBLESHOOTING.md): 常见问题和解决方案
- [实现文档](IMPLEMENTATION.md): SDK 实现细节
- [完成文档](COMPLETE.md): 完成状态说明
- [变更日志](CHANGELOG.md): 版本变更记录

## 示例代码

### 目录结构

```
examples/
├── basic_usage.py           # 基本使用示例
├── policy_check.py          # 策略检查示例
├── fail_safe_demo.py        # Fail-safe 演示
├── langchain_integration.py # LangChain 集成
├── langgraph_integration.py # LangGraph 集成
└── autogen_integration.py   # AutoGen 集成
```

### 运行示例

```bash
# 基本使用
python examples/basic_usage.py

# 策略检查
python examples/policy_check.py

# Fail-safe 演示
python examples/fail_safe_demo.py

# LangChain 集成
python examples/langchain_integration.py

# LangGraph 集成
python examples/langgraph_integration.py

# AutoGen 集成
python examples/autogen_integration.py
```

## 最佳实践

1. **始终传递 trace_id**：确保 trace_id 在所有被治理的函数中传递
2. **使用上下文管理器**：使用 `with guard.govern()` 自动管理生命周期
3. **处理 GovernanceHaltError**：始终捕获并处理策略中止异常
4. **指定正确的步骤类型**：根据实际情况选择 StepType（LLM_CALL、TOOL_CALL 等）
5. **提供有意义的元数据**：添加有用的元数据以便更好地观察和调试
6. **测试 Fail-safe 模式**：验证治理后端不可用时 Agent 是否继续运行
7. **启用数据脱敏**：生产环境建议启用 `redact_sensitive_data=True`

## 常见问题

### Q: SDK 不会影响 Agent 性能吗？

A: SDK 采用 Fail-safe 设计和异步上报，对性能影响极小。性能基准测试显示：
- 启动 Trace：< 50ms
- 记录 Step：< 10ms
- 数据脱敏：< 100μs（短字符串）

### Q: 后端不可用会发生什么？

A: 在 Fail-safe 模式下（`fail_safe_mode=True`），Agent 继续正常执行，SDK 会记录警告日志。只有策略 HALT 决策会停止 Agent。

### Q: 如何处理策略中止？

A: 捕获 `GovernanceHaltError` 异常并优雅处理：

```python
from vgovern_sdk import create_guard, GovernanceHaltError

guard = create_guard(
    backend_url="https://governance.example.com",
    api_key="your-api-key",
    agent_id="my-agent",
)

try:
    with guard.govern() as trace_id:
        result = agent_function(trace_id=trace_id)
except GovernanceHaltError as e:
    return {
        "error": "blocked_by_policy",
        "reason": e.reason,
        "policy_id": e.policy_id,
    }
```

### Q: 如何计算 LLM 成本？

A: SDK 提供 `calculate_cost_usd()` 工具函数，支持自定义价格：

```python
from vgovern_sdk import calculate_cost_usd

cost = calculate_cost_usd(
    prompt_tokens=1000,
    completion_tokens=500,
    model="gpt-4",
    prompt_price_per_1k=0.03,      # 输入每 1k tokens 的价格
    completion_price_per_1k=0.06,  # 输出每 1k tokens 的价格
)
```

内置模型价格需要在实际配置中添加，当前版本支持通过参数自定义。

### Q: 支持哪些 LLM 模型？

A: SDK 框架无关，支持所有 LLM 模型。通过 `model` 参数指定模型名称，成本计算通过 `calculate_cost_usd()` 自定义。

更多问题请查看 [故障排查文档](TROUBLESHOOTING.md)。

## 许可证

MIT License - 详见 [LICENSE](LICENSE) 文件

## 贡献

欢迎贡献！请查看贡献指南（待添加）。

## 联系方式

- 问题反馈：请提交 Issue
- 功能建议：请提交 Issue 或 Pull Request

---

**vgovern SDK** - 为 AI Agent 提供生产级的治理与可观测性能力