示例库

仓库 examples/ 目录下有 24 个可运行示例:4 个语言特性示例 + 20 个办公场景示例,全部通过 flow check。每个示例的完整源码都内联在本页(默认折叠,点击展开),无需再翻仓库;examples/dist/ 是办公场景示例编译出的受限 JavaScript。

node packages/cli/dist/main.js check examples/delivery.flow
node packages/cli/dist/main.js run examples/simple.flow \
  --input '{"text": "hello agent flow"}'

语言特性示例

simple.flow — 最小 workflow

单个 agent 调用,请求 → 回答

查看完整代码

workflow hello(input: Request) -> Result {
  use agent "writer" as writer

  type Request {
    text: text
  }

  type Result {
    answer: text
  }

  pipeline {
    stage answer -> Result {
      return agent(writer) {
        task "Answer the request"
        input {
          request: input
        }
        tools none
        expect Result
      }
    }

    return answer
  }
}

parallel.flow — 并发执行 backlog

parallel map + limit 并发实现 backlog,team.member(feature.owner) 动态目标

查看完整代码

workflow build_features(input: Backlog) -> Bundle {
  use team "feature-team" as features

  type Feature {
    id: text
    owner: member<features>
    brief: text
  }

  type Backlog {
    features: Feature[1..4]
  }

  type Change {
    feature_id: text
    summary: text
  }

  type Bundle {
    changes: Change[]
    review: text
  }

  limits {
    concurrency: 3
    agent_runs: 10
    duration: 10m
  }

  pipeline deliver {
    stage changes -> Change[] {
      return parallel map input.features as feature limit 2 {
        agent(features.member(feature.owner)) {
          task "Implement the feature"
          input {
            feature
          }
          tools [read_file, write_file]
          expect Change
          timeout 5m
        }
      }
    }

    stage review(changes) -> text {
      return agent(features.main) {
        task "Review all changes and produce a verdict"
        input {
          changes
        }
        tools [read_file]
        expect text
      }
    }

    return {
      changes
      review
    }
  }
}

verify.flow — 并行检查与闸门

verify 并行检查组(测试 + 评审),pass when Runtime 判定,require 闸门

查看完整代码

workflow guarded_change(input: ChangeRequest) -> Outcome {
  use agent "implementer" as implementer
  use agent "tester" as tester
  use agent "reviewer" as reviewer

  type ChangeRequest {
    requirement: text
  }

  type Implementation {
    files: path[]
    summary: text
  }

  type TestReport {
    passed: bool
    command: text
  }

  type ReviewReport {
    blockers: text[]
  }

  type Outcome {
    passed: bool
    summary: text
  }

  pipeline {

    stage implement -> Implementation {
      return agent(implementer) {
        task "Implement the requirement"
        input {
          requirement: input.requirement
        }
        tools [read_file, write_file, exec]
        expect Implementation
        timeout 10m
      }
    }

    stage verification(implement) -> bool {
      let result = verify {
        parallel {
          check tests = agent(tester) {
            task "Run the test suite"
            input {
              implement
            }
            tools [read_file, exec]
            expect TestReport
          }
          check review = agent(reviewer) {
            task "Review for correctness and risk"
            input {
              implement
            }
            tools [read_file]
            expect ReviewReport
          }
        }
        pass when
          tests.passed
          and review.blockers.empty
      }
      require result.passed
        else fail "verification failed"
      return result.passed
    }

    return {
      passed: verification
      summary: implement.summary
    }
  }
}

delivery.flow — 旗舰示例

计划拆解 → 并行执行(disjoint 写范围互斥)→ 测试 → 安全评审

查看完整代码

workflow deliver_change(input: ChangeRequest) -> Delivery {
  use team "engineering-team" as engineering
  use agent "secure-code-reviewer" as security

  limits {
    concurrency: 4
    agent_runs: 12
    duration: 30m
  }

  type WorkItem {
    id: text
    owner: member<engineering>
    objective: text
    writes: path[]
    acceptance: text[]
  }

  type Plan {
    summary: text
    work: WorkItem[1..6]
  }

  type ChangeResult {
    work_id: text
    changed_files: path[]
    summary: text
    evidence: text[]
  }

  type TestResult {
    passed: bool
    command: text
    summary: text
  }

  type ReviewResult {
    blockers: text[]
    warnings: text[]
  }

  type Delivery {
    summary: text
    changes: ChangeResult[]
    tests: TestResult
    review: ReviewResult
  }

  pipeline delivery {

    stage plan -> Plan {
      let result = agent(engineering.main) {
        task """
        Analyze the requirement and the current project.

        Split the work into well-scoped, independently completable work items.

        Do not modify files.
        """

        input {
          request: input.request
          project: context.project
        }

        tools [read_file, list_dir, grep]

        expect Plan
        timeout 5m
      }

      require result.work.count in 1..6
        else fail "plan must contain 1 to 6 work items"

      require unique(result.work[*].id)
        else fail "work item ids must be unique"

      require disjoint(result.work[*].writes)
        else fail "parallel work items must not have overlapping write scopes"

      emit progress {
        phase: "plan"
        message: result.summary
      }

      return result
    }


    stage changes(plan) -> ChangeResult[] {
      return parallel map plan.work as item limit 4 {
        agent(engineering.member(item.owner)) {
          task """
          Complete the current work item.

          Only modify files under the declared writes paths.

          Return the changed files, an implementation summary and acceptance evidence.
          """

          input {
            request: input.request
            work: item
          }

          tools [
            read_file,
            list_dir,
            grep,
            write_file,
            edit_file,
            apply_patch,
            exec
          ]

          write item.writes

          expect ChangeResult
          timeout 10m

          retry {
            attempts: 1
            on: [timeout, rate_limit]
          }
        }
      }
    }


    stage verification(plan, changes) -> Verification {
      let result = verify {

        parallel {

          check tests = agent(
            engineering.member("test-engineer")
          ) {
            task """
            Execute real tests against the acceptance criteria.

            Do not modify product code.
            """

            input {
              plan
              changes
            }

            tools [
              read_file,
              list_dir,
              grep,
              exec
            ]

            expect TestResult
            timeout 10m
          }


          check security = agent(security) {
            task """
            Review the changes for security, architectural boundaries and privilege escalation risks.
            """

            input {
              plan
              changes
            }

            tools [
              read_file,
              list_dir,
              grep
            ]

            expect ReviewResult
            timeout 8m
          }
        }

        pass when
          tests.passed
          and security.blockers.empty
      }

      return result
    }


    require verification.passed
      else fail "final verification failed"


    stage summary(
      plan,
      changes,
      verification
    ) -> text {

      return agent(engineering.main) {
        task """
        Produce a concise delivery summary from the execution results.

        Do not alter test results or hide risks.
        """

        input {
          plan
          changes
          verification
        }

        tools none

        expect text
      }
    }


    return {
      summary
      changes
      tests: verification.checks.tests
      review: verification.checks.security
    }
  }
}

沟通与记录

email-triage.flow — 邮件分流与回复草拟

parallel map + limit 4、if/else 双路 emit progress

查看完整代码

// 场景: 批量分流收件箱邮件,分类并标记垃圾邮件,对需要回复的邮件逐封草拟回复
workflow triage_inbox(input: InboxSnapshot) -> TriageResult {
  use agent "mail-assistant" as assistant

  type Email {
    id: text
    from: text
    subject: text
    body: text
  }

  type InboxSnapshot {
    mailbox: text
    emails: Email[1..20]
  }

  type ClassifiedEmail {
    id: text
    category: text
    needs_reply: bool
    reason: text
  }

  type Draft {
    email_id: text
    to: text
    body: text
  }

  type TriageSummary {
    total: number
    spam_count: number
    urgent_count: number
  }

  type TriageResult {
    mailbox: text
    classified: ClassifiedEmail[]
    drafts: Draft[]
    summary: TriageSummary
  }

  limits {
    concurrency: 4
    agent_runs: 45
    duration: 20m
  }

  pipeline triage {
    stage classify -> ClassifiedEmail[] {
      let items = parallel map input.emails as item limit 4 {
        agent(assistant) {
          task """
          Classify one email from the inbox snapshot.

          Pick exactly one category: urgent, normal or spam. Mark needs_reply
          true only when the sender explicitly waits for an answer. Give a
          one sentence reason quoting the decisive cue.
          """
          input {
            email: item
          }
          tools none
          expect ClassifiedEmail
          timeout 2m
          retry { attempts: 1 on: [transient] }
        }
      }

      require unique(items[*].id)
        else fail "classified ids must stay unique"

      emit progress {
        phase: "classify"
        mailbox: input.mailbox
        total: items.count
      }

      return items
    }

    stage drafts(classify) -> Draft[] {
      let replies = agent(assistant) {
        task """
        Draft polite replies for the emails classified with needs_reply true.

        Return exactly one draft per such email and none for the others.
        Keep each reply under 150 words and reuse the language of the sender.
        """
        input {
          emails: input.emails
          classified: classify
        }
        tools none
        expect Draft[]
        timeout 6m
      }

      require replies.count <= classify.count
        else fail "cannot draft more replies than classified emails"

      require unique(replies[*].email_id)
        else fail "at most one reply per email"

      return replies
    }

    stage summary(classify, drafts) -> TriageSummary {
      let stats = agent(assistant) {
        task """
        Summarize the triage run. total is the number of classified emails,
        spam_count how many were marked spam, urgent_count how many urgent.
        Count strictly from the classified list, do not guess.
        """
        input {
          classified: classify
          drafted: drafts
        }
        tools none
        expect TriageSummary
        timeout 2m
      }

      require stats.total == classify.count
        else fail "summary total must match the classified emails"

      return stats
    }

    if summary.urgent_count > 0 {
      emit progress {
        phase: "done"
        mailbox: input.mailbox
        attention: true
        urgent: summary.urgent_count
      }
    } else {
      emit progress {
        phase: "done"
        mailbox: input.mailbox
        attention: false
        urgent: summary.urgent_count
      }
    }

    return {
      mailbox: input.mailbox
      classified: classify
      drafts: drafts
      summary: summary
    }
  }
}

