声明性工作流 - 概述

声明性工作流允许使用 YAML 配置文件定义工作流逻辑，而不是编写编程代码。此方法使工作流更易于跨团队读取、修改和共享。

概述

使用声明性工作流，描述工作流需要做什么，而不是如何实现它。框架处理基础执行，将 YAML 定义转换为可执行的工作流图。

主要优势：

可读格式：YAML 语法易于理解，即使对于非开发人员也是如此
可移植：无需更改代码即可共享、版本控制和修改工作流定义
快速迭代：通过编辑配置文件更改工作流行为
一致的结构：预定义的作类型确保工作流遵循最佳做法

何时使用声明性工作流与编程工作流

Scenario	建议的方法
标准编排模式	声明性
经常更改的工作流	声明性
非开发人员需要修改工作流	声明性
复杂的自定义逻辑	Programmatic
最大灵活性和控制	Programmatic
与现有 Python 代码集成	Programmatic

基本 YAML 结构

YAML 结构在 C# 和 Python 实现之间略有不同。有关详细信息，请参阅下面的特定于语言的部分。

操作类型

声明性工作流支持各种操作类型，包括变量管理、控制流、代理和工具调用、HTTP 和 MCP 集成、人工循环和聊天控制。完整的特定于语言的引用将显示在下面的每个区域中;有关这两种语言的概览可用性矩阵，请参阅本文底部的 “快速参考表 ”。

C# YAML 结构

C# 声明性工作流使用基于触发器的结构：

#
# Workflow description as a comment
#
kind: Workflow
trigger:

  kind: OnConversationStart
  id: my_workflow
  actions:

    - kind: ActionType
      id: unique_action_id
      displayName: Human readable name
      # Action-specific properties

结构元素

元素	必选	Description
`kind`	是的	必须是 `Workflow`
`trigger.kind`	是的	触发器类型（通常 `OnConversationStart`）
`trigger.id`	是的	工作流的唯一标识符
`trigger.actions`	是的	要执行的操作列表

Python YAML 结构

Python 声明性工作流使用具有可选输入的基于名称的结构：

name: my-workflow
description: A brief description of what this workflow does

inputs:
  parameterName:
    type: string
    description: Description of the parameter

actions:
  - kind: ActionType
    id: unique_action_id
    displayName: Human readable name
    # Action-specific properties

结构元素

元素	必选	Description
`name`	是的	工作流的唯一标识符
`description`	否	人工可读说明
`inputs`	否	工作流接受的输入参数
`actions`	是的	要执行的操作列表

先决条件

在开始之前，请确保具备：

.NET 8.0 或更高版本
具有至少一个已部署代理的 Microsoft Foundry 项目
已安装以下 NuGet 包：

dotnet add package Microsoft.Agents.AI.Workflows.Declarative --prerelease
dotnet add package Microsoft.Agents.AI.Workflows.Declarative.AzureAI --prerelease

如果您打算将 MCP 工具调用动作添加到工作流中，请安装以下 NuGet 包：

dotnet add package Microsoft.Agents.AI.Workflows.Declarative.Mcp --prerelease

基本熟悉 YAML 语法
了解工作流概念

你的第一个声明性工作流

让我们创建一个简单的工作流，根据用户的输入来问候用户。

步骤 1：创建 YAML 文件

创建名为 greeting-workflow.yaml：的文件

#
# This workflow demonstrates a simple greeting based on user input.
# The user's message is captured via System.LastMessage.
#
# Example input: 
# Alice
#
kind: Workflow
trigger:

  kind: OnConversationStart
  id: greeting_workflow
  actions:

    # Capture the user's input from the last message
    - kind: SetVariable
      id: capture_name
      displayName: Capture user name
      variable: Local.userName
      value: =System.LastMessage.Text

    # Set a greeting prefix
    - kind: SetVariable
      id: set_greeting
      displayName: Set greeting prefix
      variable: Local.greeting
      value: Hello

    # Build the full message using an expression
    - kind: SetVariable
      id: build_message
      displayName: Build greeting message
      variable: Local.message
      value: =Concat(Local.greeting, ", ", Local.userName, "!")

    # Send the greeting to the user
    - kind: SendActivity
      id: send_greeting
      displayName: Send greeting to user
      activity: =Local.message

步骤 2：配置代理提供程序

创建 C# 控制台应用程序以执行工作流。首先，配置连接到 Foundry 的代理提供程序：

using Azure.Identity;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Agents.AI.Workflows.Declarative;
using Microsoft.Extensions.Configuration;

// Load configuration (endpoint should be set in user secrets or environment variables)
IConfiguration configuration = new ConfigurationBuilder()
    .AddUserSecrets<Program>()
    .AddEnvironmentVariables()
    .Build();

string foundryEndpoint = configuration["FOUNDRY_PROJECT_ENDPOINT"] 
    ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT not configured");

// Create the agent provider that connects to Foundry
// WARNING: DefaultAzureCredential is convenient for development but requires 
// careful consideration in production environments.
AzureAgentProvider agentProvider = new(
    new Uri(foundryEndpoint), 
    new DefaultAzureCredential());

步骤 3：生成并运行工作流

// Define workflow options with the agent provider
DeclarativeWorkflowOptions options = new(agentProvider)
{
    Configuration = configuration,
    // LoggerFactory = loggerFactory, // Optional: Enable logging
    // ConversationId = conversationId, // Optional: Continue existing conversation
};

// Build the workflow from the YAML file
string workflowPath = Path.Combine(AppContext.BaseDirectory, "greeting-workflow.yaml");
Workflow workflow = DeclarativeWorkflowBuilder.Build<string>(workflowPath, options);

Console.WriteLine($"Loaded workflow from: {workflowPath}");
Console.WriteLine(new string('-', 40));

// Create a checkpoint manager (in-memory for this example)
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();

// Execute the workflow with input
string input = "Alice";
StreamingRun run = await InProcessExecution.RunStreamingAsync(
    workflow, 
    input, 
    checkpointManager);

// Process workflow events
await foreach (WorkflowEvent workflowEvent in run.WatchStreamAsync())
{
    switch (workflowEvent)
    {
        case MessageActivityEvent activityEvent:
            Console.WriteLine($"Activity: {activityEvent.Message}");
            break;
        case AgentResponseEvent responseEvent:
            Console.WriteLine($"Response: {responseEvent.Response.Text}");
            break;
        case WorkflowErrorEvent errorEvent:
            Console.WriteLine($"Error: {errorEvent.Data}");
            break;
    }
}

Console.WriteLine("Workflow completed!");

预期输出

Loaded workflow from: C:\path\to\greeting-workflow.yaml
----------------------------------------
Activity: Hello, Alice!
Workflow completed!

核心概念

变量命名空间

C# 中的声明性工作流使用命名空间变量来组织状态：

Namespace	Description	Example
`Local.*`	工作流本地变量	`Local.message`
`System.*`	系统提供的值	`System.ConversationId`、`System.LastMessage`

注释

C# 声明性工作流不使用 Workflow.Inputs 或 Workflow.Outputs 命名空间。通过 System.LastMessage 接收输入，并通过SendActivity发送输出。

系统变量

Variable	Description
`System.ConversationId`	当前聊天标识符
`System.LastMessage`	最新用户消息
`System.LastMessage.Text`	最后一条消息的文本内容

表达式语言

前缀 = 的值使用 PowerFx 表达式语言计算为表达式：

# Literal value (no evaluation)
value: Hello

# Expression (evaluated at runtime)
value: =Concat("Hello, ", Local.userName)

# Access last message text
value: =System.LastMessage.Text

常见函数包括：

Concat(str1, str2, ...) - 连接字符串
If(condition, trueValue, falseValue) - 条件表达式
IsBlank(value) - 检查值是否为空
Upper(text) / Lower(text) - 大小写转换
Find(searchText, withinText) - 在字符串中查找文本
MessageText(message) - 从消息对象中提取文本
UserMessage(text) - 从文本创建用户消息
AgentMessage(text) - 从文本创建代理消息

配置选项

该 DeclarativeWorkflowOptions 类为工作流执行提供配置：

DeclarativeWorkflowOptions options = new(agentProvider)
{
    // Application configuration for variable substitution
    Configuration = configuration,

    // Continue an existing conversation (optional)
    ConversationId = "existing-conversation-id",

    // Enable logging (optional)
    LoggerFactory = loggerFactory,

    // MCP tool handler for InvokeMcpTool actions (optional)
    McpToolHandler = mcpToolHandler,

    // HTTP request handler for HttpRequestAction actions (optional)
    HttpRequestHandler = new DefaultHttpRequestHandler(),

    // PowerFx expression limits (optional)
    MaximumCallDepth = 50,
    MaximumExpressionLength = 10000,

    // Telemetry configuration (optional)
    ConfigureTelemetry = opts => { /* configure telemetry */ },
    TelemetryActivitySource = activitySource,
};

代理提供程序设置

将 AzureAgentProvider 连接至 Foundry 代理，整合你的工作流程：

using Azure.Identity;
using Microsoft.Agents.AI.Workflows.Declarative;

