在PromptFlow中创建动态列表工具输入的完整指南

在PromptFlow中创建动态列表工具输入的完整指南

你是否曾经遇到过这样的场景：需要让用户从一组动态生成的数据中选择输入，而不是固定的预定义选项？PromptFlow的动态列表输入功能正是为此而生！本文将深入探讨如何在PromptFlow中创建和使用动态列表工具输入，让你的AI应用更加智能和灵活。

什么是动态列表输入？

动态列表输入（Dynamic List Input）是PromptFlow中的一个强大功能，它允许工具在运行时动态生成输入选项列表。与传统的静态下拉菜单不同，动态列表可以根据上下文、用户输入或其他参数实时生成可选项。

核心优势

动态列表函数的基本结构

在PromptFlow中，动态列表输入通过特殊的函数来实现。这些函数需要遵循特定的返回格式：

from typing import List, Dict, Union

def dynamic_list_function(**kwargs) -> List[Dict[str, Union[str, int, float, list, Dict]]]:
    # 函数实现
    return [
        {
            "value": "实际值",
            "display_value": "显示值（可选）",
            "hyperlink": "超链接（可选）", 
            "description": "描述信息（可选）"
        }
    ]

返回字段详解

实战：创建第一个动态列表工具

让我们通过一个完整的示例来学习如何创建动态列表工具。

步骤1：创建工具包结构

首先，我们需要创建一个工具包。假设我们的项目结构如下：

my_tool_package/
├── __init__.py
├── tools/
│   ├── __init__.py
│   └── tool_with_dynamic_list_input.py
└── setup.py

步骤2：实现动态列表函数

在 tool_with_dynamic_list_input.py 中，我们创建两个关键函数：

from promptflow.core import tool
from typing import List, Union, Dict
import random

def my_list_func(prefix: str = "", size: int = 10, **kwargs) -> List[Dict[str, Union[str, int, float, list, Dict]]]:
    """生成动态水果列表的示例函数
    
    :param prefix: 每个选项的前缀
    :param size: 生成的选项数量
    :return: 包含动态选项的列表
    """
    words = ["苹果", "香蕉", "樱桃", "枣", "接骨木莓", 
             "无花果", "葡萄", "蜜瓜", "猕猴桃", "柠檬"]
    
    result = []
    for i in range(size):
        random_word = f"{random.choice(words)}{i}"
        cur_item = {
            "value": random_word,
            "display_value": f"{prefix}_{random_word}",
            "hyperlink": f'https://www.example.com/search?q={random_word}',
            "description": f"这是第{i}个水果选项"
        }
        result.append(cur_item)
    
    return result

@tool
def my_tool(input_prefix: str, input_text: list, endpoint_name: str) -> str:
    """使用动态列表输入的主工具函数"""
    return f"你好 {input_prefix} {','.join(input_text)} {endpoint_name}"

步骤3：配置Flow定义

创建 flow.dag.yaml 文件来定义我们的流程：

$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json

inputs: {}
outputs:
  output:
    type: string
    reference: ${My_Dynamic_List_Tool.output}

nodes:
- name: My_Dynamic_List_Tool
  type: python
  source:
    type: package
    tool: my_tool_package.tools.tool_with_dynamic_list_input.my_tool
  inputs:
    input_prefix: 欢迎
    input_text:
    - 葡萄3
    - 接骨木莓5
    endpoint_name: 我的终端点

高级应用场景

场景1：Azure资源动态选择

动态列表输入特别适合与云资源集成。以下是一个获取Azure ML终端点列表的示例：

