菜单

④使用指南

发布于 2025年11月26日

详细的使用指南，帮助您掌握CAI框架的各种功能和使用技巧

Agent使用

学习如何有效使用和管理各种专业Agent，从基础运行到高级配置。

运行Agent

如何启动和使用各种专业Agent

from cai import Agent

# 创建Agent实例
agent = Agent(
    model="openai/gpt-4",
    name="redteam_specialist",
    role="渗透测试专家"
)

# 配置Agent
agent.configure({
    max_iterations: 10,
    timeout: 300,
    guardrails=True
})

# 运行Agent任务
result = agent.run(
    "对目标网站 example.com 进行渗透测试",
    context={
        "target": "example.com",
        "scope": "web_application"
    }
)

print(f"任务完成: {result.status}")
print(f"结果: {result.output}")

Agent切换

在不同Agent间切换和组合使用

from cai import Agent, AgentRegistry

# 获取Agent注册表
registry = AgentRegistry()

# 列出所有可用Agent
available_agents = registry.list_agents()
print(f"可用Agent: {available_agents}")

# 根据任务类型选择Agent
def select_agent(task_type: str):
    if task_type == "reconnaissance":
        return registry.get_agent("recon_agent")
    elif task_type == "exploitation":
        return registry.get_agent("exploit_agent")
    else:
        return registry.get_agent("general_agent")

# 使用选择的Agent
agent = select_agent("reconnaissance")
result = agent.run("扫描网络发现服务")

性能优化

优化Agent运行性能和安全设置

from cai import Agent
from cai.optimization import PerformanceOptimizer

# 创建优化配置
optimizer = PerformanceOptimizer({
    "max_concurrent_tasks": 5,
    "timeout_per_task": 180,
    "memory_limit": "512MB",
    "cpu_limit": "50%"
})

# 应用优化设置
optimized_agent = Agent(
    model="openai/gpt-4",
    optimizer=optimizer
)

# 批量执行任务
tasks = [
    "扫描端口80",
    "检查目录遍历",
    "测试SQL注入"
]

results = optimized_agent.run_batch(
    tasks,
    parallel=True,
    max_workers=3
)

结果处理

掌握如何正确处理、分析和利用Agent的执行结果。

结果解析

解析和理解Agent返回的结果

result = agent.run("漏洞扫描")

# 检查执行状态
if result.status == "success":
    print("任务成功完成")
    
    # 访问主要输出
    main_output = result.output
    print(f"主要结果: {main_output}")
    
    # 访问工具调用结果
    tool_results = result.tool_calls
    for call in tool_results:
        print(f"工具 {call.tool_name}: {call.result}")
        
    # 访问思考过程
    reasoning = result.reasoning
    print(f"推理过程: {reasoning}")
    
else:
    print(f"任务失败: {result.error}")
    print(f"错误详情: {result.error_details}")

结果持久化

保存结果到文件或数据库

import json
from datetime import datetime

# 保存为JSON文件
def save_result(result, filename=None):
    if not filename:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"result_{timestamp}.json"
    
    result_data = {
        "timestamp": datetime.now().isoformat(),
        "agent_name": result.agent_name,
        "task": result.task,
        "status": result.status,
        "output": result.output,
        "tool_calls": [call.to_dict() for call in result.tool_calls],
        "metadata": result.metadata
    }
    
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(result_data, f, indent=2, ensure_ascii=False)
    
    return filename

# 使用示例
result = agent.run("安全评估")
saved_file = save_result(result)
print(f"结果已保存到: {saved_file}")

结果分析

分析结果数据并生成报告

from cai.analysis import ResultAnalyzer

# 创建分析器
analyzer = ResultAnalyzer()

# 分析单个结果
analysis = analyzer.analyze(result)
print(f"风险等级: {analysis.risk_level}")
print(f"发现的问题数: {len(analysis.vulnerabilities)}")

# 分析多个结果
results_batch = [result1, result2, result3]
batch_analysis = analyzer.analyze_batch(results_batch)

# 生成报告
report = analyzer.generate_report(batch_analysis)
print(report.to_markdown())

# 导出分析结果
analyzer.export_report(batch_analysis, "security_report.pdf")

流式传输

使用流式传输实时接收Agent输出，改善用户体验和实时反馈能力。

流式传输配置

实时接收Agent输出，支持进度显示和增量更新

from cai.streaming import StreamingConfig, StreamingResult

