DSL 指南¶

其他语言¶

目录¶

概述¶

Symbi DSL 是一种专为创建自主、策略感知智能体而设计的领域特定语言。它将传统编程结构与高级安全功能、加密操作和声明式策略定义相结合。

主要特性¶

安全优先设计：内置策略执行和审计功能
声明式策略：以代码形式表达安全要求
加密操作：原生支持加密、签名和证明
智能体间通信：内置消息传递和协作模式
类型安全：具有安全感知类型注释的强类型系统

语言语法¶

基本结构¶

每个 Symbi 程序都由可选的元数据和智能体定义组成：

metadata {
    version: "1.0.0"
    author: "developer"
    description: "Example agent"
}

agent process_data(input: DataSet) -> Result {
    // Agent implementation
}

计划功能 — Import 语法计划在未来版本中提供。
import data_processing as dp;
import security_utils;

注释¶

// Single-line comment
# Hash-style comment (also supported)

/*
 * Multi-line comment
 * Supports markdown formatting
 */

元数据块¶

元数据提供关于您的智能体的基本信息：

metadata {
    version: "1.2.0"
    author: "ThirdKey Security Team"
    description: "Healthcare data analysis agent with HIPAA compliance"
    license: "Proprietary"
    tags: ["healthcare", "hipaa", "analysis"]
}

元数据键值对可使用 = 或 : 作为分隔符。

元数据字段¶

字段	类型	必需	描述
`version`	String	是	智能体的语义版本
`author`	String	是	智能体作者或组织
`description`	String	是	智能体功能的简要描述
`license`	String	否	许可证标识符
`tags`	Array[String]	否	分类标签

托管 CLI 智能体（Mode B）还能识别以下额外的元数据键，由 symbi run 读取以启动一个受治理的 Claude Code 子进程（参见 agents/code_reviewer.symbi）：

字段	类型	描述
`executor`	String	设为 `"claude_code"` 可将智能体作为受治理的 Claude Code 子进程运行，而非使用推理循环
`model`	String	传递给子进程的模型（例如 `"claude-sonnet-4-5"`）
`allowed_tools`	String	子进程的逗号分隔工具允许列表（例如 `"Read,Grep,Glob"`）。必填 —— 未声明该字段的智能体会被拒绝，而不会以不受限制的方式启动
`system_prompt`	String	为子进程追加的额外系统提示
`permission_mode`	String	可选。作为 `--permission-mode` 透传。未设置时将省略该标志，子进程保留自身默认值；该默认值对 `allowed_tools` 之外的任何操作仍会请求确认。需要无人值守运行的智能体请设为 `"dontAsk"`

启动会作为 tool_call::claude_code 经过策略门控，该决策读取自 policies/managed-cli/ —— 而非 policies/run/。启动子进程与进程内推理循环的影响范围不同，因此这两个界面不共享策略目录。最小的 permit 如下：

// policies/managed-cli/claude_code.cedar
permit(principal, action == Action::"tool_call::claude_code", resource);

启动决策并不是唯一的决策。子进程运行在所选的 Docker、gVisor 或 Firecracker 工作进程内，只有临时暂存存储以及私有的推理和工具通道 —— 没有源码挂载，没有外部网络，没有主机登录状态，也没有主机凭据。CLI 内置工具以及项目/插件的自动发现均被禁用，--plugin-dir 会被拒绝。对源码的访问由一组已注册的 ToolClad 工具提供，并且子进程发起的每一次工具调用都会经由运行时代为中转：准备与规范化调用、取得任何强制性的精确审批、评估 Cedar、持久化必需的作用前记录、派发，然后记录结果。allowed_tools 指定的是已注册工具的一个精确子集；通配符权限表达式和 CLI 内置工具名不会被接受为注册项。

这正是 allowed_tools 为必填、而 permission_mode 采用选择加入的原因：它们限定子进程可以请求什么，而实际发生什么则由运行时决定，而不是由子进程决定。每个托管 CLI 会话都保留自己的签名日志。镜像、挂载、策略和 [managed_cli.inference] 的配置参见托管 CLI 收容，中转契约参见受治理工具中转器。

智能体定义¶

基本智能体结构¶