// Create the agent provider with Azure credentials
AzureAgentProvider agentProvider = new(
    new Uri("https://your-project.api.azureml.ms"), 
    new DefaultAzureCredential())
{
    // Optional: Define functions that agents can automatically invoke
    Functions = [
        AIFunctionFactory.Create(myPlugin.GetData),
        AIFunctionFactory.Create(myPlugin.ProcessItem),
    ],

    // Optional: Allow concurrent function invocation
    AllowConcurrentInvocation = true,

    // Optional: Allow multiple tool calls per response
    AllowMultipleToolCalls = true,
};

工作流执行

用于 InProcessExecution 运行工作流并处理事件：

using Microsoft.Agents.AI.Workflows;
using Microsoft.Agents.AI.Workflows.Checkpointing;

// Create checkpoint manager (choose in-memory or file-based)
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
// Or persist to disk:
// var checkpointFolder = Directory.CreateDirectory("./checkpoints");
// var checkpointManager = CheckpointManager.CreateJson(
//     new FileSystemJsonCheckpointStore(checkpointFolder));

// Start workflow execution
StreamingRun run = await InProcessExecution.RunStreamingAsync(
    workflow, 
    input, 
    checkpointManager);

// Process events as they occur
await foreach (WorkflowEvent workflowEvent in run.WatchStreamAsync())
{
    switch (workflowEvent)
    {
        case MessageActivityEvent activity:
            Console.WriteLine($"Message: {activity.Message}");
            break;

        case AgentResponseUpdateEvent streamEvent:
            Console.Write(streamEvent.Update.Text); // Streaming text
            break;

        case AgentResponseEvent response:
            Console.WriteLine($"Agent: {response.Response.Text}");
            break;

        case RequestInfoEvent request:
            // Handle external input requests (human-in-the-loop)
            var userInput = await GetUserInputAsync(request);
            await run.SendResponseAsync(request.Request.CreateResponse(userInput));
            break;

        case SuperStepCompletedEvent checkpoint:
            // Checkpoint created - can resume from here if needed
            var checkpointInfo = checkpoint.CompletionInfo?.Checkpoint;
            break;

        case WorkflowErrorEvent error:
            Console.WriteLine($"Error: {error.Data}");
            break;
    }
}

从检查点恢复

可以从检查点恢复工作流，以实现容错。

// Save checkpoint info when workflow yields
CheckpointInfo? lastCheckpoint = null;

await foreach (WorkflowEvent workflowEvent in run.WatchStreamAsync())
{
    if (workflowEvent is SuperStepCompletedEvent checkpointEvent)
    {
        lastCheckpoint = checkpointEvent.CompletionInfo?.Checkpoint;
    }
}

// Later: Resume from the saved checkpoint
if (lastCheckpoint is not null)
{
    // Recreate the workflow (can be on a different machine)
    Workflow workflow = DeclarativeWorkflowBuilder.Build<string>(workflowPath, options);

    StreamingRun resumedRun = await InProcessExecution.ResumeStreamingAsync(
        workflow, 
        lastCheckpoint, 
        checkpointManager);

    // Continue processing events...
}

操作参考

动作是声明性工作流的基础模块。每个动作执行特定操作，并且动作按照 YAML 文件中显示的顺序执行。

动作结构

所有操作具备共同属性：

- kind: ActionType      # Required: The type of action
  id: unique_id         # Optional: Unique identifier for referencing
  displayName: Name     # Optional: Human-readable name for logging
  # Action-specific properties...

变量管理操作

设置变量

将变量设置为指定的值。

- kind: SetVariable
  id: set_greeting
  displayName: Set greeting message
  variable: Local.greeting
  value: Hello World

使用表达式：

- kind: SetVariable
  variable: Local.fullName
  value: =Concat(Local.firstName, " ", Local.lastName)

性能：

资产	必选	Description
`variable`	是的	变量路径（例如， `Local.name`， `Workflow.Outputs.result`）
`value`	是的	要设置的值（文本或表达式）

设置多个变量

在单个操作中设置多个变量。

- kind: SetMultipleVariables
  id: initialize_vars
  displayName: Initialize variables
  variables:
    Local.counter: 0
    Local.status: pending
    Local.message: =Concat("Processing order ", Local.orderId)

性能：

资产	必选	Description
`variables`	是的	将变量路径映射到值

SetTextVariable （仅限 C#）

将文本变量设置为指定的字符串值。

- kind: SetTextVariable
  id: set_text
  displayName: Set text content
  variable: Local.description
  value: This is a text description

性能：

资产	必选	Description
`variable`	是的	文本值的变量路径
`value`	是的	要设置的文本值

重置变量

清除变量的值。

- kind: ResetVariable
  id: clear_counter
  variable: Local.counter

性能：

资产	必选	Description
`variable`	是的	要重置的变量路径

ClearAllVariables （仅限 C#）

重置当前上下文中的所有变量。

- kind: ClearAllVariables
  id: clear_all
  displayName: Clear all workflow variables

ParseValue （仅限 C#）

将数据提取或转换为可用格式。

- kind: ParseValue
  id: parse_json
  displayName: Parse JSON response
  source: =Local.rawResponse
  variable: Local.parsedData

性能：

资产	必选	Description
`source`	是的	返回要分析的值的表达式
`variable`	是的	用于存储已分析结果的变量路径

EditTableV2 （仅限 C#）

修改结构化表格式的数据。

- kind: EditTableV2
  id: update_table
  displayName: Update configuration table
  table: Local.configTable
  operation: update
  row:
    key: =Local.settingName
    value: =Local.settingValue

性能：

资产	必选	Description
`table`	是的	表格的可变路径
`operation`	是的	作类型（添加、更新、删除）
`row`	是的	操作的行数据

控制流操作

If

根据条件执行操作。

- kind: If
  id: check_age
  displayName: Check user age
  condition: =Local.age >= 18
  then:
    - kind: SendActivity
      activity:
        text: "Welcome, adult user!"
  else:
    - kind: SendActivity
      activity:
        text: "Welcome, young user!"

性能：

资产	必选	Description
`condition`	是的	计算结果为 true/false 的表达式
`then`	是的	如果条件为真，则执行的动作
`else`	否	如果条件为假，则执行的操作

ConditionGroup

计算多个条件，如 switch/case 语句。

- kind: ConditionGroup
  id: route_by_category
  displayName: Route based on category
  conditions:
    - condition: =Local.category = "electronics"
      id: electronics_branch
      actions:
        - kind: SetVariable
          variable: Local.department
          value: Electronics Team
    - condition: =Local.category = "clothing"
      id: clothing_branch
      actions:
        - kind: SetVariable
          variable: Local.department
          value: Clothing Team
  elseActions:
    - kind: SetVariable
      variable: Local.department
      value: General Support

性能：

资产	必选	Description
`conditions`	是的	条件/作对列表（第一场比赛获胜）
`elseActions`	否	如果没有条件匹配时的操作步骤

Foreach

循环访问集合。

- kind: Foreach
  id: process_items
  displayName: Process each item
  source: =Local.items
  itemName: item
  indexName: index
  actions:
    - kind: SendActivity
      activity:
        text: =Concat("Processing item ", index, ": ", item)

性能：

资产	必选	Description
`source`	是的	返回集合的表达式
`itemName`	否	当前项的变量名称（默认值： `item`）
`indexName`	否	当前索引的变量名称（默认值： `index`）
`actions`	是的	要为每个项目执行的操作

BreakLoop

立即退出当前循环。

- kind: Foreach
  source: =Local.items
  actions:
    - kind: If
      condition: =item = "stop"
      then:
        - kind: BreakLoop
    - kind: SendActivity
      activity:
        text: =item

ContinueLoop

跳过循环的下一次迭代。

- kind: Foreach
  source: =Local.numbers
  actions:
    - kind: If
      condition: =item < 0
      then:
        - kind: ContinueLoop
    - kind: SendActivity
      activity:
        text: =Concat("Positive number: ", item)

GotoAction

按 ID 跳转到特定操作。

- kind: SetVariable
  id: start_label
  variable: Local.attempts
  value: =Local.attempts + 1

- kind: SendActivity
  activity:
    text: =Concat("Attempt ", Local.attempts)

- kind: If
  condition: =And(Local.attempts < 3, Not(Local.success))
  then:
    - kind: GotoAction
      actionId: start_label

性能：

资产	必选	Description
`actionId`	是的	要跳转到的动作 ID

输出动作

SendActivity

向用户发送消息。

- kind: SendActivity
  id: send_welcome
  displayName: Send welcome message
  activity:
    text: "Welcome to our service!"

使用表达式：

- kind: SendActivity
  activity:
    text: =Concat("Hello, ", Local.userName, "! How can I help you today?")

性能：

资产	必选	Description
`activity`	是的	要发送的活动
`activity.text`	是的	消息文本（文本或表达式）

代理调用操作

InvokeAzureAgent

调用 Foundry 代理。

基本调用：

- kind: InvokeAzureAgent
  id: call_assistant
  displayName: Call assistant agent
  agent:
    name: AssistantAgent
  conversationId: =System.ConversationId

通过输入和输出配置：