def list_endpoint_names(subscription_id: str = None,
                        resource_group_name: str = None,
                        workspace_name: str = None,
                        prefix: str = "") -> List[Dict[str, str]]:
    """获取Azure ML终端点列表的动态函数"""
    
    if not all([subscription_id, resource_group_name, workspace_name]):
        return []
    
    from azure.ai.ml import MLClient
    from azure.identity import DefaultAzureCredential

    credential = DefaultAzureCredential()
    ml_client = MLClient(
        credential=credential,
        subscription_id=subscription_id,
        resource_group_name=resource_group_name,
        workspace_name=workspace_name)
    
    result = []
    for endpoint in ml_client.online_endpoints.list():
        hyperlink = (
            f"https://ml.azure.com/endpoints/realtime/{endpoint.name}/detail"
            f"?wsid=/subscriptions/{subscription_id}/resourceGroups/"
            f"{resource_group_name}/providers/Microsoft.MachineLearningServices/"
            f"workspaces/{workspace_name}"
        )
        
        cur_item = {
            "value": endpoint.name,
            "display_value": f"{prefix}_{endpoint.name}",
            "hyperlink": hyperlink,
            "description": f"终端点: {endpoint.name}"
        }
        result.append(cur_item)
    
    return result

场景2：数据库驱动动态列表

def get_user_list(department: str = "", **kwargs) -> List[Dict[str, str]]:
    """从数据库获取用户列表的动态函数"""
    
    import sqlite3
    import os
    
    # 假设我们有一个用户数据库
    db_path = "/path/to/users.db"
    if not os.path.exists(db_path):
        return []
    
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    if department:
        cursor.execute("SELECT user_id, username, email FROM users WHERE department = ?", 
                      (department,))
    else:
        cursor.execute("SELECT user_id, username, email FROM users")
    
    users = cursor.fetchall()
    conn.close()
    
    result = []
    for user_id, username, email in users:
        result.append({
            "value": user_id,
            "display_value": username,
            "description": f"邮箱: {email}",
            "hyperlink": f"mailto:{email}"
        })
    
    return result

最佳实践和调试技巧

1. 错误处理

def safe_dynamic_list(**kwargs) -> List[Dict]:
    """带有错误处理的动态列表函数"""
    try:
        # 你的业务逻辑
        return generate_items()
    except Exception as e:
        import logging
        logging.error(f"动态列表生成失败: {e}")
        return [{
            "value": "error",
            "display_value": "生成选项时出错",
            "description": str(e)
        }]

2. 性能优化

3. 测试策略

创建完整的测试套件来验证动态列表功能：

import pytest
from my_tool_package.tools.tool_with_dynamic_list_input import my_list_func

def test_dynamic_list_generation():
    """测试动态列表生成功能"""
    result = my_list_func(prefix="test", size=5)
    
    # 验证返回类型
    assert isinstance(result, list)
    assert len(result) == 5
    
    # 验证每个项的格式
    for item in result:
        assert "value" in item
        assert "display_value" in item
        assert item["display_value"].startswith("test_")

常见问题解答

Q1: 动态列表函数可以接受哪些参数？

动态列表函数可以接受任何类型的参数，但建议使用命名参数并提供默认值，以增强函数的健壮性。

Q2: 如何处理大量数据的动态列表？

对于大数据集，建议实现分页或搜索功能，或者限制返回的项数量以避免性能问题。

Q3: 动态列表支持异步操作吗？

是的，动态列表函数支持异步操作，但需要确保返回类型符合要求。

Q4: 如何调试动态列表不显示的问题？

检查函数返回值格式是否正确，确保每个项都包含必需的value字段。

总结

动态列表输入是PromptFlow中一个极其强大的功能，它让工具能够根据运行时上下文提供智能的输入选项。通过本文的指南，你应该已经掌握了：

1. ✅ 动态列表输入的基本概念和优势
2. ✅ 如何创建符合规范的动态列表函数
3. ✅ 高级应用场景和实际用例
4. ✅ 最佳实践和调试技巧
5. ✅ 常见问题的解决方案

现在，你可以开始在你的PromptFlow项目中实现动态列表输入，为用户提供更加智能和灵活的交互体验了！

记住，动态列表的核心价值在于其灵活性——让你的AI应用能够适应不断变化的业务需求，而不是被固定的输入选项所限制。