meeting-notes.flow — 会议纪要与行动项

parallel 双分支、member<ops> 行动项归属

查看完整代码

// 场景: 把会议逐字稿蒸馏成结构化纪要,提取指派到人的行动项,并草拟跟进邮件
workflow write_minutes(input: MeetingRecord) -> Minutes {
  use team "ops-team" as ops

  type Attendee {
    name: text
    role: text
  }

  type MeetingRecord {
    title: text
    date: text
    transcript: text
    attendees: Attendee[2..12]
  }

  type Condensed {
    summary: text
    highlights: text[]
  }

  type ActionItem {
    description: text
    owner: member<ops>
    due: text
  }

  type Digest {
    summary: text
    highlights: text[]
    actions: ActionItem[]
  }

  type Minutes {
    title: text
    date: text
    summary: text
    highlights: text[]
    actions: ActionItem[]
    followup_email: text
  }

  limits {
    concurrency: 2
    agent_runs: 6
    duration: 15m
  }

  pipeline minutes {
    stage digest -> Digest {
      let parts = parallel {
        condensed = agent(ops.main) {
          task """
          Distill the meeting transcript. summary is a five sentence recap of
          the decisions and their context. highlights lists the two to five
          most important outcomes, one line each.
          """
          input {
            meeting: input
          }
          tools none
          expect Condensed
          timeout 5m
        }
        actions = agent(ops.main) {
          task """
          Extract the action items agreed in the meeting. Each item names one
          concrete deliverable, its owner as a team member id among the
          attendees, and a due date as stated, or the words "not stated".
          Skip vague intentions without an owner.
          """
          input {
            meeting: input
          }
          tools none
          expect ActionItem[]
          timeout 5m
        }
      }

      require parts.actions.count in 0..8
        else fail "keep the action list focused, at most 8 items"

      emit progress {
        meeting: input.title
        actions: parts.actions.count
      }

      return {
        summary: parts.condensed.summary
        highlights: parts.condensed.highlights
        actions: parts.actions
      }
    }

    stage followup(digest) -> text {
      let email = agent(ops.main) {
        task """
        Write the follow-up email for the meeting. Thank the attendees,
        restate each action item with its owner and due date as a checklist,
        and close with where to raise questions. Plain text, no markdown.
        """
        input {
          meeting: input
          minutes: digest
        }
        tools none
        expect text
        timeout 4m
      }

      return email
    }

    return {
      title: input.title
      date: input.date
      summary: digest.summary
      highlights: digest.highlights
      actions: digest.actions
      followup_email: followup
    }
  }
}

weekly-report.flow — 团队周报汇总

pipeline 级 verify + require passed、union(投影).empty

查看完整代码

// 场景: 把团队成员的零散周更新结构化,汇总亮点与风险,经质量闸门后生成周报
workflow compile_weekly_report(input: TeamUpdates) -> WeeklyReport {
  use agent "team-assistant" as assistant

  type MemberUpdate {
    member: text
    notes: text
    blockers: text[]
  }

  type TeamUpdates {
    week: text
    members: MemberUpdate[1..10]
  }

  type StructuredUpdate {
    member: text
    done: text[]
    planned: text[]
    blockers: text[]
  }

  type HighlightList {
    items: text[]
  }

  type RiskList {
    items: text[]
  }

  type WeeklyReport {
    week: text
    by_member: StructuredUpdate[]
    highlights: text[]
    risks: text[]
    narrative: text
  }

  limits {
    concurrency: 5
    agent_runs: 15
    duration: 15m
  }

  pipeline weekly {
    stage collect -> StructuredUpdate[] {
      let updates = parallel map input.members as item {
        agent(assistant) {
          task """
          Structure one member's raw weekly notes. done lists shipped work,
          planned lists next week's commitments, blockers lists impediments
          needing help. Use the member's own words where possible and never
          invent work items.
          """
          input {
            member: item
          }
          tools none
          expect StructuredUpdate
          timeout 2m
          retry { attempts: 1 on: [transient] }
        }
      }

      return updates
    }

    stage narrative(collect) -> text {
      let story = agent(assistant) {
        task """
        Write the narrative body of the weekly report: two short paragraphs
        for management, progress first, then risks and asks. Base every
        sentence on the structured updates.
        """
        input {
          week: input.week
          updates: collect
        }
        tools none
        expect text
        timeout 4m
      }

      return story
    }

    let quality = verify {
      check highlights = agent(assistant) {
        task """
        From all structured updates extract the week's key achievements.
        Two to five items, one line each, outcomes rather than activities.
        """
        input {
          updates: collect
        }
        tools none
        expect HighlightList
        timeout 3m
      }
      check risks = agent(assistant) {
        task """
        Aggregate every blocker into the risk list: merge duplicates, name
        the affected members and the help needed. Produce at least one entry
        whenever any blocker exists.
        """
        input {
          updates: collect
        }
        tools none
        expect RiskList
        timeout 3m
      }
      pass when highlights.items.any and (risks.items.any or union(collect[*].blockers).empty)
    }

    require quality.passed
      else fail "weekly report must surface highlights, and risks whenever blockers exist"

    return {
      week: input.week
      by_member: collect
      highlights: quality.checks.highlights.items
      risks: quality.checks.risks.items
      narrative: narrative
    }
  }
}

文档与内容

document-review.flow — 文档三维度并行审阅

parallel 三分支、verify 混合确定性/agent check

查看完整代码

// 场景: 风格、事实、合规三个审阅员并行审阅,合并发现项并由 Runtime 判定发布闸门
workflow review_document(input: ReviewRequest) -> ReviewReport {
  use agent "style-editor" as styler
  use agent "fact-checker" as facter
  use agent "policy-compliance" as policy

  type ReviewRequest {
    doc_path: path
    doc_text: text
    audience: text
  }

  type Finding {
    issue: text
    severity: text
    suggestion: text
  }

  type Findings {
    items: Finding[]
  }

  type Evidence {
    style: Findings
    accuracy: Findings
    compliance: Findings
  }

  type BlockingList {
    items: text[]
  }

  type ReviewReport {
    style: Finding[]
    accuracy: Finding[]
    compliance: Finding[]
    blocking: text[]
    verdict: text
  }

  limits {
    concurrency: 3
    agent_runs: 8
    duration: 20m
  }

  pipeline review {
    stage evidence -> Evidence {
      let found = parallel {
        style = agent(styler) {
          task """
          Review the document for style only: tone fit for the audience,
          clarity, consistent terminology, length discipline. severity is
          minor, major or critical. Offer one concrete suggestion per finding.
          """
          input {
            doc: input.doc_text
            audience: input.audience
          }
          tools none
          expect Findings
          timeout 5m
        }
        accuracy = agent(facter) {
          task """
          Check the document for factual and internal-consistency problems:
          numbers that contradict each other, impossible claims, undefined
          references. Quote the offending sentence in each issue.
          """
          input {
            doc: input.doc_text
          }
          tools none
          expect Findings
          timeout 6m
          retry { attempts: 1 on: [timeout] }
        }
        compliance = agent(policy) {
          task """
          Check the document against company policy: confidential data, legal
          claims, regulated wording. severity critical is reserved for
          must-fix violations before publication.
          """
          input {
            doc: input.doc_text
          }
          tools none
          expect Findings
          timeout 5m
          retry { attempts: 1 on: [transient, rate_limit] }
        }
      }

      emit progress {
        doc: input.doc_path
        style: found.style.items.count
        accuracy: found.accuracy.items.count
        compliance: found.compliance.items.count
      }

      return found
    }

    stage report(evidence) -> ReviewReport {
      let gate = verify {
        parallel {
          check style_clean = evidence.style.items.empty
          check compliance_clean = evidence.compliance.items.empty
        }
        check blockers = agent(facter) {
          task """
          From the three finding lists select the issues that must block
          publication: any critical severity, or any compliance or accuracy
          violation. Return them as one-line statements. Empty list when
          nothing blocks.
          """
          input {
            evidence: evidence
          }
          tools none
          expect BlockingList
          timeout 3m
        }
        pass when blockers.items.empty and style_clean and compliance_clean
      }

      if gate.passed {
        emit progress {
          doc: input.doc_path
          gate: "passed"
        }
      } else {
        emit progress {
          doc: input.doc_path
          gate: "blocked"
        }
      }

      let verdict = agent(styler) {
        task """
        Write the verdict paragraph for the review report. Explain what the
        reviewers found and what the author should change first. When the
        gate failed, lead with the blocking issues. Three to five sentences.
        """
        input {
          evidence: evidence
          blocking: gate.checks.blockers.items
          passed: gate.passed
        }
        tools none
        expect text
        timeout 3m
      }

      return {
        style: evidence.style.items
        accuracy: evidence.accuracy.items
        compliance: evidence.compliance.items
        blocking: gate.checks.blockers.items
        verdict: verdict
      }
    }

    return report
  }
}