- kind: InvokeAzureAgent
  id: call_analyst
  displayName: Call analyst agent
  agent:
    name: AnalystAgent
  conversationId: =System.ConversationId
  input:
    messages: =Local.userMessage
    arguments:
      topic: =Local.topic
  output:
    responseObject: Local.AnalystResult
    messages: Local.AnalystMessages
    autoSend: true

使用外部循环（持续运行直到满足条件）：

- kind: InvokeAzureAgent
  id: support_agent
  agent:
    name: SupportAgent
  input:
    externalLoop:
      when: =Not(Local.IsResolved)
  output:
    responseObject: Local.SupportResult

性能：

资产	必选	Description
`agent.name`	是的	已注册代理的名称
`conversationId`	否	对话上下文标识符
`input.messages`	否	发送给代理的消息
`input.arguments`	否	代理的其他参数
`input.externalLoop.when`	否	继续代理循环的条件
`output.responseObject`	否	存储代理响应的路径
`output.messages`	否	存储对话消息的路径
`output.autoSend`	否	自动向用户发送响应

工具和 HTTP 操作

InvokeFunctionTool

直接从工作流调用函数工具，而无需通过 AI 代理。

- kind: InvokeFunctionTool
  id: invoke_get_data
  displayName: Get data from function
  functionName: GetUserData
  conversationId: =System.ConversationId
  requireApproval: true
  arguments:
    userId: =Local.userId
  output:
    autoSend: true
    result: Local.UserData
    messages: Local.FunctionMessages

性能：

资产	必选	Description
`functionName`	是的	要调用的函数的名称
`conversationId`	否	对话上下文标识符
`requireApproval`	否	是否在执行前要求用户批准
`arguments`	否	要传递给函数的参数
`output.result`	否	存储函数结果的路径
`output.messages`	否	存储函数消息的路径
`output.autoSend`	否	自动向用户发送结果

InvokeFunctionTool 的 C# 设置：

函数必须注册到WorkflowRunner或通过外部输入进行处理：

// Define functions that can be invoked
AIFunction[] functions = [
    AIFunctionFactory.Create(myPlugin.GetUserData),
    AIFunctionFactory.Create(myPlugin.ProcessOrder),
];

// Create workflow runner with functions
WorkflowRunner runner = new(functions) { UseJsonCheckpoints = true };
await runner.ExecuteAsync(workflowFactory.CreateWorkflow, input);

InvokeMcpTool

在 MCP（模型上下文协议）服务器上调用工具。

- kind: InvokeMcpTool
  id: invoke_docs_search
  displayName: Search documentation
  serverUrl: https://learn.microsoft.com/api/mcp
  serverLabel: microsoft_docs
  toolName: microsoft_docs_search
  conversationId: =System.ConversationId
  requireApproval: false
  headers:
    X-Custom-Header: custom-value
  arguments:
    query: =Local.SearchQuery
  output:
    autoSend: true
    result: Local.SearchResults

使用托管方案的连接名称：

- kind: InvokeMcpTool
  id: invoke_hosted_mcp
  serverUrl: https://mcp.ai.azure.com
  toolName: my_tool
  # Connection name is used in hosted scenarios to connect to a ProjectConnectionId in Foundry.
  # Note: This feature is not fully supported yet.
  connection:
    name: my-foundry-connection
  output:
    result: Local.ToolResult

性能：

资产	必选	Description
`serverUrl`	是的	MCP 服务器的 URL
`serverLabel`	否	服务器的人类可读标签
`toolName`	是的	要调用的工具的名称
`conversationId`	否	对话上下文标识符
`requireApproval`	否	是否需要用户批准
`arguments`	否	要传递给工具的参数
`headers`	否	请求的自定义 HTTP 标头
`connection.name`	否	托管场景的命名连接（连接到 Foundry 中的 ProjectConnectionId；尚未完全支持）
`output.result`	否	存储工具结果的路径
`output.messages`	否	存储结果消息的路径
`output.autoSend`	否	自动向用户发送结果

InvokeMcpTool 的 C# 设置：

在您的工作流工厂中配置McpToolHandler：

using Azure.Core;
using Azure.Identity;
using Microsoft.Agents.AI.Workflows.Declarative;

// Create MCP tool handler with authentication callback
DefaultAzureCredential credential = new();
DefaultMcpToolHandler mcpToolHandler = new(
    httpClientProvider: async (serverUrl, cancellationToken) =>
    {
        if (serverUrl.StartsWith("https://mcp.ai.azure.com", StringComparison.OrdinalIgnoreCase))
        {
            // Acquire token for Azure MCP server
            AccessToken token = await credential.GetTokenAsync(
                new TokenRequestContext(["https://mcp.ai.azure.com/.default"]),
                cancellationToken);

            HttpClient httpClient = new();
            httpClient.DefaultRequestHeaders.Authorization =
                new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", token.Token);
            return httpClient;
        }

        // Return null for servers that don't require authentication
        return null;
    });

// Configure workflow factory with MCP handler
WorkflowFactory workflowFactory = new("workflow.yaml", foundryEndpoint)
{
    McpToolHandler = mcpToolHandler
};

HttpRequestAction

通过配置的 IHttpRequestHandler 发送 HTTP 请求。在分配之前，成功的 JSON 响应会被解析；非 2xx 的响应会导致操作失败。

- kind: HttpRequestAction
  id: fetch_repo_info
  method: GET
  url: "https://api.github.com/repos/Microsoft/agent-framework"
  headers:
    Accept: application/vnd.github+json
    User-Agent: agent-framework
  queryParameters:
    per_page: 10
  response: Local.RepoInfo
  responseHeaders: Local.RepoHeaders

性能：

资产	必选	Description
`url`	是的	绝对请求 URL
`method`	否	HTTP 方法;默认值为 `GET`
`headers`	否	请求标头
`queryParameters`	否	附加到 URL 的查询参数
`body`	否	请求正文：使用`kind: json`、`raw`或`none`
`requestTimeoutInMilliseconds`	否	每个请求的超时设置
`conversationId`	否	将响应正文成功地添加到对话中
`response`	否	存储已分析响应正文的路径
`responseHeaders`	否	存储响应标头的路径

HttpRequestAction 的 C# 设置：

在生成工作流时设置 HttpRequestHandler 。需要重试或加入 URL 白名单时，请使用自定义处理程序。

DeclarativeWorkflowOptions options = new(agentProvider)
{
    HttpRequestHandler = new DefaultHttpRequestHandler(),
};

Workflow workflow = DeclarativeWorkflowBuilder.Build<string>("workflow.yaml", options);

人机互动操作

问题

向用户提问并存储响应。

- kind: Question
  id: ask_name
  displayName: Ask for user name
  question:
    text: "What is your name?"
  variable: Local.userName
  default: "Guest"

性能：

资产	必选	Description
`question.text`	是的	要提问的问题
`variable`	是的	存储响应的路径
`default`	否	如果未响应，则默认值

RequestExternalInput

从外部系统或进程请求输入。

- kind: RequestExternalInput
  id: request_approval
  displayName: Request manager approval
  prompt:
    text: "Please provide approval for this request."
  variable: Local.approvalResult
  default: "pending"

性能：

资产	必选	Description
`prompt.text`	是的	所需输入的说明
`variable`	是的	存储输入的路径
`default`	否	默认值

工作流控制动作

EndWorkflow

终止工作流执行。

- kind: EndWorkflow
  id: finish
  displayName: End workflow

EndConversation

结束当前对话。

- kind: EndConversation
  id: end_chat
  displayName: End conversation

CreateConversation

创建新的对话上下文。

- kind: CreateConversation
  id: create_new_conv
  displayName: Create new conversation
  conversationId: Local.NewConversationId

性能：

资产	必选	Description
`conversationId`	是的	存储新会话 ID 的路径

对话操作（仅限 C#）

添加对话消息

将消息添加到会话线程。

- kind: AddConversationMessage
  id: add_system_message
  displayName: Add system context
  conversationId: =System.ConversationId
  message:
    role: system
    content: =Local.contextInfo

性能：

资产	必选	Description
`conversationId`	是的	目标会话标识符
`message`	是的	待添加的消息
`message.role`	是的	消息角色（系统、用户、助理）
`message.content`	是的	消息内容

复制对话消息

将消息从一个对话复制到另一个对话。

- kind: CopyConversationMessages
  id: copy_context
  displayName: Copy conversation context
  sourceConversationId: =Local.SourceConversation
  targetConversationId: =System.ConversationId
  limit: 10

性能：

资产	必选	Description
`sourceConversationId`	是的	源对话标识符
`targetConversationId`	是的	目标会话标识符
`limit`	否	要复制的最大消息数

获取会话消息

从对话中检索特定消息。

- kind: RetrieveConversationMessage
  id: get_message
  displayName: Get specific message
  conversationId: =System.ConversationId
  messageId: =Local.targetMessageId
  variable: Local.retrievedMessage

性能：

资产	必选	Description
`conversationId`	是的	对话标识符
`messageId`	是的	待检索的消息标识符
`variable`	是的	存储检索到的消息的路径

检索会话消息

从会话中检索多个消息。

- kind: RetrieveConversationMessages
  id: get_history
  displayName: Get conversation history
  conversationId: =System.ConversationId
  limit: 20
  newestFirst: true
  variable: Local.conversationHistory

性能：

