vgovern-sdk 0.1.0

生产级、低侵入的 AI Agent 治理与可观测性 SDK

vgovern SDK

生产级、低侵入的 AI Agent 治理与可观测性 SDK

简介

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

快速开始

安装

pip install vgovern-sdk

或安装特定框架集成：

# 仅安装基础包
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]"

基本使用

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}

装饰器方式

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 集成

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/ 目录。

配置选项

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	停止执行（需处理异常）

# 即使后端不可用，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

详细文档请查看 使用示例。

测试

运行测试

# 运行所有测试
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

测试覆盖率

# 生成覆盖率报告
python -m coverage run -m unittest discover tests
python -m coverage report
python -m coverage html

当前覆盖率：82%

• 单元测试：90 个测试用例，100% 通过
• 集成测试：14 个测试用例，覆盖主要场景
• 性能测试：7 个基准测试

文档

• 使用示例: 详细的使用示例
• 故障排查: 常见问题和解决方案
• 实现文档: SDK 实现细节
• 完成文档: 完成状态说明
• 变更日志: 版本变更记录

示例代码

目录结构

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 集成

运行示例

# 基本使用
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 异常并优雅处理：

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() 工具函数，支持自定义价格：

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() 自定义。

更多问题请查看 故障排查文档。

许可证

MIT License - 详见 LICENSE 文件

贡献

欢迎贡献！请查看贡献指南（待添加）。

联系方式

• 问题反馈：请提交 Issue
• 功能建议：请提交 Issue 或 Pull Request

---

vgovern SDK - 为 AI Agent 提供生产级的治理与可观测性能力