translation-flow.flow — 文档分段并行翻译

union([投影]) 数组拼接、术语一致性审计

查看完整代码

// 场景: 长文档按章节并行翻译,合并全文后再由术语审计 agent 依据术语表统一措辞并输出修订版。
workflow translate_document(input: TranslationJob) -> TranslationResult {
  use agent "section-translator" as translator
  use agent "translation-editor" as editor
  use agent "terminology-auditor" as auditor

  limits {
    concurrency: 3
    agent_runs: 60
    duration: 30m
  }

  type Section {
    id: text
    source_text: text
  }

  type TranslationJob {
    source_lang: text
    target_lang: text
    glossary: text[]
    sections: Section[1..40]
  }

  type TranslatedSection {
    id: text
    translated: text
  }

  type MergedDraft {
    full_text: text
    boundary_notes: text[]
  }

  type FinalDoc {
    full_text: text
    glossary_issues: text[]
  }

  type TranslationResult {
    translated_sections: TranslatedSection[]
    full_text: text
    glossary_issues: text[]
  }

  pipeline translation {
    stage translate -> TranslatedSection[] {
      require unique(input.sections[*].id)
        else fail "section ids must be unique before translation starts"

      let results = parallel map input.sections as item limit 3 {
        agent(translator) {
          task """
          Translate one section of a longer document.

          Follow the glossary exactly: each glossary term must keep the glossary rendering.
          Never summarize or omit content; keep section boundaries intact for ordered rejoining.
          """

          input {
            source_lang: input.source_lang
            target_lang: input.target_lang
            glossary: input.glossary
            section: item
          }

          tools none
          expect TranslatedSection
          timeout 5m
        }
      }

      require results.count == input.sections.count
        else fail "every section must yield exactly one translated section"
      return results
    }

    stage merge(translate) -> MergedDraft {
      require unique(translate[*].id)
        else fail "translated section ids must stay unique after parallel mapping"

      let chunks = union([translate[*].translated])
      require not chunks.empty
        else fail "merge received no translated chunks"

      emit progress {
        phase: "merge"
        sections: chunks.count
      }

      return agent(editor) {
        task """
        Join the translated chunks into one coherent document.

        Keep the chunk order, repair punctuation and whitespace at section
        boundaries, and flag sentences that were cut across adjacent sections.
        """

        input {
          target_lang: input.target_lang
          chunks
        }

        tools none
        expect MergedDraft
        timeout 5m
      }
    }

    stage consistency(translate, merge) -> FinalDoc {
      return agent(auditor) {
        task """
        Audit the merged translation against the glossary.

        Every glossary term must be rendered the same way across all sections.
        Return the revised full text with unified terminology, plus one glossary_issues
        entry per fix: term, divergent rendering, section id where it appeared.
        """

        input {
          source_lang: input.source_lang
          target_lang: input.target_lang
          glossary: input.glossary
          sections: translate
          draft: merge
        }

        tools none
        expect FinalDoc
        timeout 8m
      }
    }

    return {
      translated_sections: translate
      full_text: consistency.full_text
      glossary_issues: consistency.glossary_issues
    }
  }
}

release-notes.flow — 发布说明生成

投影 + .any 检测破坏性变更、有界数组 [1..60]

查看完整代码

// 场景: 变更条目批量归类为用户可感知的分类,自动生成发布说明,破坏性变更时附迁移指引
workflow write_release_notes(input: ReleaseInput) -> ReleaseNotes {
  use agent "release-writer" as writer

  type ChangeEntry {
    id: text
    kind: text
    title: text
    author: text
  }

  type ReleaseInput {
    product: text
    version: text
    changes: ChangeEntry[1..60]
  }

  type UserFacing {
    id: text
    category: text
    summary: text
    breaking: bool
  }

  type ReleaseNotes {
    version: text
    highlights: text[]
    breaking_changes: text[]
    fixes: text[]
    notes_markdown: text
  }

  limits {
    concurrency: 5
    agent_runs: 70
    duration: 25m
  }

  pipeline notes {
    stage classify -> UserFacing[] {
      let mapped = parallel map input.changes as item limit 5 {
        agent(writer) {
          task """
          Classify one change entry for the release notes. category is one of
          feature, fix, breaking, internal. breaking is true only when
          existing user-facing behavior, output or APIs change. summary is a
          one line sentence from the user's perspective, not the PR title.
          """
          input {
            product: input.product
            change: item
          }
          tools none
          expect UserFacing
          timeout 2m
        }
      }

      require unique(mapped[*].id)
        else fail "each change entry must be classified exactly once"

      emit progress {
        version: input.version
        changes: mapped.count
        breaking: mapped[*].breaking.any
      }

      return mapped
    }

    stage notes(classify) -> ReleaseNotes {
      let has_breaking = classify[*].breaking.any

      if has_breaking {
        emit progress {
          version: input.version
          migration_section: true
        }
      } else {
        emit progress {
          version: input.version
          migration_section: false
        }
      }

      let draft = agent(writer) {
        task """
        Write the release notes markdown from the classified changes.

        Sections: Highlights (features), Fixes, Breaking changes. Skip
        internal entries entirely. When has_breaking is true add a final
        "Migration steps" section with concrete upgrade actions. One bullet
        per line, plain markdown.
        """
        input {
          product: input.product
          version: input.version
          classified: classify
          has_breaking: has_breaking
        }
        tools none
        expect ReleaseNotes
        timeout 5m
      }

      require draft.fixes.count <= classify.count
        else fail "fix bullets cannot exceed classified changes"

      return draft
    }

    return notes
  }
}

social-calendar.flow — 一周社媒日历

有界输入 text[5..7]、verify 双 check

查看完整代码

// 场景: 生成下周社媒内容日历,先定每天选题,再并发写帖,品牌口吻与一天一帖由 verify 把关
workflow plan_social_week(input: CampaignBrief) -> ContentCalendar {
  use agent "content-strategist" as strategist

  type CampaignBrief {
    brand: text
    product: text
    audience: text
    goals: text[]
    days: text[5..7]
  }

  type IdeaHook {
    day: text
    hook: text
    angle: text
  }

  type SocialPost {
    day: text
    platform: text
    copy: text
    hashtags: text[]
  }

  type VoiceCheck {
    ok: bool
    notes: text[]
  }

  type ContentCalendar {
    posts: SocialPost[]
    review_notes: text[]
    calendar_markdown: text
  }

  limits {
    concurrency: 4
    agent_runs: 16
    duration: 20m
  }

  pipeline calendar {
    stage ideas -> IdeaHook[] {
      let hooks = agent(strategist) {
        task """
        Propose the content ideas for next week: exactly one per scheduled
        day, in the same order. hook is a scroll-stopping first line, angle
        is the value it delivers to the audience and which campaign goal it
        serves. Vary the formats across the week.
        """
        input {
          brief: input
        }
        tools none
        expect IdeaHook[]
        timeout 4m
      }

      require hooks.count == input.days.count
        else fail "one idea per scheduled day"

      require unique(hooks[*].day)
        else fail "each day needs exactly one idea"

      return hooks
    }

    stage posts(ideas) -> SocialPost[] {
      let written = parallel map ideas as item {
        agent(strategist) {
          task """
          Write the social post for one day from its hook. copy fits the
          platform conventions and stays under 280 characters, hashtags has
          at most three entries, no hype words banned by the brand.
          """
          input {
            brief: input
            idea: item
          }
          tools none
          expect SocialPost
          timeout 3m
        }
      }

      return written
    }

    stage review(posts) -> ContentCalendar {
      let audit = verify {
        check one_per_day = unique(posts[*].day)
        check brand_voice = agent(strategist) {
          task """
          Check every post against the brand voice: consistent tone, no
          banned hype words, claims the product can keep. ok true when all
          posts pass, notes lists the required rewrites otherwise.
          """
          input {
            brief: input
            posts: posts
          }
          tools none
          expect VoiceCheck
          timeout 4m
        }
        pass when one_per_day and brand_voice.ok
      }

      require audit.passed
        else fail "calendar must keep one post per day and the brand voice"

      let calendar = agent(strategist) {
        task """
        Assemble the final content calendar: the reviewed posts, the review
        notes for the social media manager and a markdown table of the week
        with day, platform, copy and hashtags.
        """
        input {
          brief: input
          posts: posts
          notes: audit.checks.brand_voice.notes
        }
        tools none
        expect ContentCalendar
        timeout 3m
      }

      return calendar
    }

    return review
  }
}