资产	必选	Description
`conversationId`	是的	对话标识符
`limit`	否	要检索的最大消息数（默认值：20）
`newestFirst`	否	按降序返回
`after`	否	分页的游标
`before`	否	分页的游标
`variable`	是的	存储检索到的消息的路径

操作快速参考

Action	类别	C#	Python	Description
`SetVariable`	Variable	✅	✅	设置单个变量
`SetMultipleVariables`	Variable	✅	✅	设置多个变量
`SetTextVariable`	Variable	✅	❌	设置文本变量
`AppendValue`	Variable	❌	✅	追加到列表/字符串
`ResetVariable`	Variable	✅	✅	清除变量
`ClearAllVariables`	Variable	✅	❌	清除所有变量
`ParseValue`	Variable	✅	❌	分析/转换数据
`EditTableV2`	Variable	✅	❌	修改表数据
`If`	控制流	✅	✅	条件分支
`ConditionGroup`	控制流	✅	✅	多分支交换机
`Foreach`	控制流	✅	✅	循环访问集合
`RepeatUntil`	控制流	❌	✅	循环直到满足条件
`BreakLoop`	控制流	✅	✅	退出当前循环
`ContinueLoop`	控制流	✅	✅	跳至下一次迭代
`GotoAction`	控制流	✅	✅	跳转到动作按 ID
`SendActivity`	输出	✅	✅	向用户发送消息
`EmitEvent`	输出	❌	✅	发出自定义事件
`InvokeAzureAgent`	代理人	✅	✅	调用 Azure AI 代理
`InvokeFunctionTool`	工具	✅	✅	直接调用函数
`InvokeMcpTool`	工具	✅	✅	调用 MCP 服务器工具
`HttpRequestAction`	HTTP	✅	✅	调用 HTTP 终结点
`Question`	人在环	✅	✅	向用户提问
`Confirmation`	人在环	❌	✅	是/否确认
`RequestExternalInput`	人在环	✅	✅	请求外部输入
`WaitForInput`	人在环	❌	✅	等待输入
`EndWorkflow`	工作流控件	✅	✅	终止工作流
`EndConversation`	工作流控件	✅	✅	结束对话
`CreateConversation`	工作流控件	✅	✅	创建新对话
`AddConversationMessage`	对话	✅	❌	将消息添加到线程
`CopyConversationMessages`	对话	✅	❌	复制消息
`RetrieveConversationMessage`	对话	✅	❌	获取单条消息
`RetrieveConversationMessages`	对话	✅	❌	获取多个消息

高级模式

多代理编排

顺序代理管道

按顺序通过多个代理传递工作。

#
# Sequential agent pipeline for content creation
#
kind: Workflow
trigger:

  kind: OnConversationStart
  id: content_workflow
  actions:

    # First agent: Research
    - kind: InvokeAzureAgent
      id: invoke_researcher
      displayName: Research phase
      conversationId: =System.ConversationId
      agent:
        name: ResearcherAgent

    # Second agent: Write draft
    - kind: InvokeAzureAgent
      id: invoke_writer
      displayName: Writing phase
      conversationId: =System.ConversationId
      agent:
        name: WriterAgent

    # Third agent: Edit
    - kind: InvokeAzureAgent
      id: invoke_editor
      displayName: Editing phase
      conversationId: =System.ConversationId
      agent:
        name: EditorAgent

C# 设置：

using Azure.AI.Projects;
using Azure.AI.Projects.OpenAI;
using Azure.Identity;

// Ensure agents exist in Foundry
AIProjectClient aiProjectClient = new(foundryEndpoint, new DefaultAzureCredential());

await aiProjectClient.CreateAgentAsync(
    agentName: "ResearcherAgent",
    agentDefinition: new DeclarativeAgentDefinition(modelName)
    {
        Instructions = "You are a research specialist..."
    },
    agentDescription: "Research agent for content pipeline");

// Create and run workflow
WorkflowFactory workflowFactory = new("content-pipeline.yaml", foundryEndpoint);
WorkflowRunner runner = new();
await runner.ExecuteAsync(workflowFactory.CreateWorkflow, "Create content about AI");

条件代理路由

根据条件将请求路由到不同的代理。

#
# Route to specialized support agents based on category
#
kind: Workflow
trigger:

  kind: OnConversationStart
  id: support_router
  actions:

    # Capture category from user input or set via another action
    - kind: SetVariable
      id: set_category
      variable: Local.category
      value: =System.LastMessage.Text

    - kind: ConditionGroup
      id: route_request
      displayName: Route to appropriate agent
      conditions:
        - condition: =Local.category = "billing"
          id: billing_route
          actions:
            - kind: InvokeAzureAgent
              id: billing_agent
              agent:
                name: BillingAgent
              conversationId: =System.ConversationId
        - condition: =Local.category = "technical"
          id: technical_route
          actions:
            - kind: InvokeAzureAgent
              id: technical_agent
              agent:
                name: TechnicalAgent
              conversationId: =System.ConversationId
      elseActions:
        - kind: InvokeAzureAgent
          id: general_agent
          agent:
            name: GeneralAgent
          conversationId: =System.ConversationId

工具集成模式

使用 InvokeFunctionTool 预提取数据

在调用代理之前提取数据：

#
# Pre-fetch menu data before agent interaction
#
kind: Workflow
trigger:

  kind: OnConversationStart
  id: menu_workflow
  actions:
    # Pre-fetch today's specials
    - kind: InvokeFunctionTool
      id: get_specials
      functionName: GetSpecials
      requireApproval: true
      output:
        autoSend: true
        result: Local.Specials

    # Agent uses pre-fetched data
    - kind: InvokeAzureAgent
      id: menu_agent
      conversationId: =System.ConversationId
      agent:
        name: MenuAgent
      input:
        messages: =UserMessage("Describe today's specials: " & Local.Specials)

MCP 工具集成

使用 MCP 调用外部服务器：

#
# Search documentation using MCP
#
kind: Workflow
trigger:

  kind: OnConversationStart
  id: docs_search
  actions:

    - kind: SetVariable
      variable: Local.SearchQuery
      value: =System.LastMessage.Text

    # Search Microsoft Learn
    - kind: InvokeMcpTool
      id: search_docs
      serverUrl: https://learn.microsoft.com/api/mcp
      toolName: microsoft_docs_search
      conversationId: =System.ConversationId
      arguments:
        query: =Local.SearchQuery
      output:
        result: Local.SearchResults
        autoSend: true

    # Summarize results with agent
    - kind: InvokeAzureAgent
      id: summarize
      agent:
        name: SummaryAgent
      conversationId: =System.ConversationId
      input:
        messages: =UserMessage("Summarize these search results")

先决条件

在开始之前，请确保具备：

Python 3.10 - 3.13（由于 PowerFx 兼容性，尚不支持 Python 3.14）
已安装的 Agent Framework 声明性包：

pip install agent-framework-declarative --pre

此包会自动引入基础agent-framework-core。

基本熟悉 YAML 语法
了解工作流概念

你的第一个声明性工作流

让我们创建一个简单的工作流，该工作流按名称问候用户。

步骤 1：创建 YAML 文件

创建名为 greeting-workflow.yaml：的文件

name: greeting-workflow
description: A simple workflow that greets the user

inputs:
  name:
    type: string
    description: The name of the person to greet

actions:
  # Set a greeting prefix
  - kind: SetVariable
    id: set_greeting
    displayName: Set greeting prefix
    variable: Local.greeting
    value: Hello

  # Build the full message using an expression
  - kind: SetVariable
    id: build_message
    displayName: Build greeting message
    variable: Local.message
    value: =Concat(Local.greeting, ", ", Workflow.Inputs.name, "!")

  # Send the greeting to the user
  - kind: SendActivity
    id: send_greeting
    displayName: Send greeting to user
    activity:
      text: =Local.message

  # Store the result in outputs
  - kind: SetVariable
    id: set_output
    displayName: Store result in outputs
    variable: Workflow.Outputs.greeting
    value: =Local.message

步骤 2：加载并运行工作流

创建 Python 文件以执行工作流：

import asyncio
from pathlib import Path

from agent_framework.declarative import WorkflowFactory


async def main() -> None:
    """Run the greeting workflow."""
    # Create a workflow factory
    factory = WorkflowFactory()

    # Load the workflow from YAML
    workflow_path = Path(__file__).parent / "greeting-workflow.yaml"
    workflow = factory.create_workflow_from_yaml_path(workflow_path)

    print(f"Loaded workflow: {workflow.name}")
    print("-" * 40)

    # Run with a name input
    result = await workflow.run({"name": "Alice"})
    for output in result.get_outputs():
        print(f"Output: {output}")
    for output in result.get_intermediate_outputs():
        print(f"Intermediate: {output}")


if __name__ == "__main__":
    asyncio.run(main())

预期输出

Loaded workflow: greeting-workflow
----------------------------------------
Output: Hello, Alice!

核心概念

变量命名空间

声明性工作流使用命名空间变量来组织状态：

Namespace	Description	Example
`Local.*`	工作流本地变量	`Local.message`
`Workflow.Inputs.*`	输入参数	`Workflow.Inputs.name`
`Workflow.Outputs.*`	输出值	`Workflow.Outputs.result`
`System.*`	系统提供的值	`System.ConversationId`