# 配置流式传输
streaming_config = StreamingConfig({
    "buffer_size": 1024,
    "timeout": 30,
    "retry_attempts": 3
})

# 使用流式传输
agent = Agent(
    model="openai/gpt-4",
    streaming=True,
    streaming_config=streaming_config
)

# 实时接收结果
result_stream = agent.run_streaming(
    "生成详细的安全评估报告",
    on_chunk=lambda chunk: print(f"接收数据: {chunk.content}"),
    on_complete=lambda final_result: print("任务完成")
)

# 处理流式结果
streaming_result = StreamingResult(result_stream)
for update in streaming_result:
    print(f"进度: {update.progress}%")
    if update.content:
        print(f"当前内容: {update.content}")

上下文管理

有效管理对话上下文和会话状态，确保Agent之间的信息传递和状态维护。

上下文管理最佳实践

创建、管理和利用上下文信息，提升Agent协作效率

from cai import Agent
from cai.context import ContextManager, ConversationContext

# 创建上下文管理器
context_manager = ContextManager()

# 创建对话上下文
conversation = ConversationContext(
    session_id="pentest_session_001",
    max_history=100
)

# 添加上下文信息
conversation.add_context({
    "target_scope": "192.168.1.0/24",
    "authorized_actions": ["reconnaissance", "vulnerability_scan"],
    "time_constraints": "2_hours",
    "documentation_level": "detailed"
})

# 关联Agent与上下文
agent = Agent(
    model="openai/gpt-4",
    context=conversation
)

# 使用上下文执行任务
result = agent.run_with_context(
    "对指定网段进行安全扫描",
    context={"scan_type": "network_discovery"}
)

# 维护上下文状态
context_manager.save_context(conversation)
loaded_context = context_manager.load_context("pentest_session_001")

追踪调试

使用Phoenix等工具进行详细的执行追踪和性能监控。

执行追踪配置

详细记录Agent执行过程，便于调试和性能分析

from cai import Agent
from cai.tracing import TracingConfig, PhoenixTracer

# 配置追踪
tracing_config = TracingConfig({
    "enable_tracing": True,
    "log_level": "DEBUG",
    "save_traces": True,
    "trace_format": "json"
})

# 初始化Phoenix追踪器
phoenix = PhoenixTracer(
    project_name="cai-security-tests",
    tracing_config=tracing_config
)

# 在Agent中启用追踪
agent = Agent(
    model="openai/gpt-4",
    tracer=phoenix
)

# 执行任务（带追踪）
with agent.trace("security_assessment") as span:
    span.set_attribute("target", "example.com")
    span.set_attribute("scan_type", "comprehensive")
    
    result = agent.run("执行完整的安全评估")
    
    span.set_attribute("result_status", result.status)
    span.set_attribute("vulnerabilities_found", len(result.vulnerabilities))

# 查看追踪结果
traces = phoenix.get_traces()
for trace in traces:
    print(f"追踪ID: {trace.trace_id}")
    print(f"持续时间: {trace.duration}")
    print(f"操作: {trace.operations}")

移交机制

实现Agent间的智能移交和协作，构建复杂的多Agent工作流。

Agent协作工作流

配置Agent间的智能交接规则，构建端到端的安全任务处理流程

from cai import Agent
from cai.handoffs import HandoffManager, HandoffRule

# 创建交接管理器
handoff_manager = HandoffManager()

# 定义交接规则
handoff_rules = [
    HandoffRule(
        trigger="reconnaissance_complete",
        from_agent="recon_agent",
        to_agent="exploit_agent",
        context_data=["discovered_services", "vulnerabilities"]
    ),
    HandoffRule(
        trigger="exploitation_successful",
        from_agent="exploit_agent", 
        to_agent="post_exploitation_agent",
        context_data=["gained_access", "privilege_level"]
    )
]

# 注册交接规则
for rule in handoff_rules:
    handoff_manager.add_rule(rule)

# 创建多Agent链
recon_agent = Agent("recon_agent", model="gpt-4")
exploit_agent = Agent("exploit_agent", model="claude-3")

# 执行带交接的任务链
result = handoff_manager.execute_chain(
    agents=[recon_agent, exploit_agent],
    initial_task="对目标进行渗透测试",
    rules=handoff_rules
)

print(f"链式任务结果: {result}")

上一篇 ⑤开发文档
下一篇 ③高级功能

Powered by RTAi-Lab