kb-audit.flow — 知识库审计与更新

write input.write_scope 写范围、读写工具分离

查看完整代码

// 场景: 知识库文章并发逐篇判定是否过时,过时文章交给限定写范围的 agent 重写,产出审计报告
workflow audit_knowledge_base(input: KbAuditRequest) -> KbAuditReport {
  use agent "kb-curator" as curator

  type ArticleRef {
    id: text
    title: text
    last_updated: text
  }

  type KbAuditRequest {
    kb_path: path
    write_scope: path[]
    articles: ArticleRef[1..30]
  }

  type ArticleVerdict {
    id: text
    status: text
    stale_parts: text[]
    rewrite_needed: bool
  }

  type RewriteOutcome {
    id: text
    updated: bool
    change_note: text
  }

  type KbAuditReport {
    audited: number
    verdicts: ArticleVerdict[]
    rewrites: RewriteOutcome[]
    summary: text
  }

  limits {
    concurrency: 4
    agent_runs: 80
    duration: 30m
  }

  pipeline audit {
    stage audit -> ArticleVerdict[] {
      let verdicts = parallel map input.articles as item limit 4 {
        agent(curator) {
          task """
          Judge whether one knowledge base article is stale. status is one of
          current, minor-drift, outdated. stale_parts names the specific
          sections that aged and why. rewrite_needed is true only for
          outdated articles.
          """
          input {
            kb: input.kb_path
            article: item
          }
          tools [read_file, grep]
          expect ArticleVerdict
          timeout 3m
        }
      }

      require unique(verdicts[*].id)
        else fail "each article must be audited exactly once"

      return verdicts
    }

    stage rewrite(audit) -> RewriteOutcome[] {
      let outcomes = parallel map audit as item limit 2 {
        agent(curator) {
          task """
          Update one knowledge base article. When rewrite_needed is false,
          return updated false with an empty change_note and touch nothing.
          Otherwise rewrite only the stale parts, keep the article id and
          structure, and summarize what changed in change_note.
          """
          input {
            kb: input.kb_path
            article: item
          }
          tools [read_file, write_file]
          write input.write_scope
          expect RewriteOutcome
          timeout 6m
          retry { attempts: 1 on: [transient] }
        }
      }

      require unique(outcomes[*].id)
        else fail "each article gets at most one rewrite outcome"

      emit progress {
        kb: input.kb_path
        audited: audit.count
        rewritten: outcomes.count
      }

      return outcomes
    }

    stage summarize(audit, rewrite) -> KbAuditReport {
      let report = agent(curator) {
        task """
        Summarize the knowledge base audit: audited is the number of audited
        articles, verdicts and rewrites are passed through unchanged.
        summary states how many articles are current versus outdated, what
        was rewritten and which topics still need a subject-matter owner.
        """
        input {
          kb: input.kb_path
          verdicts: audit
          rewrites: rewrite
        }
        tools none
        expect KbAuditReport
        timeout 3m
      }

      return report
    }

    return summarize
  }
}

财务与法务

invoice-processing.flow — 发票提取与报销路由

unique+union 锚点模拟 every、in 1..50000 范围

查看完整代码

// 场景: 批量读取发票图片提取费用信息,逐张校验金额与有效性,并按单张金额阈值路由到直接报销或人工审批。
workflow invoice_processing(input: InvoiceBatch) -> ExpenseClaim {
  use agent "invoice-reader" as reader
  use agent "finance-preparer" as preparer
  use agent "finance-approver" as approver

  limits {
    concurrency: 5
    agent_runs: 40
  }

  type InvoiceScan {
    id: text
    image_ref: text
  }

  type InvoiceBatch {
    submitter: text
    invoices: InvoiceScan[1..30]
  }

  type InvoiceData {
    invoice_id: text
    vendor: text
    date: text
    amount: number
    category: text
    valid: bool
  }

  type AuditReport {
    items: InvoiceData[]
    needs_manual_review: bool
    total_note: text
  }

  type ExpenseClaim {
    items: InvoiceData[]
    total_note: text
    needs_manual_review: bool
    note: text
  }

  pipeline reimburse {
    stage extract -> InvoiceData[] {
      require unique(input.invoices[*].id)
        else fail "invoice ids must be unique within a batch"

      emit progress { phase: "extract", message: "reading invoice images",
        invoices: input.invoices[*].id }
      return parallel map input.invoices as item limit 5 {
        agent(reader) {
          task """
          Read the invoice image and extract its expense data. Return exactly:
          - invoice_id, vendor and date (YYYY-MM-DD) as printed on the document
          - amount: the grand total as a plain number, no currency symbol
          - category: one of travel, meal, office, software, other
          - valid: false only when the image is unreadable or looks forged
          Never guess values that are not visible on the document.
          """
          input { scan: item }
          tools [read_file]
          expect InvoiceData
          timeout 2m
        }
      }
    }

    stage audit(extract) -> AuditReport {
      require extract.count == input.invoices.count
        else fail "every invoice must yield one extracted record"

      let in_range_flags = parallel map extract as it limit 5 { it.amount in 1..50000 }
      let over_limit_flags = parallel map extract as it limit 5 { it.amount > 5000 }

      // 语言无 every/any 内建,用 unique+union 锚点表达布尔聚合:
      // 先拼接一个 true 锚点,全表元素一致(即全为 true)时 unique 为真 => "所有 flag 均为真"。
      require unique(union([[true], in_range_flags]))
        else fail "every invoice amount must stay within 1..50000"
      require unique(union([[true], extract[*].valid]))
        else fail "every invoice must be readable and valid"

      // 改用 false 锚点:全表一致(即全为 false)表示无人超限,取 not 即 "至少一张超阈值"。
      let needs_manual_review = not unique(union([[false], over_limit_flags]))

      emit progress { phase: "audit", message: "batch audited", review: needs_manual_review }

      let total_note = agent(preparer) {
        task """
        Summarize the audited batch for the expense report: invoice count,
        expense total, and the share of each category. Use only the provided
        records; do not invent numbers.
        """
        input { submitter: input.submitter, invoices: extract }
        tools none
        expect text
        timeout 2m
      }

      return { items: extract, needs_manual_review, total_note }
    }

    stage route(audit) -> text {
      if audit.needs_manual_review {
        emit progress { phase: "route", decision: "manual_review" }

        return agent(approver) {
          task """
          Write the manual approval request for this expense batch: list every
          invoice above the 5000 CNY single-invoice threshold with vendor, date,
          amount and category, and state what the approver should confirm.
          """
          input { submitter: input.submitter, invoices: audit.items }
          tools none
          expect text
          timeout 3m
        }
      } else {
        emit progress { phase: "route", decision: "direct_reimbursement" }
      }
      return agent(preparer) {
        task """
        Produce the direct reimbursement list: one line per invoice with
        invoice_id, vendor, date, amount and category, then the batch total
        and the payee submitter name.
        """
        input { submitter: input.submitter, invoices: audit.items }
        tools none
        expect text
        timeout 3m
      }
    }

    return {
      items: audit.items
      total_note: audit.total_note
      needs_manual_review: audit.needs_manual_review
      note: route
    }
  }
}

contract-review.flow — 合同条款与风险标记

parallel 三分支、嵌套 type

查看完整代码