表达式语言

值以 = 为前缀时，将被计算为表达式：

# Literal value (no evaluation)
value: Hello

# Expression (evaluated at runtime)
value: =Concat("Hello, ", Workflow.Inputs.name)

常见函数包括：

Concat(str1, str2, ...) - 连接字符串
If(condition, trueValue, falseValue) - 条件表达式
IsBlank(value) - 检查值是否为空

操作类型

声明性工作流支持各种动作类型：

类别	行动
变量管理	`SetVariable`、`SetMultipleVariables`、`AppendValue`、`ResetVariable`
控制流	`If`、`ConditionGroup`、`Foreach`、`RepeatUntil`、`BreakLoop`、`ContinueLoop`、`GotoAction`
输出	`SendActivity`、`EmitEvent`
代理调用	`InvokeAzureAgent`
工具调用	`InvokeFunctionTool`、`InvokeMcpTool`
HTTP	`HttpRequestAction`
人在环	`Question`、`Confirmation`、`RequestExternalInput`、`WaitForInput`
工作流控件	`EndWorkflow`、`EndConversation`、`CreateConversation`

操作参考

动作是声明性工作流的基础模块。每个动作执行特定操作，并且动作按照 YAML 文件中显示的顺序执行。

动作结构

所有操作具备共同属性：

- kind: ActionType      # Required: The type of action
  id: unique_id         # Optional: Unique identifier for referencing
  displayName: Name     # Optional: Human-readable name for logging
  # Action-specific properties...

变量管理操作

设置变量

将变量设置为指定的值。

- kind: SetVariable
  id: set_greeting
  displayName: Set greeting message
  variable: Local.greeting
  value: Hello World

使用表达式：

- kind: SetVariable
  variable: Local.fullName
  value: =Concat(Workflow.Inputs.firstName, " ", Workflow.Inputs.lastName)

性能：

资产	必选	Description
`variable`	是的	变量路径（例如， `Local.name`， `Workflow.Outputs.result`）
`value`	是的	要设置的值（文本或表达式）

注释

Python 还支持SetValue操作类型，该类型使用path而不是variable来作为目标属性。 SetVariable （与variable）和 SetValue （与path）都取得了相同的结果。例如：

- kind: SetValue
  id: set_greeting
  path: Local.greeting
  value: Hello World

设置多个变量

在单个操作中设置多个变量。

- kind: SetMultipleVariables
  id: initialize_vars
  displayName: Initialize variables
  variables:
    Local.counter: 0
    Local.status: pending
    Local.message: =Concat("Processing order ", Workflow.Inputs.orderId)

性能：

资产	必选	Description
`variables`	是的	将变量路径映射到值

AppendValue

将值追加到列表或连接到字符串。

- kind: AppendValue
  id: add_item
  variable: Local.items
  value: =Workflow.Inputs.newItem

性能：

资产	必选	Description
`variable`	是的	要追加到的变量路径
`value`	是的	要追加的值

重置变量

清除变量的值。

- kind: ResetVariable
  id: clear_counter
  variable: Local.counter

性能：

资产	必选	Description
`variable`	是的	要重置的变量路径

控制流操作

If

根据条件执行操作。

- kind: If
  id: check_age
  displayName: Check user age
  condition: =Workflow.Inputs.age >= 18
  then:
    - kind: SendActivity
      activity:
        text: "Welcome, adult user!"
  else:
    - kind: SendActivity
      activity:
        text: "Welcome, young user!"

嵌套条件：

- kind: If
  condition: =Workflow.Inputs.role = "admin"
  then:
    - kind: SendActivity
      activity:
        text: "Admin access granted"
  else:
    - kind: If
      condition: =Workflow.Inputs.role = "user"
      then:
        - kind: SendActivity
          activity:
            text: "User access granted"
      else:
        - kind: SendActivity
          activity:
            text: "Access denied"

性能：

资产	必选	Description
`condition`	是的	计算结果为 true/false 的表达式
`then`	是的	如果条件为真，则执行的动作
`else`	否	如果条件为假，则执行的操作

ConditionGroup

计算多个条件，如 switch/case 语句。

- kind: ConditionGroup
  id: route_by_category
  displayName: Route based on category
  conditions:
    - condition: =Workflow.Inputs.category = "electronics"
      id: electronics_branch
      actions:
        - kind: SetVariable
          variable: Local.department
          value: Electronics Team
    - condition: =Workflow.Inputs.category = "clothing"
      id: clothing_branch
      actions:
        - kind: SetVariable
          variable: Local.department
          value: Clothing Team
    - condition: =Workflow.Inputs.category = "food"
      id: food_branch
      actions:
        - kind: SetVariable
          variable: Local.department
          value: Food Team
  elseActions:
    - kind: SetVariable
      variable: Local.department
      value: General Support

性能：

资产	必选	Description
`conditions`	是的	条件/作对列表（第一场比赛获胜）
`elseActions`	否	如果没有条件匹配时的操作步骤

Foreach

循环访问集合。

- kind: Foreach
  id: process_items
  displayName: Process each item
  source: =Workflow.Inputs.items
  itemName: item
  indexName: index
  actions:
    - kind: SendActivity
      activity:
        text: =Concat("Processing item ", index, ": ", item)

性能：

资产	必选	Description
`source`	是的	返回集合的表达式
`itemName`	否	当前项的变量名称（默认值： `item`）
`indexName`	否	当前索引的变量名称（默认值： `index`）
`actions`	是的	要为每个项目执行的操作

重复直到

重复动作，直到条件为真。

- kind: SetVariable
  variable: Local.counter
  value: 0

- kind: RepeatUntil
  id: count_loop
  displayName: Count to 5
  condition: =Local.counter >= 5
  actions:
    - kind: SetVariable
      variable: Local.counter
      value: =Local.counter + 1
    - kind: SendActivity
      activity:
        text: =Concat("Counter: ", Local.counter)

性能：

资产	必选	Description
`condition`	是的	循环一直持续到此为真
`actions`	是的	要重复的动作

BreakLoop

立即退出当前循环。

- kind: Foreach
  source: =Workflow.Inputs.items
  actions:
    - kind: If
      condition: =item = "stop"
      then:
        - kind: BreakLoop
    - kind: SendActivity
      activity:
        text: =item

ContinueLoop

跳过循环的下一次迭代。

- kind: Foreach
  source: =Workflow.Inputs.numbers
  actions:
    - kind: If
      condition: =item < 0
      then:
        - kind: ContinueLoop
    - kind: SendActivity
      activity:
        text: =Concat("Positive number: ", item)

GotoAction

按 ID 跳转到特定操作。

- kind: SetVariable
  id: start_label
  variable: Local.attempts
  value: =Local.attempts + 1

- kind: SendActivity
  activity:
    text: =Concat("Attempt ", Local.attempts)

- kind: If
  condition: =And(Local.attempts < 3, Not(Local.success))
  then:
    - kind: GotoAction
      actionId: start_label

性能：

资产	必选	Description
`actionId`	是的	要跳转到的动作 ID

输出动作

SendActivity

向用户发送消息。

- kind: SendActivity
  id: send_welcome
  displayName: Send welcome message
  activity:
    text: "Welcome to our service!"

使用表达式：

- kind: SendActivity
  activity:
    text: =Concat("Hello, ", Workflow.Inputs.name, "! How can I help you today?")

性能：

资产	必选	Description
`activity`	是的	要发送的活动
`activity.text`	是的	消息文本（文本或表达式）

EmitEvent

发出自定义事件。

- kind: EmitEvent
  id: emit_status
  displayName: Emit status event
  eventType: order_status_changed
  data:
    orderId: =Workflow.Inputs.orderId
    status: =Local.newStatus

性能：

资产	必选	Description
`eventType`	是的	事件的类型标识符
`data`	否	事件有效负载数据

代理调用操作

InvokeAzureAgent

调用 Azure AI 代理。

基本调用：

- kind: InvokeAzureAgent
  id: call_assistant
  displayName: Call assistant agent
  agent:
    name: AssistantAgent
  conversationId: =System.ConversationId

通过输入和输出配置：

- kind: InvokeAzureAgent
  id: call_analyst
  displayName: Call analyst agent
  agent:
    name: AnalystAgent
  conversationId: =System.ConversationId
  input:
    messages: =Local.userMessage
    arguments:
      topic: =Workflow.Inputs.topic
  output:
    responseObject: Local.AnalystResult
    messages: Local.AnalystMessages
    autoSend: true

使用外部循环（持续运行直到满足条件）：

- kind: InvokeAzureAgent
  id: support_agent
  agent:
    name: SupportAgent
  input:
    externalLoop:
      when: =Not(Local.IsResolved)
  output:
    responseObject: Local.SupportResult

性能：

资产	必选	Description
`agent.name`	是的	已注册代理的名称
`conversationId`	否	对话上下文标识符
`input.messages`	否	发送给代理的消息
`input.arguments`	否	代理的其他参数
`input.externalLoop.when`	否	继续代理循环的条件
`output.responseObject`	否	存储代理响应的路径
`output.messages`	否	存储对话消息的路径
`output.autoSend`	否	自动向用户发送响应