agent agent_name(param1: Type1, param2: Type2) -> ReturnType {
    capabilities: [capability1, capability2]

    policy policy_name {
        // Policy rules
    }

    with configuration_options {
        // Agent implementation
    }
}

智能体参数¶

支持各种参数类型：

agent complex_agent(
    // Basic types
    name: String,
    age: int,
    active: bool,

    // Optional parameters
    email: Optional<String>,

    // Complex types
    data: Array<Record>,
    config: Map<String, Value>,

    // Security-aware types
    sensitive_data: EncryptedData<PersonalInfo>,
    credentials: SecureString
) -> ProcessingResult {
    // Implementation
}

能力声明¶

声明您的智能体能够做什么：

agent data_processor(input: DataSet) -> Analysis {
    capabilities: [
        data_analysis,          // Core data processing
        statistical_modeling,   // Advanced analytics
        report_generation,      // Output formatting
        audit_logging           // Compliance tracking
    ]

    // Implementation
}

策略定义¶

策略定义在运行时强制执行的安全和合规规则。

策略结构¶

policy policy_name {
    allow: action_list if condition
    deny: action_list if condition
    require: requirement_list
    audit: audit_specification
}

访问控制策略¶

policy medical_data_access {
    allow: ["read", "analyze"] if user.role == "doctor"
    allow: ["read"] if user.role == "nurse"
    deny: ["export", "print"] if data.contains_pii == true
    require: [
        user.clearance >= "medical_professional",
        session.mfa_verified == true,
        audit_trail = true
    ]
}

数据分类策略¶

policy data_classification {
    allow: process(data) if data.anonymized == true
    deny: store(data) if data.classification == "restricted"
    audit: all_operations with digital_signature
}

复杂策略逻辑¶

policy dynamic_access_control {
    allow: read(resource) if (
        user.department == resource.owner_department ||
        user.role == "administrator" ||
        (user.role == "auditor" && current_time.business_hours)
    )

    deny: write(resource) if (
        resource.locked == true ||
        user.last_training < 30d ||
        system.maintenance_mode == true
    )

    require: approval("supervisor") for operations on sensitive_data
}

类型系统¶

基本类型¶

// Basic types
let name: String = "Alice";
let count: int = 42;
let rate: float = 3.14;
let active: bool = true;

集合类型¶

// Arrays
let numbers: Array<int> = [1, 2, 3, 4, 5];
let names: Array<String> = ["Alice", "Bob", "Charlie"];

// Maps
let config: Map<String, String> = {
    "host": "localhost",
    "port": "8080",
    "ssl": "true"
};

// Sets
let unique_ids: Set<String> = {"id1", "id2", "id3"};

安全感知类型¶

// Encrypted types
let secret: EncryptedString = encrypt("sensitive_data", key);
let secure_number: Encrypted<int> = encrypt(42, key);

// Private data with differential privacy
let private_data: PrivateData<float> = PrivateData::new(value, epsilon=1.0);

// Verifiable results with zero-knowledge proofs
let verified_result: VerifiableResult<Analysis> = VerifiableResult {
    value: analysis,
    proof: generate_proof(analysis),
    signature: sign(analysis)
};

自定义类型¶

// Type aliases
type UserId = String;
type EncryptedPersonalInfo = EncryptedData<PersonalInfo>;

计划功能 — struct 和 enum 定义计划在未来版本中提供。目前仅支持 type 别名。

// Struct definitions (planned)
struct PersonalInfo {
    name: String,
    email: EncryptedString,
    phone: Optional<String>,
    birth_date: Date
}

// Enum definitions (planned)
enum SecurityLevel {
    Public,
    Internal,
    Confidential,
    Restricted
}

执行上下文¶

使用 with 子句配置智能体的执行方式：

内存管理¶

agent persistent_agent(data: DataSet) -> Result {
    with memory = "persistent", storage = "encrypted" {
        // Agent state persists across sessions
        store_knowledge(data);
        return process_with_history(data);
    }
}

agent ephemeral_agent(query: String) -> Answer {
    with memory = "ephemeral", cleanup = "immediate" {
        // Agent state is discarded after execution
        return quick_answer(query);
    }
}

隐私设置¶