// 场景: 合同条款抽取与风险标记 —— 抽取关键条款,三路并行分析义务/风险/数值指标,双检查筛查后输出红线修改建议
workflow review_contract(input: ContractReviewRequest) -> ContractReview {
  use agent "legal-analyst" as legal

  limits { concurrency: 3, agent_runs: 8, duration: 15m }

  type ContractReviewRequest { contract_name: text, contract_text: text, our_role: text }

  // 嵌套数据结构: 各 stage 产出与最终报告引用的字段类型
  type Clause { kind: text, summary: text, location: text }
  type Obligation { description: text, deadline: text }
  type RiskItem { level: text, description: text, mitigation: text }
  type Metric { name: text, value: number, unit: text }
  type Analysis { obligations: Obligation[], risks: RiskItem[], metrics: Metric[] }
  type Audit { findings: text[] }
  type Verdict { redlines: text[], summary: text }

  type ContractReview {
    clauses: Clause[]
    obligations: Obligation[]
    risks: RiskItem[]
    metrics: Metric[]
    redlines: text[]
    verdict: text
  }

  pipeline review {
    stage extract -> Clause[1..20] {
      let clauses = agent(legal) {
        task """
        Read the contract text and extract the clauses that decide deal risk:
        parties, term, payment, delivery, liability, termination, IP,
        confidentiality and dispute resolution.
        One entry per clause kind: name the kind, summarize it in plain
        language and cite its location (section or paragraph).
        """
        input { contract_name: input.contract_name, contract_text: input.contract_text }
        tools none
        expect Clause[1..20]
        timeout 5m
        retry { attempts: 1 on: [timeout, transient] }
      }

      require clauses.count in 3..20 else fail "a reviewable contract should yield at least 3 clauses"
      require unique(clauses[*].kind) else fail "each clause kind may appear only once"

      return clauses
    }

    stage analyze(extract) -> Analysis {
      return parallel {
        obligations = agent(legal) {
          task """
          List every obligation the contract puts on the party named in our_role:
          payments, deliverables, notices and reporting duties. For each one
          record what must be done and the deadline or trigger that governs it.
          """
          input { our_role: input.our_role, contract_text: input.contract_text, clauses: extract }
          tools none
          expect Obligation[]
          timeout 4m
        }

        risks = agent(legal) {
          task """
          Identify clauses that are risky for the party named in our_role.
          Rate each risk high, medium or low, describe the exposure and
          propose a concrete mitigation edit worth raising before signing.
          """
          input { our_role: input.our_role, contract_text: input.contract_text, clauses: extract }
          tools none
          expect RiskItem[]
          timeout 4m
        }

        metrics = agent(legal) {
          task """
          Pull the numeric terms out of the contract: total price, payment
          milestones, term length, penalty rates, liability caps and notice
          periods. One entry per number: plain name, numeric value and unit.
          """
          input { contract_text: input.contract_text, clauses: extract }
          tools none
          expect Metric[]
          timeout 4m
        }
      }
    }

    stage verdict(extract, analyze) -> Verdict {
      let screening = verify {
        parallel {
          check critical = agent(legal) {
            task """
            Check the extracted clauses for fatal gaps that expose our side to
            unbounded loss: no liability cap, no termination right, no payment
            precondition, no IP ownership rule. One finding per gap, empty if none.
            """
            input { our_role: input.our_role, clauses: extract }
            tools none
            expect Audit
            timeout 3m
          }

          check balanced = agent(legal) {
            task """
            Judge whether duties and remedies are balanced between the parties.
            Look for asymmetric penalties, sole-discretion rights, unlimited
            indemnity from our side, or exit rights only for the counterparty.
            Return one finding text per imbalance, empty when the deal is fair.
            """
            input { our_role: input.our_role, obligations: analyze.obligations, risks: analyze.risks }
            tools none
            expect Audit
            timeout 3m
          }
        }
        pass when critical.findings.empty and balanced.findings.empty
      }

      let redlines = union([screening.checks.critical.findings, screening.checks.balanced.findings])
      let summary = agent(legal) {
        task """
        Write the final review verdict from the position of our_role: sign,
        negotiate or reject. Lead with the highest risks and their clause
        locations, then turn each redline into a concrete edit. Never omit one.
        """
        input { contract_name: input.contract_name, our_role: input.our_role, clauses: extract, analysis: analyze, redlines }
        tools none
        expect text
        timeout 5m
      }
      return { redlines, summary }
    }

    return { clauses: extract, obligations: analyze.obligations, risks: analyze.risks,
      metrics: analyze.metrics, redlines: verdict.redlines, verdict: verdict.summary }
  }
}

客户与增长

support-triage.flow — 客服工单分类升级

team.main 分类、team.member(owner_id) 升级路由

查看完整代码

// 场景: 客服工单自动分类分级并草拟回复,需要升级时动态路由到负责人生成交接说明
workflow triage_ticket(input: Ticket) -> TicketResult {
  use team "support-team" as support

  type Ticket {
    id: text
    customer: text
    subject: text
    body: text
    channel: text
  }

  type Classification {
    category: text
    severity: number
    needs_escalation: bool
    owner_id: text
  }

  type TicketResult {
    category: text
    severity: number
    reply: text
    escalated: bool
    handover: text
  }

  pipeline triage {
    stage classify -> Classification {
      let result = agent(support.main) {
        task """
        Classify the incoming support ticket.

        Pick one category from billing, technical, account, product or other.
        Rate severity from 1 (minor question) to 4 (customer blocked, outage,
        data loss or security issue). Set needs_escalation to true only for
        severity 4 or an angry repeat contact, and then fill owner_id with
        the team member who should own the escalation.
        Use only facts stated in the ticket.
        """
        input {
          ticket: input
        }
        tools none
        expect Classification
        timeout 3m
      }

      require result.severity in 1..4
        else fail "severity must stay within 1..4"

      emit progress {
        phase: "classify"
        ticket: input.id
        category: result.category
        severity: result.severity
      }

      return result
    }

    stage draft(classify) -> text {
      let reply = agent(support.main) {
        task """
        Draft a customer-facing reply based on the ticket and its classification.

        Greet the customer by name, acknowledge the reported problem in one
        sentence, then state the next concrete step. Do not promise unverified
        timelines and never mention internal severity or escalation.
        """
        input {
          ticket: input
          classification: classify
        }
        tools none
        expect text
        timeout 3m
      }

      emit progress {
        phase: "draft"
        ticket: input.id
      }

      return reply
    }

    stage escalate(classify, draft) -> text {
      if classify.needs_escalation {
        let note = agent(support.member(classify.owner_id)) {
          task """
          Write an internal handover note for this escalated ticket.

          Summarize the customer issue, the classification decision and the
          draft reply, then list open questions and the recommended first
          action for the receiving owner.
          """
          input {
            ticket: input
            classification: classify
            draft_reply: draft
          }
          tools none
          expect text
          timeout 3m
        }

        emit progress {
          phase: "escalate"
          ticket: input.id
          owner: classify.owner_id
        }

        return note
      } else {
        emit progress {
          phase: "escalate"
          ticket: input.id
          skipped: true
        }
      }

      return ""
    }

    emit progress {
      phase: "complete"
      ticket: input.id
      escalated: classify.needs_escalation
    }

    return {
      category: classify.category
      severity: classify.severity
      reply: draft
      escalated: classify.needs_escalation
      handover: escalate
    }
  }
}

lead-qualification.flow — 线索评估分级

tools [web_search]、verify 审计闸门

查看完整代码

// 场景: 销售线索批量补全与 BANT 打分,生成排序与 AE 交接包,verify 审计分数合法性
workflow qualify_leads(input: LeadPacket) -> QualificationResult {
  use agent "sales-analyst" as analyst
  use team "sales-team" as sales

  type Lead {
    company: text
    contact: text
    source: text
    notes: text
  }

  type LeadPacket {
    campaign: text
    leads: Lead[1..20]
  }

  type EnrichedLead {
    company: text
    industry: text
    headcount_band: text
    signals: text[]
  }

  type ScoredLead {
    company: text
    score: number
    tier: text
    reason: text
  }

  type Handoff {
    hot_count: number
    hot_leads: text[]
    brief: text
    nurture: text
  }

  type AuditCheck {
    ok: bool
    flags: text[]
  }

  type QualificationResult {
    campaign: text
    scored: ScoredLead[]
    hot_handoff: bool
    handoff: Handoff
  }

  limits {
    concurrency: 4
    agent_runs: 45
    duration: 30m
  }

  pipeline qualify {
    stage enrich -> EnrichedLead[] {
      let enriched = parallel map input.leads as item {
        agent(analyst) {
          task """
          Enrich one sales lead from its notes: industry, headcount band
          (1-10, 11-50, 51-200, 200+) and the buying signals you can actually
          observe. signals must quote evidence from the notes, never
          assumptions.
          """
          input {
            lead: item
          }
          tools [web_search]
          expect EnrichedLead
          timeout 3m
          retry { attempts: 1 on: [rate_limit] }
        }
      }

      return enriched
    }

    stage score(enrich) -> ScoredLead[] {
      let ranked = agent(analyst) {
        task """
        BANT score every enriched lead. score is an integer from 1 to 100.
        tier is hot for 80 and above, warm for 50 to 79, cold below 50.
        reason is one clause naming the decisive factor. Keep the input
        order, one ScoredLead per company.
        """
        input {
          campaign: input.campaign
          leads: input.leads
          enriched: enrich
        }
        tools none
        expect ScoredLead[]
        timeout 5m
      }

      require ranked.count == enrich.count
        else fail "every lead must be scored"

      require unique(ranked[*].company)
        else fail "companies must appear exactly once in the ranking"

      return ranked
    }

    stage route(score) -> Handoff {
      let plan = agent(sales.main) {
        task """
        Turn the scores into a routing plan. hot_count is the number of hot
        tier leads. When there is at least one, hot_leads lists the companies
        and brief is an AE handoff pack: context, pain signals, suggested
        opening move. nurture is the drip campaign suggestion for the rest.
        """
        input {
          campaign: input.campaign
          scored: score
        }
        tools none
        expect Handoff
        timeout 4m
      }

      let gate = verify {
        check bounds = agent(analyst) {
          task """
          Audit the scored leads: every score between 1 and 100, tiers
          consistent with the score bands, hot_count equal to the number of
          hot tier leads. ok true when everything checks out, otherwise list
          the problems in flags.
          """
          input {
            scored: score
            handoff: plan
          }
          tools none
          expect AuditCheck
          timeout 3m
        }
        pass when bounds.ok
      }

      require gate.passed
        else fail "lead scores failed the audit gate"

      return plan
    }

    if route.hot_count > 0 {
      emit progress {
        campaign: input.campaign
        handoff: true
        hot: route.hot_count
      }
    } else {
      emit progress {
        campaign: input.campaign
        handoff: false
        hot: route.hot_count
      }
    }

    return {
      campaign: input.campaign
      scored: score
      hot_handoff: route.hot_count > 0
      handoff: route
    }
  }
}