工具和 HTTP 操作

InvokeFunctionTool

直接从工作流调用已注册的 Python 函数，而无需通过 AI 代理。

- kind: InvokeFunctionTool
  id: invoke_weather
  displayName: Get weather data
  functionName: get_weather
  arguments:
    location: =Local.location
    unit: =Local.unit
  output:
    result: Local.weatherInfo
    messages: Local.weatherToolCallItems
    autoSend: true

性能：

资产	必选	Description
`functionName`	是的	要调用的已注册函数的名称
`arguments`	否	要传递给函数的参数
`output.result`	否	存储函数结果的路径
`output.messages`	否	存储函数消息的路径
`output.autoSend`	否	自动向用户发送结果

InvokeFunctionTool 的 Python 设置：

函数必须注册到 WorkflowFactory 使用 register_tool：

from agent_framework.declarative import WorkflowFactory

# Define your functions
def get_weather(location: str, unit: str = "F") -> dict:
    """Get weather information for a location."""
    # Your implementation here
    return {"location": location, "temp": 72, "unit": unit}

def format_message(template: str, data: dict) -> str:
    """Format a message template with data."""
    return template.format(**data)

# Register functions with the factory
factory = (
    WorkflowFactory()
    .register_tool("get_weather", get_weather)
    .register_tool("format_message", format_message)
)

# Load and run the workflow
workflow = factory.create_workflow_from_yaml_path("workflow.yaml")
result = await workflow.run({"location": "Seattle", "unit": "F"})

InvokeMcpTool

通过配置的 MCPToolHandler 在 MCP 服务器上调用工具。

- kind: InvokeMcpTool
  id: search_docs
  serverUrl: https://learn.microsoft.com/api/mcp
  serverLabel: microsoft_docs
  toolName: microsoft_docs_search
  arguments:
    query: =Local.searchQuery
  output:
    result: Local.searchResults
    messages: Local.toolMessage
    autoSend: true

性能：

资产	必选	Description
`serverUrl`	是的	MCP 服务器 URL
`toolName`	是的	MCP 服务器上的工具名称
`serverLabel`	否	人工可读服务器标签
`arguments`	否	传递给工具的参数
`headers`	否	请求标头;跳过空值
`connection.name`	否	自定义处理程序的命名连接
`conversationId`	否	将成功生成的工具输出添加到对话中
`requireApproval`	否	在调用该工具之前请求审批
`output.result`	否	存储已分析工具输出的路径
`output.messages`	否	存储工具消息的路径
`output.autoSend`	否	向工作流结果发出工具输出;默认值为 `true`

InvokeMcpTool 的 Python 环境设置：

将 MCP 工具处理程序传递给 WorkflowFactory。需要身份验证、托管连接或 URL 允许列表时，请使用自定义处理程序。

from agent_framework.declarative import DefaultMCPToolHandler, WorkflowFactory

factory = WorkflowFactory(mcp_tool_handler=DefaultMCPToolHandler())
workflow = factory.create_workflow_from_yaml_path("workflow.yaml")

HttpRequestAction

通过配置的 HttpRequestHandler 发送 HTTP 请求。在分配之前，成功的 JSON 响应会被解析；非 2xx 响应会导致操作失败。

- kind: HttpRequestAction
  id: fetch_repo_info
  method: GET
  url: =Concat("https://api.github.com/repos/", Local.repoName)
  headers:
    Accept: application/vnd.github+json
    User-Agent: agent-framework
  queryParameters:
    per_page: 10
  response: Local.repoInfo
  responseHeaders: Local.repoHeaders

性能：

资产	必选	Description
`url`	是的	绝对请求 URL
`method`	否	HTTP 方法;默认值为 `GET`
`headers`	否	请求标头
`queryParameters`	否	附加到 URL 的查询参数
`body`	否	请求正文；使用`kind: json`、`raw`或`none`。
`requestTimeoutInMilliseconds`	否	每个请求超时
`connection.name`	否	自定义处理程序的命名连接
`conversationId`	否	将成功的响应正文添加到对话
`response`	否	存储已分析响应正文的路径
`responseHeaders`	否	存储响应标头的路径

Python 设置的 HttpRequestAction:

将 HTTP 请求处理程序传递给 WorkflowFactory. 需要身份验证、重试或 URL 允许列表时，请使用自定义处理程序。

from agent_framework.declarative import DefaultHttpRequestHandler, WorkflowFactory

factory = WorkflowFactory(http_request_handler=DefaultHttpRequestHandler())
workflow = factory.create_workflow_from_yaml_path("workflow.yaml")

人机互动操作

问题

向用户提问并存储响应。

- kind: Question
  id: ask_name
  displayName: Ask for user name
  question:
    text: "What is your name?"
  variable: Local.userName
  default: "Guest"

性能：

资产	必选	Description
`question.text`	是的	要提问的问题
`variable`	是的	存储响应的路径
`default`	否	如果未响应，则默认值

确认

要求用户进行“是/否”确认。

- kind: Confirmation
  id: confirm_delete
  displayName: Confirm deletion
  question:
    text: "Are you sure you want to delete this item?"
  variable: Local.confirmed

性能：

资产	必选	Description
`question.text`	是的	确认问题
`variable`	是的	用于存储布尔结果的路径

RequestExternalInput

从外部系统或进程请求输入。

- kind: RequestExternalInput
  id: request_approval
  displayName: Request manager approval
  prompt:
    text: "Please provide approval for this request."
  variable: Local.approvalResult
  default: "pending"

性能：

资产	必选	Description
`prompt.text`	是的	所需输入的说明
`variable`	是的	存储输入的路径
`default`	否	默认值

等待输入

暂停工作流并等待外部输入。

- kind: WaitForInput
  id: wait_for_response
  variable: Local.externalResponse

性能：

资产	必选	Description
`variable`	是的	接收到输入时的存储路径

工作流控制动作

EndWorkflow

终止工作流执行。

- kind: EndWorkflow
  id: finish
  displayName: End workflow

EndConversation

结束当前对话。

- kind: EndConversation
  id: end_chat
  displayName: End conversation

CreateConversation

创建新的对话上下文。

- kind: CreateConversation
  id: create_new_conv
  displayName: Create new conversation
  conversationId: Local.NewConversationId

性能：

资产	必选	Description
`conversationId`	是的	存储新会话 ID 的路径

操作快速参考

Action	类别	Description
`SetVariable`	Variable	设置单个变量
`SetMultipleVariables`	Variable	设置多个变量
`AppendValue`	Variable	追加到列表/字符串
`ResetVariable`	Variable	清除变量
`If`	控制流	条件分支
`ConditionGroup`	控制流	多分支交换机
`Foreach`	控制流	循环访问集合
`RepeatUntil`	控制流	循环直到满足条件
`BreakLoop`	控制流	退出当前循环
`ContinueLoop`	控制流	跳至下一次迭代
`GotoAction`	控制流	跳转到动作按 ID
`SendActivity`	输出	向用户发送消息
`EmitEvent`	输出	发出自定义事件
`InvokeAzureAgent`	代理人	调用 Azure AI 代理
`InvokeFunctionTool`	工具	调用已注册的函数
`InvokeMcpTool`	工具	调用 MCP 服务器工具
`HttpRequestAction`	HTTP	调用 HTTP 终结点
`Question`	人在环	向用户提问
`Confirmation`	人在环	是/否确认
`RequestExternalInput`	人在环	请求外部输入
`WaitForInput`	人在环	等待输入
`EndWorkflow`	工作流控件	终止工作流
`EndConversation`	工作流控件	结束对话
`CreateConversation`	工作流控件	创建新对话

表达式语法

声明性工作流使用类似于 PowerFx 的表达式语言来管理状态和计算动态值。带有前缀 = 的值在运行时作为表达式计算。

变量命名空间详细信息

Namespace	Description	Access
`Local.*`	工作流局部变量	读/写
`Workflow.Inputs.*`	传递给工作流的输入参数	只读
`Workflow.Outputs.*`	从工作流返回的值	读/写
`System.*`	系统提供的值	只读
`Agent.*`	代理调用的结果	只读

系统变量

Variable	Description
`System.ConversationId`	当前聊天标识符
`System.LastMessage`	最新消息
`System.Timestamp`	当前时间戳

代理变量

调用代理后，通过输出变量访问响应数据：

actions:
  - kind: InvokeAzureAgent
    id: call_assistant
    agent:
      name: MyAgent
    output:
      responseObject: Local.AgentResult

  # Access agent response
  - kind: SendActivity
    activity:
      text: =Local.AgentResult.text

文本值与表达式值

# Literal string (stored as-is)
value: Hello World

# Expression (evaluated at runtime)
value: =Concat("Hello ", Workflow.Inputs.name)

# Literal number
value: 42

# Expression returning a number
value: =Workflow.Inputs.quantity * 2

字符串操作

Concat

连接多个字符串：

value: =Concat("Hello, ", Workflow.Inputs.name, "!")
# Result: "Hello, Alice!" (if Workflow.Inputs.name is "Alice")

value: =Concat(Local.firstName, " ", Local.lastName)
# Result: "John Doe" (if firstName is "John" and lastName is "Doe")