agent privacy_preserving_agent(sensitive_data: PersonalInfo) -> Statistics {
    with privacy = "differential", epsilon = 1.0 {
        // Add differential privacy noise
        let noisy_stats = compute_statistics(sensitive_data);
        return add_privacy_noise(noisy_stats, epsilon);
    }
}

安全配置¶

agent high_security_agent(classified_data: ClassifiedInfo) -> Report {
    with
        security = "maximum",
        sandbox = "firecracker",
        encryption = "homomorphic",
        requires = "top_secret_clearance"
    {
        // High-security processing
        return process_classified(classified_data);
    }
}

可选的 Linux 原生后端使用 sandbox = "landlock"。通过 HTTP 注册的智能体和计划智能体必须明确指定它。Landlock 是独立的准入分支，不满足 Docker、gVisor、 Firecracker 或托管执行要求。

内置函数¶

数据处理¶

// Validation functions
if (validate_input(data)) {
    // Process valid data
}

// Data transformation
let cleaned_data = sanitize(raw_data);
let normalized = normalize(cleaned_data);

加密操作¶

// Encryption/Decryption
let encrypted = encrypt(plaintext, public_key);
let decrypted = decrypt(ciphertext, private_key);

// Digital signatures
let signature = sign(message, private_key);
let valid = verify(message, signature, public_key);

// Zero-knowledge proofs
let proof = prove(statement);
let verified = verify_proof(proof, public_statement);

审计和日志记录¶

// Audit logging
audit_log("operation_started", {
    "operation": "data_processing",
    "user": user.id,
    "timestamp": now()
});

// Security events
security_event("policy_violation", {
    "policy": "data_access",
    "user": user.id,
    "resource": resource.id
});

智能体间通信¶

直接消息传递¶

agent coordinator(task: Task) -> Result {
    with communication = "secure" {
        // Send task to specialized agent
        let result = agent security_analyzer.analyze(task);

        if (result.safe) {
            let processed = agent data_processor.process(task);
            return processed;
        } else {
            return reject("Security check failed");
        }
    }
}

发布-订阅模式¶

agent event_publisher(event: Event) -> Confirmation {
    with communication = "broadcast" {
        // Broadcast event to all subscribers
        broadcast(EventNotification {
            type: event.type,
            data: event.data,
            timestamp: now()
        });

        return Confirmation { sent: true };
    }
}

agent event_subscriber() -> Void {
    with communication = "subscribe" {
        // Subscribe to specific events
        let events = subscribe(EventNotification);

        for event in events {
            process_event(event);
        }
    }
}

安全通信¶

agent secure_collaborator(request: SecureRequest) -> SecureResponse {
    with
        communication = "encrypted",
        authentication = "mutual_tls"
    {
        // Establish secure channel
        let channel = establish_secure_channel(request.source);

        // Send encrypted response
        let response = process_request(request);
        return encrypt_response(response, channel.key);
    }
}

错误处理¶

Try-Catch 块¶

agent robust_processor(data: DataSet) -> Result {
    try {
        let validated = validate_data(data);
        let processed = process_data(validated);
        return Ok(processed);
    } catch (ValidationError e) {
        audit_log("validation_failed", e.details);
        return Error("Invalid input data");
    } catch (ProcessingError e) {
        audit_log("processing_failed", e.details);
        return Error("Processing failed");
    }
}

错误恢复¶

agent fault_tolerant_agent(input: Input) -> Result {
    let max_retries = 3;
    let retry_count = 0;

    while (retry_count < max_retries) {
        try {
            return process_with_fallback(input);
        } catch (TransientError e) {
            retry_count += 1;
            sleep(exponential_backoff(retry_count));
        } catch (PermanentError e) {
            return Error(e.message);
        }
    }

    return Error("Max retries exceeded");
}

高级功能¶

条件编译¶

计划功能 — 条件编译计划在未来版本中提供。

agent development_agent(data: DataSet) -> Result {
    capabilities: [development, testing]

    #if debug {
        debug_log("Processing data: " + data.summary);
    }

    #if feature.enhanced_security {
        policy strict_security {
            require: multi_factor_authentication
            audit: all_operations with timestamps
        }
    }

    // Implementation
}