content-moderation.flow — UGC 批量合规审核

verify 内 parallel 检查组、90s 超时

查看完整代码

// 场景: 用户生成内容批量并发合规审核,逐条出裁决,verify 并行检查组复核覆盖率与边界样本
workflow moderate_batch(input: ModerationBatch) -> ModerationReport {
  use agent "content-moderator" as moderator

  type ContentItem {
    id: text
    author: text
    body: text
  }

  type ModerationBatch {
    platform: text
    items: ContentItem[1..50]
  }

  type ModerationVerdict {
    id: text
    allowed: bool
    violations: text[]
    confidence: number
  }

  type CoverageCheck {
    ok: bool
    gaps: text[]
  }

  type EdgeCheck {
    ok: bool
    suspects: text[]
  }

  type ModerationReport {
    platform: text
    verdicts: ModerationVerdict[]
    rejected_count: number
    policy_note: text
  }

  limits {
    concurrency: 6
    agent_runs: 60
    duration: 25m
  }

  pipeline moderate {
    stage screen -> ModerationVerdict[] {
      let verdicts = parallel map input.items as item limit 6 {
        agent(moderator) {
          task """
          Review one piece of user generated content against the platform
          policy: no harassment, hate speech, spam, adult content or exposed
          personal data. Set allowed false when any violation exists and name
          the violated policy in violations. Confidence is between 0 and 1.
          """
          input {
            platform: input.platform
            item: item
          }
          tools none
          expect ModerationVerdict
          timeout 90s
          retry { attempts: 1 on: [transient] }
        }
      }

      require unique(verdicts[*].id)
        else fail "every content item must receive exactly one verdict"

      emit progress {
        platform: input.platform
        reviewed: verdicts.count
      }

      return verdicts
    }

    stage summarize(screen) -> ModerationReport {
      let gate = verify {
        parallel {
          check coverage = agent(moderator) {
            task """
            Confirm the moderation run is complete: every submitted item id
            appears exactly once among the verdicts. ok true when nothing is
            missing, otherwise list the missing ids in gaps.
            """
            input {
              expected: input.items[*].id
              verdicts: screen
            }
            tools none
            expect CoverageCheck
            timeout 2m
          }
          check edge = agent(moderator) {
            task """
            Spot-check the boundary cases: verdicts marked allowed with
            confidence below 0.8. ok true when none of them actually violates
            policy, otherwise collect their ids in suspects.
            """
            input {
              verdicts: screen
              items: input.items
            }
            tools none
            expect EdgeCheck
            timeout 3m
          }
        }
        pass when coverage.ok and edge.ok
      }

      require gate.passed
        else fail "moderation gate failed: coverage or edge check not satisfied"

      let report = agent(moderator) {
        task """
        Summarize the batch for the trust and safety dashboard. rejected_count
        is the number of verdicts with allowed false. policy_note names the
        most frequent violations and any policy reminder for the community.
        """
        input {
          platform: input.platform
          verdicts: screen
          suspects: gate.checks.edge.suspects
        }
        tools none
        expect ModerationReport
        timeout 3m
      }

      return report
    }

    return summarize
  }
}

人力与团队

onboarding-plan.flow — 新员工入职计划

team.member 常量路由、对象简写字段

查看完整代码

// 场景: 新员工入职——生成首周工作日计划、账号与设备清单、欢迎邮件草稿。
workflow onboarding_plan(input: OnboardingRequest) -> OnboardingPlan {
  use team "people-ops" as hr

  type OnboardingRequest {
    name: text
    role: text
    start_date: text
    team_name: text
  }

  type DayPlan {
    day: number
    agenda: text[]
    owner: member<hr>
  }

  type Logistics {
    accounts: text[]
    equipment: text[]
  }

  type OnboardingPlan {
    plan: DayPlan[]
    accounts: text[]
    equipment: text[]
    welcome_email: text
  }

  pipeline onboard {

    stage plan -> DayPlan[] {
      let days = agent(hr.main) {
        task """
        Draft the day-by-day onboarding schedule for the new hire.

        Cover the first five working days after the start date. Every day
        needs concrete agenda items and a people-ops owner who runs it.
        """
        input {
          request: input
        }
        tools none
        expect DayPlan[]
        timeout 5m
      }

      require days.count in 1..10
        else fail "onboarding plan must cover between 1 and 10 workdays"

      require unique(days[*].day)
        else fail "each workday number may appear only once"

      return days
    }

    stage logistics(plan) -> Logistics {
      let result = parallel {
        accounts = agent(hr.main) {
          task """
          List every account and permission the new hire needs on day one.

          Include email, chat, calendar, code hosting, ticketing and any
          role-specific system, one entry per line with the system named.
          """
          input {
            request: input
            plan
          }
          tools none
          expect text[]
          timeout 5m
        }

        equipment = agent(hr.member("it-support")) {
          task """
          List the hardware to prepare before the start date.

          Base the list on the role and the daily agenda: laptop, monitor,
          headset, badge and anything the schedule depends on.
          """
          input {
            request: input
            plan
          }
          tools none
          expect text[]
          timeout 5m
        }
      }

      require not result.accounts.empty
        else fail "at least one account entry is required"

      require not result.equipment.empty
        else fail "at least one equipment entry is required"

      return result
    }

    stage welcome(plan, logistics) -> text {
      let draft = agent(hr.main) {
        task """
        Write the welcome email draft for the new hire.

        Greet the new hire by name, walk through the first five days at a
        glance, confirm accounts and equipment arrive before the start
        date, and name the people-ops contact for questions.
        """
        input {
          request: input
          plan
          logistics
        }
        tools none
        expect text
        timeout 5m
      }

      emit progress {
        phase: "welcome"
        message: "welcome email drafted"
      }

      return draft
    }

    return {
      plan
      accounts: logistics.accounts
      equipment: logistics.equipment
      welcome_email: welcome
    }
  }
}

recruitment-screening.flow — 简历批量筛选

parallel map + limit 5、.count == 完整性核对

查看完整代码

// 场景: 简历按岗位要求并发筛选打分并排名,verify 并行复核评分公平性与覆盖完整性
workflow screen_resumes(input: ScreeningBatch) -> ScreeningResult {
  use agent "talent-screener" as screener

  type Resume {
    candidate_id: text
    name: text
    resume_text: text
  }

  type ScreeningBatch {
    requisition: text
    must_haves: text[]
    nice_to_haves: text[]
    resumes: Resume[3..40]
  }

  type CandidateScore {
    candidate_id: text
    name: text
    meets_must_haves: bool
    match_score: number
    highlights: text[]
    concerns: text[]
  }

  type RankedCandidate {
    candidate_id: text
    rank: number
    recommendation: text
  }

  type FairnessCheck {
    ok: bool
    flags: text[]
  }

  type CoverageCheck {
    ok: bool
    missing: text[]
  }

  type ScreeningResult {
    requisition: text
    scores: CandidateScore[]
    ranking: RankedCandidate[]
    interview_slots: text[]
  }

  limits {
    concurrency: 5
    agent_runs: 50
    duration: 30m
  }

  pipeline screening {
    stage score -> CandidateScore[] {
      let scored = parallel map input.resumes as item limit 5 {
        agent(screener) {
          task """
          Screen one resume against the requisition. meets_must_haves is true
          only when every must-have is evidenced. match_score is an integer
          from 1 to 100. highlights quotes the strongest evidence, concerns
          lists gaps or inconsistencies. Never infer facts absent from the
          resume.
          """
          input {
            requisition: input.requisition
            must_haves: input.must_haves
            nice_to_haves: input.nice_to_haves
            resume: item
          }
          tools none
          expect CandidateScore
          timeout 3m
          retry { attempts: 1 on: [transient] }
        }
      }

      require unique(scored[*].candidate_id)
        else fail "each candidate must be screened exactly once"

      return scored
    }

    stage rank(score) -> RankedCandidate[] {
      let ranked = agent(screener) {
        task """
        Rank the screened candidates. Only candidates with meets_must_haves
        true enter the ranking, rank 1 is the strongest. recommendation is
        one of interview, hold, reject with one clause of justification.
        """
        input {
          requisition: input.requisition
          scores: score
        }
        tools none
        expect RankedCandidate[]
        timeout 4m
      }

      require ranked.count <= score.count
        else fail "ranking cannot exceed screened candidates"

      return ranked
    }

    stage gate(score, rank) -> ScreeningResult {
      let quality = verify {
        parallel {
          check fairness = agent(screener) {
            task """
            Spot-check the scoring fairness: for the top three ranked
            candidates confirm each match_score is backed by quoted resume
            evidence. ok true when no score is unjustified, flags lists the
            candidate ids in doubt.
            """
            input {
              scores: score
              ranking: rank
            }
            tools none
            expect FairnessCheck
            timeout 4m
          }
          check coverage = agent(screener) {
            task """
            Confirm completeness: every candidate with meets_must_haves true
            appears exactly once in the ranking. ok true when complete,
            missing lists the candidate ids absent from the ranking.
            """
            input {
              scores: score
              ranking: rank
            }
            tools none
            expect CoverageCheck
            timeout 3m
          }
        }
        pass when fairness.ok and coverage.ok
      }

      require quality.passed
        else fail "screening gate failed: fairness or coverage not satisfied"

      let final = agent(screener) {
        task """
        Assemble the final screening pack for the hiring manager: the scores,
        the ranking and interview_slots proposing time slots for the top
        candidates following up to five ranked interviews.
        """
        input {
          requisition: input.requisition
          scores: score
          ranking: rank
          flags: quality.checks.fairness.flags
        }
        tools none
        expect ScreeningResult
        timeout 3m
      }

      emit progress {
        requisition: input.requisition
        screened: score.count
        ranked: rank.count
      }

      return final
    }

    return gate
  }
}