IsBlank

检查值是否为空或未定义：

condition: =IsBlank(Workflow.Inputs.optionalParam)
# Returns true if the parameter is not provided

value: =If(IsBlank(Workflow.Inputs.name), "Guest", Workflow.Inputs.name)
# Returns "Guest" if name is blank, otherwise returns the name

条件表达式

If 函数

根据条件返回不同的值：

value: =If(Workflow.Inputs.age < 18, "minor", "adult")

value: =If(Local.count > 0, "Items found", "No items")

# Nested conditions
value: =If(Workflow.Inputs.role = "admin", "Full access", If(Workflow.Inputs.role = "user", "Limited access", "No access"))

比较运算符

Operator	Description	Example
`=`	等于	`=Workflow.Inputs.status = "active"`
`<>`	不等于	`=Workflow.Inputs.status <> "deleted"`
`<`	小于	`=Workflow.Inputs.age < 18`
`>`	大于	`=Workflow.Inputs.count > 0`
`<=`	小于或等于	`=Workflow.Inputs.score <= 100`
`>=`	大于或等于	`=Workflow.Inputs.quantity >= 1`

布尔函数

# Or - returns true if any condition is true
condition: =Or(Workflow.Inputs.role = "admin", Workflow.Inputs.role = "moderator")

# And - returns true if all conditions are true
condition: =And(Workflow.Inputs.age >= 18, Workflow.Inputs.hasConsent)

# Not - negates a condition
condition: =Not(IsBlank(Workflow.Inputs.email))

数学运算

# Addition
value: =Workflow.Inputs.price + Workflow.Inputs.tax

# Subtraction
value: =Workflow.Inputs.total - Workflow.Inputs.discount

# Multiplication
value: =Workflow.Inputs.quantity * Workflow.Inputs.unitPrice

# Division
value: =Workflow.Inputs.total / Workflow.Inputs.count

实用表达式示例

用户分类

name: categorize-user
inputs:
  age:
    type: integer
    description: User's age

actions:
  - kind: SetVariable
    variable: Local.age
    value: =Workflow.Inputs.age

  - kind: SetVariable
    variable: Local.category
    value: =If(Local.age < 13, "child", If(Local.age < 20, "teenager", If(Local.age < 65, "adult", "senior")))

  - kind: SendActivity
    activity:
      text: =Concat("You are categorized as: ", Local.category)

  - kind: SetVariable
    variable: Workflow.Outputs.category
    value: =Local.category

条件问候

name: smart-greeting
inputs:
  name:
    type: string
    description: User's name (optional)
  timeOfDay:
    type: string
    description: morning, afternoon, or evening

actions:
  # Set the greeting based on time of day
  - kind: SetVariable
    variable: Local.timeGreeting
    value: =If(Workflow.Inputs.timeOfDay = "morning", "Good morning", If(Workflow.Inputs.timeOfDay = "afternoon", "Good afternoon", "Good evening"))

  # Handle optional name
  - kind: SetVariable
    variable: Local.userName
    value: =If(IsBlank(Workflow.Inputs.name), "friend", Workflow.Inputs.name)

  # Build the full greeting
  - kind: SetVariable
    variable: Local.fullGreeting
    value: =Concat(Local.timeGreeting, ", ", Local.userName, "!")

  - kind: SendActivity
    activity:
      text: =Local.fullGreeting

输入验证

name: validate-order
inputs:
  quantity:
    type: integer
    description: Number of items to order
  email:
    type: string
    description: Customer email

actions:
  # Check if inputs are valid
  - kind: SetVariable
    variable: Local.isValidQuantity
    value: =And(Workflow.Inputs.quantity > 0, Workflow.Inputs.quantity <= 100)

  - kind: SetVariable
    variable: Local.hasEmail
    value: =Not(IsBlank(Workflow.Inputs.email))

  - kind: SetVariable
    variable: Local.isValid
    value: =And(Local.isValidQuantity, Local.hasEmail)

  - kind: If
    condition: =Local.isValid
    then:
      - kind: SendActivity
        activity:
          text: "Order validated successfully!"
    else:
      - kind: SendActivity
        activity:
          text: =If(Not(Local.isValidQuantity), "Invalid quantity (must be 1-100)", "Email is required")

高级模式

随着工作流的复杂性的增长，需要处理多步骤流程、代理协调和交互式方案的模式。

多代理编排

顺序代理管道

按顺序传递多个代理，其中每个代理都基于上一个代理的输出生成。

用例：不同专家处理研究、写作和编辑的内容创建管道。

name: content-pipeline
description: Sequential agent pipeline for content creation

kind: Workflow
trigger:
  kind: OnConversationStart
  id: content_workflow
  actions:
    # First agent: Research and analyze
    - kind: InvokeAzureAgent
      id: invoke_researcher
      displayName: Research phase
      conversationId: =System.ConversationId
      agent:
        name: ResearcherAgent

    # Second agent: Write draft based on research
    - kind: InvokeAzureAgent
      id: invoke_writer
      displayName: Writing phase
      conversationId: =System.ConversationId
      agent:
        name: WriterAgent

    # Third agent: Edit and polish
    - kind: InvokeAzureAgent
      id: invoke_editor
      displayName: Editing phase
      conversationId: =System.ConversationId
      agent:
        name: EditorAgent

Python 设置：

from agent_framework.declarative import WorkflowFactory

# Create factory and register agents
factory = WorkflowFactory()
factory.register_agent("ResearcherAgent", researcher_agent)
factory.register_agent("WriterAgent", writer_agent)
factory.register_agent("EditorAgent", editor_agent)

# Load and run
workflow = factory.create_workflow_from_yaml_path("content-pipeline.yaml")
result = await workflow.run({"topic": "AI in healthcare"})

条件代理路由

根据输入或中间结果将请求路由到不同的代理。

用例：支持基于问题类型路由到专用代理的系统。

name: support-router
description: Route to specialized support agents

inputs:
  category:
    type: string
    description: Support category (billing, technical, general)

actions:
  - kind: ConditionGroup
    id: route_request
    displayName: Route to appropriate agent
    conditions:
      - condition: =Workflow.Inputs.category = "billing"
        id: billing_route
        actions:
          - kind: InvokeAzureAgent
            id: billing_agent
            agent:
              name: BillingAgent
            conversationId: =System.ConversationId
      - condition: =Workflow.Inputs.category = "technical"
        id: technical_route
        actions:
          - kind: InvokeAzureAgent
            id: technical_agent
            agent:
              name: TechnicalAgent
            conversationId: =System.ConversationId
    elseActions:
      - kind: InvokeAzureAgent
        id: general_agent
        agent:
          name: GeneralAgent
        conversationId: =System.ConversationId

具有外部循环的代理

继续代理交互，直到满足条件，例如问题已解决。

用例：支持在解决用户问题之前继续的对话。

name: support-conversation
description: Continue support until resolved

actions:
  - kind: SetVariable
    variable: Local.IsResolved
    value: false

  - kind: InvokeAzureAgent
    id: support_agent
    displayName: Support agent with external loop
    agent:
      name: SupportAgent
    conversationId: =System.ConversationId
    input:
      externalLoop:
        when: =Not(Local.IsResolved)
    output:
      responseObject: Local.SupportResult

  - kind: SendActivity
    activity:
      text: "Thank you for contacting support. Your issue has been resolved."

循环控制模式

迭代代理对话

使用受控迭代在代理之间创建来回对话。

用例：学生教师方案、辩论模拟或迭代优化。

name: student-teacher
description: Iterative learning conversation between student and teacher

kind: Workflow
trigger:
  kind: OnConversationStart
  id: learning_session
  actions:
    # Initialize turn counter
    - kind: SetVariable
      id: init_counter
      variable: Local.TurnCount
      value: 0

    - kind: SendActivity
      id: start_message
      activity:
        text: =Concat("Starting session for: ", Workflow.Inputs.problem)

    # Student attempts solution (loop entry point)
    - kind: SendActivity
      id: student_label
      activity:
        text: "\n[Student]:"

    - kind: InvokeAzureAgent
      id: student_attempt
      conversationId: =System.ConversationId
      agent:
        name: StudentAgent

    # Teacher reviews
    - kind: SendActivity
      id: teacher_label
      activity:
        text: "\n[Teacher]:"

    - kind: InvokeAzureAgent
      id: teacher_review
      conversationId: =System.ConversationId
      agent:
        name: TeacherAgent
      output:
        messages: Local.TeacherResponse

    # Increment counter
    - kind: SetVariable
      id: increment
      variable: Local.TurnCount
      value: =Local.TurnCount + 1

    # Check completion conditions
    - kind: ConditionGroup
      id: check_completion
      conditions:
        # Success: Teacher congratulated student
        - condition: =Not(IsBlank(Find("congratulations", Local.TeacherResponse)))
          id: success_check
          actions:
            - kind: SendActivity
              activity:
                text: "Session complete - student succeeded!"
            - kind: SetVariable
              variable: Workflow.Outputs.result
              value: success
        # Continue: Under turn limit
        - condition: =Local.TurnCount < 4
          id: continue_check
          actions:
            - kind: GotoAction
              actionId: student_label
      elseActions:
        # Timeout: Reached turn limit
        - kind: SendActivity
          activity:
            text: "Session ended - turn limit reached."
        - kind: SetVariable
          variable: Workflow.Outputs.result
          value: timeout