宏和代码生成¶

计划功能 — 宏定义计划在未来版本中提供。

// Define reusable policy template
macro secure_data_policy($classification: String) {
    policy secure_access {
        allow: read(data) if user.clearance >= $classification
        deny: export(data) if data.contains_pii
        audit: all_operations with signature
    }
}

agent classified_processor(data: ClassifiedData) -> Report {
    // Use the macro
    secure_data_policy!("secret");

    // Implementation
}

外部系统集成¶

agent api_integrator(request: APIRequest) -> APIResponse {
    capabilities: [api_access, data_transformation]

    policy api_access {
        allow: call(external_api) if api.rate_limit_ok
        require: valid_api_key
        audit: all_api_calls with response_codes
    }

    with
        timeout = 30.seconds,
        retry_policy = "exponential_backoff"
    {
        let response = call_external_api(request);
        return transform_response(response);
    }
}

最佳实践¶

安全指南¶

始终为数据访问和操作定义策略
对敏感数据使用加密类型
为合规性实施审计日志记录
在处理之前验证所有输入
在策略定义中使用最小权限原则

性能优化¶

对短期智能体使用临时内存
尽可能批量操作
实施适当的错误处理和重试机制
在执行上下文中监控资源使用情况
为您的用例使用适当的数据类型

代码组织¶

将相关策略分组在同一块中
使用描述性的能力名称
用注释记录复杂的策略逻辑
将关注点分离到不同的智能体中
使用共享策略定义重用常见模式

示例¶

医疗数据处理器¶

metadata {
    version: "2.1.0"
    author: "Medical AI Team"
    description: "HIPAA-compliant patient data analyzer"
    tags: ["healthcare", "hipaa", "privacy"]
}

agent medical_analyzer(patient_data: EncryptedPatientRecord) -> MedicalInsights {
    capabilities: [
        medical_analysis,
        privacy_preservation,
        audit_logging,
        report_generation
    ]

    policy hipaa_compliance {
        allow: analyze(data) if user.medical_license.valid
        deny: export(data) if data.contains_identifiers
        require: [
            user.hipaa_training.completed,
            session.secure_connection,
            audit_trail = true
        ]
    }

    with
        memory = "encrypted",
        privacy = "differential",
        security = "high",
        requires = "medical_clearance"
    {
        try {
            let decrypted = decrypt(patient_data, medical_key);
            let anonymized = anonymize_data(decrypted);
            let insights = analyze_medical_data(anonymized);

            audit_log("analysis_completed", {
                "patient_id_hash": hash(decrypted.id),
                "insights_generated": insights.count,
                "timestamp": now()
            });

            return insights;
        } catch (DecryptionError e) {
            security_event("decryption_failed", e.details);
            return Error("Unable to process patient data");
        }
    }
}

金融交易监控器¶

agent fraud_detector(transaction: Transaction) -> FraudAssessment {
    capabilities: [fraud_detection, risk_analysis, real_time_processing]

    policy financial_compliance {
        allow: analyze(transaction) if user.role == "fraud_analyst"
        deny: store(transaction.details) if transaction.amount > 10000
        require: [
            user.financial_license.valid,
            system.compliance_mode.active,
            real_time_monitoring = true
        ]
        audit: all_decisions with reasoning
    }

    with
        memory = "ephemeral",
        timeout = 500.milliseconds,
        priority = "high"
    {
        let risk_score = calculate_risk(transaction);
        let historical_pattern = analyze_pattern(transaction.account_id);

        if (risk_score > 0.8 || historical_pattern.suspicious) {
            alert_fraud_team(transaction, risk_score);
            return FraudAssessment {
                risk_level: "high",
                recommended_action: "block_transaction",
                confidence: risk_score
            };
        }

        return FraudAssessment {
            risk_level: "low",
            recommended_action: "approve",
            confidence: 1.0 - risk_score
        };
    }
}

下一步¶

DSL 规范 - 完整语言规范参考
运行时架构 - 了解智能体如何执行
安全模型 - 学习安全实现
API 参考 - 完整的函数和类型参考
示例 - 更多完整示例

准备构建您的第一个智能体？查看我们的入门指南或探索运行时示例。