performance-review.flow — 绩效反馈汇总

有界数组 [2..6]、union 聚合

查看完整代码

// 场景: 汇总 2~6 位反馈人的绩效反馈，结构化后交叉核验结论来源与措辞平衡，产出可交付的绩效综述。
workflow consolidate_review(input: ReviewPacket) -> ReviewSummary {
  use team "people-partners" as pp

  type FeedbackSource {
    kind: text
    author: text
    text: text
  }

  type ReviewPacket {
    subject: text
    role: text
    cycle: text
    sources: FeedbackSource[2..6]
  }

  type SourceSummary {
    kind: text
    strengths: text[]
    growth: text[]
    sentiment: text
  }

  type AuditFindings {
    findings: text[]
  }

  type ReviewSummary {
    strengths: text[]
    growth_areas: text[]
    narrative: text
    calibration_note: text
  }

  pipeline consolidate {
    stage summaries -> SourceSummary[] {
      let structured = parallel map input.sources as item limit 3 {
        agent(pp.main) {
          task """
          Structure one piece of performance feedback about the subject.
          Extract concrete strengths and growth areas as short, source-faithful points.
          Label the overall sentiment as positive, mixed or critical.
          Do not invent observations that the feedback text does not support.
          """
          input {
            subject: input.subject
            role: input.role
            source: item
          }
          tools none
          expect SourceSummary
          timeout 4m
        }
      }

      // 输入的有界数组 [2..6] 已由入参 Schema 强制，这里改为守卫 agent 产出的结构化质量。
      require not union(structured[*].strengths).empty
        else fail "feedback sources must surface at least one strength"
      require not union(structured[*].growth).empty
        else fail "feedback sources must surface at least one growth area"

      return structured
    }

    stage synthesis(summaries) -> ReviewSummary {
      let strength_pool = union(summaries[*].strengths)
      let growth_pool = union(summaries[*].growth)
      let draft = agent(pp.main) {
        task """
        Write the first draft of the performance review for the subject.
        Consolidate the structured feedback into shared strengths, prioritized growth
        areas, a balanced narrative and a calibration note for the manager conversation.
        Ground every statement in the feedback; surface disagreements between sources
        in the calibration note instead of averaging them away.
        """
        input {
          subject: input.subject
          role: input.role
          cycle: input.cycle
          strength_pool
          growth_pool
        }
        tools none
        expect ReviewSummary
        timeout 6m
      }

      require draft.strengths.any and draft.growth_areas.any
        else fail "draft must name at least one strength and one growth area"

      emit progress {
        phase: "synthesis"
        strengths: draft.strengths.count
        growth_areas: draft.growth_areas.count
      }
      return draft
    }

    stage verification(summaries, synthesis) -> Verification {
      let result = verify {
        parallel {
          check evidence = agent(pp.main) {
            task """
            Audit the draft review against the structured feedback.
            Confirm every conclusion in strengths, growth areas and narrative is
            supported by at least one source summary. Record each unsupported or
            overstated conclusion as a finding; return an empty findings array
            when all conclusions are grounded.
            """
            input {
              draft: synthesis
              summaries
            }
            tools none
            expect AuditFindings
            timeout 5m
          }
          check tone = agent(pp.main) {
            task """
            Review the wording of the draft for professional balance.
            Flag judgmental labels, unexplained jargon, biased phrasing and any wording
            that would not hold up in an HR policy review. Return an empty findings
            array when the tone is professional and balanced.
            """
            input {
              draft: synthesis
              cycle: input.cycle
            }
            tools none
            expect AuditFindings
            timeout 5m
          }
        }
        pass when
          evidence.findings.empty and tone.findings.empty
      }
      return result
    }

    require verification.passed
      else fail "draft review failed evidence grounding or tone checks"
    return synthesis
  }
}

project-kickoff.flow — 项目启动拆解执行

limits、disjoint、team.member(item.owner)、write item.writes

查看完整代码

// 场景: 项目启动工作包拆解与分配——把项目目标拆成带负责人与写入范围的工作包, 并行执行后逐包核验完成证据。
workflow project_kickoff(input: KickoffRequest) -> KickoffResult {
  use team "delivery-team" as crew

  limits {
    concurrency: 2
    agent_runs: 24
    duration: 45m
  }

  type KickoffRequest {
    project: text
    goal: text
    constraints: text[]
    budget_note: text
  }

  type WorkItem {
    id: text
    title: text
    owner: member<crew>
    writes: path[]
    done_criteria: text
  }

  type Evidence {
    item_id: text
    verified: path[]
    gaps: text[]
  }

  type KickoffRecord {
    summary: text
    verification_note: text
  }

  type KickoffResult {
    summary: text
    work: WorkItem[]
    execution: text[]
    verification_note: text
  }

  pipeline kickoff_delivery {

    stage plan -> WorkItem[] {
      let work = agent(crew.main) {
        task """
        Turn the kickoff request into a delivery-ready work breakdown. Split
        the goal into 1 to 8 concrete work items that respect the constraints
        and the budget note, assign each item to the crew member best suited
        to own it, and give every item a writes scope no other item overlaps.
        """
        input {
          project: input.project
          goal: input.goal
          constraints: input.constraints
          budget_note: input.budget_note
        }
        tools [read_file, list_dir, grep]
        expect WorkItem[1..8]
        timeout 5m
      }

      require work.count in 1..8 else fail "work breakdown must contain 1 to 8 items"
      require unique(work[*].id) else fail "work item ids must be unique"
      require disjoint(work[*].writes) else fail "work items must not overlap in write scope"

      emit progress { phase: "plan", work_items: work.count }

      return work
    }

    stage run(plan) -> text[] {
      return parallel map plan as item limit 2 {
        agent(crew.member(item.owner)) {
          task """
          Execute the assigned work item until its done criteria hold. Create
          or change files only under the writes paths declared for this item,
          and respect the project constraints. Return a short execution report
          covering what you set up, which files you touched and anything open.
          """
          input { goal: input.goal, constraints: input.constraints, work: item }
          tools [read_file, list_dir, grep, write_file, edit_file, apply_patch]
          write item.writes
          expect text
          timeout 15m
          retry { attempts: 1 on: [timeout, rate_limit] }
        }
      }
    }

    stage signoff(plan, run) -> KickoffRecord {
      let outcome = verify {
        check evidence = parallel map plan as item limit 2 {
          agent(crew.member(item.owner)) {
            task """
            Inspect the workspace and judge the work item against its done
            criteria. List the artifacts you actually inspected, and record
            one gap entry for each criterion that is not yet met. Do not
            modify any file.
            """
            input { work: item, execution: run }
            tools [read_file, list_dir, grep]
            expect Evidence
            timeout 5m
          }
        }

        pass when union(evidence[*].gaps).empty
      }

      require outcome.passed else fail "not every work package met its done criteria"

      return agent(crew.main) {
        task """
        Compose the kickoff record from the execution reports and evidence.
        The summary states what each work package delivered; the verification
        note states which packages are confirmed and by which evidence. Do
        not hide gaps and do not invent evidence.
        """
        input { work: plan, execution: run, evidence: outcome.checks.evidence }
        tools none
        expect KickoffRecord
        timeout 5m
      }
    }

    return {
      summary: signoff.summary
      work: plan
      execution: run
      verification_note: signoff.verification_note
    }
  }
}

工程与数据

bug-triage.flow — 缺陷分诊与派单