基于计数器的循环

使用变量和 GotoAction 实现传统的计数循环。

name: counter-loop
description: Process items with a counter

actions:
  - kind: SetVariable
    variable: Local.counter
    value: 0

  - kind: SetVariable
    variable: Local.maxIterations
    value: 5

  # Loop start
  - kind: SetVariable
    id: loop_start
    variable: Local.counter
    value: =Local.counter + 1

  - kind: SendActivity
    activity:
      text: =Concat("Processing iteration ", Local.counter)

  # Your processing logic here
  - kind: SetVariable
    variable: Local.result
    value: =Concat("Result from iteration ", Local.counter)

  # Check if should continue
  - kind: If
    condition: =Local.counter < Local.maxIterations
    then:
      - kind: GotoAction
        actionId: loop_start
    else:
      - kind: SendActivity
        activity:
          text: "Loop complete!"

使用 BreakLoop 提前退出

满足条件时，使用 BreakLoop 提前退出迭代。

name: search-workflow
description: Search through items and stop when found

actions:
  - kind: SetVariable
    variable: Local.found
    value: false

  - kind: Foreach
    source: =Workflow.Inputs.items
    itemName: currentItem
    actions:
      # Check if this is the item we're looking for
      - kind: If
        condition: =currentItem.id = Workflow.Inputs.targetId
        then:
          - kind: SetVariable
            variable: Local.found
            value: true
          - kind: SetVariable
            variable: Local.result
            value: =currentItem
          - kind: BreakLoop

      - kind: SendActivity
        activity:
          text: =Concat("Checked item: ", currentItem.name)

  - kind: If
    condition: =Local.found
    then:
      - kind: SendActivity
        activity:
          text: =Concat("Found: ", Local.result.name)
    else:
      - kind: SendActivity
        activity:
          text: "Item not found"

人参与循环模式

交互式调查

从用户收集多条信息。

name: customer-survey
description: Interactive customer feedback survey

actions:
  - kind: SendActivity
    activity:
      text: "Welcome to our customer feedback survey!"

  # Collect name
  - kind: Question
    id: ask_name
    question:
      text: "What is your name?"
    variable: Local.userName
    default: "Anonymous"

  - kind: SendActivity
    activity:
      text: =Concat("Nice to meet you, ", Local.userName, "!")

  # Collect rating
  - kind: Question
    id: ask_rating
    question:
      text: "How would you rate our service? (1-5)"
    variable: Local.rating
    default: "3"

  # Respond based on rating
  - kind: If
    condition: =Local.rating >= 4
    then:
      - kind: SendActivity
        activity:
          text: "Thank you for the positive feedback!"
    else:
      - kind: Question
        id: ask_improvement
        question:
          text: "What could we improve?"
        variable: Local.feedback

  # Collect additional feedback
  - kind: RequestExternalInput
    id: additional_comments
    prompt:
      text: "Any additional comments? (optional)"
    variable: Local.comments
    default: ""

  # Summary
  - kind: SendActivity
    activity:
      text: =Concat("Thank you, ", Local.userName, "! Your feedback has been recorded.")

  - kind: SetVariable
    variable: Workflow.Outputs.survey
    value:
      name: =Local.userName
      rating: =Local.rating
      feedback: =Local.feedback
      comments: =Local.comments

审批工作流

在继续执行操作之前请求审批。

name: approval-workflow
description: Request approval before processing

inputs:
  requestType:
    type: string
    description: Type of request
  amount:
    type: number
    description: Request amount

actions:
  - kind: SendActivity
    activity:
      text: =Concat("Processing ", Workflow.Inputs.requestType, " request for $", Workflow.Inputs.amount)

  # Check if approval is needed
  - kind: If
    condition: =Workflow.Inputs.amount > 1000
    then:
      - kind: SendActivity
        activity:
          text: "This request requires manager approval."

      - kind: Confirmation
        id: get_approval
        question:
          text: =Concat("Do you approve this ", Workflow.Inputs.requestType, " request for $", Workflow.Inputs.amount, "?")
        variable: Local.approved

      - kind: If
        condition: =Local.approved
        then:
          - kind: SendActivity
            activity:
              text: "Request approved. Processing..."
          - kind: SetVariable
            variable: Workflow.Outputs.status
            value: approved
        else:
          - kind: SendActivity
            activity:
              text: "Request denied."
          - kind: SetVariable
            variable: Workflow.Outputs.status
            value: denied
    else:
      - kind: SendActivity
        activity:
          text: "Request auto-approved (under threshold)."
      - kind: SetVariable
        variable: Workflow.Outputs.status
        value: auto_approved

复杂编排

支持票证工作流

组合多个模式的综合示例：代理路由、条件逻辑和聊天管理。

name: support-ticket-workflow
description: Complete support ticket handling with escalation

kind: Workflow
trigger:
  kind: OnConversationStart
  id: support_workflow
  actions:
    # Initial self-service agent
    - kind: InvokeAzureAgent
      id: self_service
      displayName: Self-service agent
      agent:
        name: SelfServiceAgent
      conversationId: =System.ConversationId
      input:
        externalLoop:
          when: =Not(Local.ServiceResult.IsResolved)
      output:
        responseObject: Local.ServiceResult

    # Check if resolved by self-service
    - kind: If
      condition: =Local.ServiceResult.IsResolved
      then:
        - kind: SendActivity
          activity:
            text: "Issue resolved through self-service."
        - kind: SetVariable
          variable: Workflow.Outputs.resolution
          value: self_service
        - kind: EndWorkflow
          id: end_resolved

    # Create support ticket
    - kind: SendActivity
      activity:
        text: "Creating support ticket..."

    - kind: SetVariable
      variable: Local.TicketId
      value: =Concat("TKT-", System.ConversationId)

    # Route to appropriate team
    - kind: ConditionGroup
      id: route_ticket
      conditions:
        - condition: =Local.ServiceResult.Category = "technical"
          id: technical_route
          actions:
            - kind: InvokeAzureAgent
              id: technical_support
              agent:
                name: TechnicalSupportAgent
              conversationId: =System.ConversationId
              output:
                responseObject: Local.TechResult
        - condition: =Local.ServiceResult.Category = "billing"
          id: billing_route
          actions:
            - kind: InvokeAzureAgent
              id: billing_support
              agent:
                name: BillingSupportAgent
              conversationId: =System.ConversationId
              output:
                responseObject: Local.BillingResult
      elseActions:
        # Escalate to human
        - kind: SendActivity
          activity:
            text: "Escalating to human support..."
        - kind: SetVariable
          variable: Workflow.Outputs.resolution
          value: escalated

    - kind: SendActivity
      activity:
        text: =Concat("Ticket ", Local.TicketId, " has been processed.")

最佳做法

命名约定

对作和变量使用清晰的描述性名称：

# Good
- kind: SetVariable
  id: calculate_total_price
  variable: Local.orderTotal

# Avoid
- kind: SetVariable
  id: sv1
  variable: Local.x

组织大型工作流

使用注释将复杂的工作流分解为逻辑部分：

actions:
  # === INITIALIZATION ===
  - kind: SetVariable
    id: init_status
    variable: Local.status
    value: started

  # === DATA COLLECTION ===
  - kind: Question
    id: collect_name
    # ...

  # === PROCESSING ===
  - kind: InvokeAzureAgent
    id: process_request
    # ...

  # === OUTPUT ===
  - kind: SendActivity
    id: send_result
    # ...

错误处理

使用条件检查来处理潜在问题：

actions:
  - kind: SetVariable
    variable: Local.hasError
    value: false

  - kind: InvokeAzureAgent
    id: call_agent
    agent:
      name: ProcessingAgent
    output:
      responseObject: Local.AgentResult

  - kind: If
    condition: =IsBlank(Local.AgentResult)
    then:
      - kind: SetVariable
        variable: Local.hasError
        value: true
      - kind: SendActivity
        activity:
          text: "An error occurred during processing."
    else:
      - kind: SendActivity
        activity:
          text: =Local.AgentResult.message

测试策略

简单开始：在添加复杂性之前测试基本流
使用默认值：为输入提供合理的默认值
添加日志记录：在开发过程中使用 SendActivity 进行调试
测试边缘事例：验证缺少或无效输入的行为

# Debug logging example
- kind: SendActivity
  id: debug_log
  activity:
    text: =Concat("[DEBUG] Current state: counter=", Local.counter, ", status=", Local.status)

后续步骤

C# 声明性工作流示例 - 探索完整的工作示例，包括：
- StudentTeacher - 与迭代学习的多代理对话
- InvokeMcpTool - MCP 服务器工具集成
- InvokeFunctionTool - 从工作流直接调用函数
- FunctionTools - 具有函数工具的代理
- ToolApproval - 工具执行的人工批准
- CustomerSupport - 复杂的支持票证工作流
- DeepResearch - 具有多个代理的研究工作流

Python 声明性工作流示例 - 探索完整的工作示例

反馈

此页面是否有帮助？

Last updated on 2026-05-11