use team + use agent 混用、tools 只读集

查看完整代码

// 场景: 缺陷报告自动分诊——代码级复现定位、严重度定级, 并按建议负责人派发工卡。

workflow bug_triage(input: BugReport) -> TriageResult {
  use team "dev-team" as dev
  use agent "repo-investigator" as investigator

  limits {
    concurrency: 2
    agent_runs: 6
    duration: 20m
  }

  type BugReport {
    id: text
    component: text
    summary: text
    steps: text[]
    reporter: text
    logs: text
  }

  type Reproduction {
    reproduced: bool
    suspect_files: path[]
    hypothesis: text
  }

  type Assessment {
    severity: number
    priority: text
    suggested_owner: member<dev>
  }

  type Assignment {
    ticket: text
  }

  type TriageResult {
    severity: number
    reproduced: bool
    hypothesis: text
    assignee: text
    ticket: text
  }

  pipeline triage {

    stage reproduce -> Reproduction {
      let result = agent(investigator) {
        task """
        Investigate the reported bug against the live codebase.

        1. Read the report, the reproduce steps and the attached logs.
        2. Locate the component and trace the failure path.
        3. Decide whether the bug reproduces; name the suspect files
           and state a root-cause hypothesis.

        Read-only investigation: do not modify any files.
        """

        input { report: input }
        tools [read_file, grep, list_dir]

        expect Reproduction
        timeout 10m
      }

      require not result.suspect_files.empty
        else fail "triage must name at least one suspect file"

      emit progress {
        phase: "reproduce"
        reproduced: result.reproduced
        suspects: result.suspect_files.count
      }

      return result
    }

    stage assess(reproduce) -> Assessment {
      return agent(dev.main) {
        task """
        Act as the dev team lead and grade the triaged bug.

        1. Rate severity on a 1..5 scale: 5 critical outage or data loss,
           4 major feature broken, 3 partial malfunction, 2 minor, 1 cosmetic.
        2. Pick one priority label: now, today, this_week or backlog.
        3. Nominate the dev-team member best suited to fix it,
           judging from the suspect files.

        Grade from the reproduction evidence only; do not re-investigate.
        """

        input { report: input, evidence: reproduce }
        tools [read_file, grep]

        expect Assessment
        timeout 5m
      }
    }

    require assess.severity in 1..5
      else fail "severity must stay on the 1..5 scale"

    require reproduce.reproduced == false or assess.severity >= 2
      else fail "a reproduced bug cannot be graded as cosmetic"

    stage assign(reproduce, assess) -> Assignment {
      return agent(dev.member(assess.suggested_owner)) {
        task """
        You are the suggested owner for this bug.

        Skim the report, the reproduction evidence and the grading,
        then draft the dispatch ticket you will work from: symptom,
        evidence, suspect files, and the first concrete fix step.

        Do not modify any files.
        """

        input { report: input, evidence: reproduce, grading: assess }
        tools [read_file, grep]

        expect Assignment
        timeout 5m
      }
    }

    return {
      severity: assess.severity
      reproduced: reproduce.reproduced
      hypothesis: reproduce.hypothesis
      assignee: assess.suggested_owner
      ticket: assign.ticket
    }
  }
}

data-report.flow — 数据表质量审计

三段 stage 依赖链、require unique

查看完整代码

// 场景: 数据表自动化质量审计(列名唯一性、行数核对、质量问题),并产出带图表规格的经营分析报告
workflow audit_table(input: TableAudit) -> DataReport {
  use agent "data-analyst" as analyst

  type Column {
    name: text
    kind: text
  }

  type TableAudit {
    table_name: text
    columns: Column[1..40]
    rows: text[]
  }

  type DataProfile {
    row_count: number
    issues: text[]
    suggestions: text[]
  }

  type ChartSpec {
    chart_type: text
    x_axis: text
    y_axis: text
    rationale: text
  }

  type DataReport {
    table: text
    profile: DataProfile
    charts: ChartSpec[]
    summary: text
  }

  limits {
    concurrency: 2
    agent_runs: 6
    duration: 20m
  }

  pipeline audit {
    stage profile -> DataProfile {
      require unique(input.columns[*].name)
        else fail "column names must be unique before profiling"

      let result = agent(analyst) {
        task """
        Profile the data table described by its columns and sampled rows.
        row_count is your best estimate from the sample metadata. issues
        lists data quality problems you can actually observe (missing values,
        mixed formats, outliers). suggestions lists concrete fixes.
        """
        input {
          table: input
        }
        tools none
        expect DataProfile
        timeout 5m
      }

      require result.row_count >= input.rows.count
        else fail "profiled row count cannot be lower than the sampled rows"

      emit progress {
        table: input.table_name
        rows: result.row_count
        issues: result.issues.count
      }

      return result
    }

    stage charts(profile) -> ChartSpec[] {
      let specs = agent(analyst) {
        task """
        Propose charts for the executive report based on the table shape and
        the profiling results. One to three charts, each with the chart type,
        the x and y axis column names and why it matters for the business.
        """
        input {
          table: input
          profile: profile
        }
        tools none
        expect ChartSpec[]
        timeout 4m
      }

      require specs.count in 1..3
        else fail "the report needs one to three charts"

      return specs
    }

    stage report(profile, charts) -> text {
      let narrative = agent(analyst) {
        task """
        Write the executive summary of the data audit: what the table holds,
        how healthy the data is, what the charts will show and the first data
        quality fix to prioritize. Three short paragraphs for management.
        """
        input {
          table: input
          profile: profile
          charts: charts
        }
        tools none
        expect text
        timeout 4m
      }

      return narrative
    }

    return {
      table: input.table_name
      profile: profile
      charts: charts
      summary: report
    }
  }
}

competitive-analysis.flow — 竞品调研矩阵

tools [web_search, read_url]、retry on [rate_limit, transient]

查看完整代码

// 场景: 对每个竞品并行联网调研，汇总生成竞品对比矩阵与竞争策略建议。
workflow competitive_analysis(input: ResearchBrief) -> AnalysisReport {
  use agent "web-researcher" as researcher

  limits {
    concurrency: 3
    agent_runs: 16
    duration: 15m
  }

  type Competitor {
    name: text
    site: text
  }

  type ResearchBrief {
    our_product: text
    market: text
    competitors: Competitor[1..6]
  }

  type CompetitorProfile {
    name: text
    pricing: text
    strengths: text[]
    weaknesses: text[]
    differentiators: text[]
  }

  type MatrixReport {
    matrix: text
    threats: text[]
    recommendations: text[]
  }

  type AnalysisReport {
    profiles: CompetitorProfile[]
    matrix: text
    threats: text[]
    recommendations: text[]
  }

  pipeline analyze {

    stage research -> CompetitorProfile[] {
      require input.competitors.any
        else fail "at least one competitor is required"

      require unique(input.competitors[*].name)
        else fail "competitor names must be unique"

      emit progress {
        phase: "research"
        message: "profiling competitors"
        competitors: input.competitors[*].name
      }

      return parallel map input.competitors as item limit 3 {
        agent(researcher) {
          task """
          Research one competitor of our product using live web sources.

          For the given competitor collect:
          - current pricing model and entry price point
          - 2 to 4 concrete strengths, each backed by a source
          - 2 to 4 concrete weaknesses or product gaps
          - the differentiators that position it against our product

          Prefer the competitor's official site and pricing page. When a claim
          cannot be verified, say so explicitly instead of guessing.
          """

          input {
            our_product: input.our_product
            market: input.market
            competitor: item
          }

          tools [web_search, read_url]

          expect CompetitorProfile
          timeout 3m

          retry {
            attempts: 1
            on: [rate_limit, transient]
          }
        }
      }
    }


    stage matrix(research) -> MatrixReport {
      require research.count == input.competitors.count
        else fail "every competitor must have a completed profile"

      require unique(research[*].name)
        else fail "profile names must stay unique"

      return agent(researcher) {
        task """
        Turn the competitor profiles into a decision-ready comparison for our product.

        Produce:
        - a markdown comparison matrix: one row per competitor, columns covering
          pricing, strengths, weaknesses and differentiators versus our product
        - the top competitive threats to our product in this market
        - actionable recommendations on positioning, pricing and roadmap moves

        Base every statement on the given profiles only; do not invent new facts.
        """

        input {
          our_product: input.our_product
          market: input.market
          profiles: research
        }

        tools none

        expect MatrixReport
        timeout 5m
      }
    }


    emit progress {
      phase: "report"
      message: "comparison matrix ready"
    }

    return {
      profiles: research
      matrix: matrix.matrix
      threats: matrix.threats
      recommendations: matrix.recommendations
    }
  }
}

风格约定

• 每个文件首行为单行中文注释说明场景;代码、标识符与 task 用英文
• 非确定性产出一律走 agent(...);确定性控制流用语言结构表达
• 没有任何 agent 被允许返回“是否通过”——验证结论只能来自 pass when
• tools 与 write 是请求能力,最终以宿主策略收窄为准