您现在的位置是：首页 >技术教程 >微软AutoGen介绍——人机交互向团队提供反馈的最佳实践网站首页技术教程

介绍

大家好，这次给大家分享的内容是微软AutoGen框架的核心功能Human-in-the-Loop(简称：人机交互)。那么这个又是什么东西呢，我们直接进入正题。

Human-in-the-Loop

在我的微软AutoGen介绍——Teams与智能体团队协作并使用-CSDN博客这篇博文中，我们已经了解了如何创建、观察和控制智能体团队。本次分享将重点介绍如何从我们的应用程序与团队进行交互，以及如何向团队提供人工反馈。

从我们的应用程序与团队进行交互主要有两种方式：

1. 在团队运行期间(即run()或run_stream()执行过程中)，通过用户代理智能体(UserProxyAgent)提供反馈。
2. 一旦运行结束，通过对下一次run()或run_stream()调用的输入来提供反馈。

运行期间提供反馈

用户代理智能体(UserProxyAgent)是一种特殊的内置智能体，充当用户向团队提供反馈的代理。

要使用用户代理智能体，我们可以创建它的一个实例，并在运行团队之前将其包含在团队中。团队会决定何时调用用户代理智能体，以向用户征求反馈。

以下图表说明了在团队运行期间，如何使用用户代理智能体从用户那里获取反馈：

加粗箭头表示团队运行期间的控制流：当团队调用用户代理智能体时，它将控制权转移给应用程序/用户，并等待反馈；一旦提供反馈，控制权就会转回给团队，团队继续执行。

由于这种方法具有阻塞性，建议仅将其用于需要用户即时反馈的简短交互场景，比如通过点击按钮来征求用户同意或不同意，或者弹出一个需立即关注的提醒，否则任务就会失败。

以下是在循环轮询小组聊天(RoundRobinGroupChat)中，针对诗歌创作任务使用用户代理智能体(UserProxyAgent)的示例：

完整代码

import asyncio
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 创建智能体。
model_client = OpenAIChatCompletionClient(model="gpt-3.5-turbo")
assistant = AssistantAgent("assistant", model_client=model_client)

# 使用input()从控制台获取用户输入.
user_proxy = UserProxyAgent("user_proxy", input_func=input)

# 创建终止条件，当用户说 “APPROVE” 时结束对话。
termination = TextMentionTermination("APPROVE")

# 创建团队。
team = RoundRobinGroupChat([assistant, user_proxy], termination_condition=termination)


async def main() -> None:
    # 运行对话并将内容输出到控制台。
    stream = team.run_stream(task="Write a 4-line poem about the ocean.")
    await Console(stream)

asyncio.run(main())

运行结果

---------- user ----------
Write a 4-line poem about the ocean.
---------- assistant ----------
Enter your response: The ocean's vast embrace, a symphony of blue,
Whispers of secrets in each wave that flew.
Where mysteries lie beneath the shimmering hue,
A realm of wonder, to explore and renew.

Terminate
APPROVE
---------- user_proxy ----------
APPROVE

进程已结束，退出代码为 0

从控制台输出中，我们可以看到团队通过用户代理(user_proxy)向用户征求反馈，以批准生成的诗歌。

我们可以向用户代理智能体(UserProxyAgent)提供自定义的输入函数，从而定制反馈流程。

为下次运行提供反馈

通常情况下，应用程序或用户会在一个交互循环中与智能体团队进行交互：团队运行直至结束，应用程序或用户提供反馈，然后团队结合反馈再次运行。
这种方法在团队与应用程序/用户之间进行异步通信的持久会话中很有用：一旦团队完成一次运行，应用程序会保存团队的状态，将其存储在持久性存储中，并在反馈到达时恢复团队运行。

分享反馈对于团队的运行和改进有什么作用？提供反馈的具体方式有哪些？如何确保反馈的有效性和准确性？

以下图表展示了这种方法中的控制流程：

实现这种方法有两种途径：

• 设定最大轮数，这样团队在达到指定轮数后总会停止运行。
• 利用诸如文本提及终止(TextMentionTermination)和交接终止(HandoffTermination)之类的终止条件，让团队依据自身内部状态，决定何时停止运行并将控制权交回。

我们可以同时使用这两种方法，以达成预期效果。

使用最大轮次设定

这种方法通过设置最大轮次，让我们能够暂停团队运行，等待用户输入。例如，我们可以将max_turns设为 1，这样团队在第一个智能体回应后就会停止。在需要用户持续参与的场景中，比如聊天机器人，这一方法尤为有用。

要实现这一点，需在RoundRobinGroupChat()构造函数中设置max_turns参数。

代码演示

team = RoundRobinGroupChat([...], max_turns=1)

团队一旦停止，轮次计数将被重置。当我们重新启动团队时，它将再次从0开始。不过，团队的内部状态会保留，例如，循环轮询小组交流(RoundRobinGroupChat)将从列表中的下一个智能体继续，且保留相同的对话历史记录。

博主笔记：max_turn特定于Team类，目前仅在循环轮询小组交流(RoundRobinGroupChat）、选择器小组交流(SelectorGroupChat)和集群(Swarm)中受支持。当与终止条件一起使用时，只要满足其中任一条件，团队就会停止运行。

以下是一个在循环轮询小组交流(RoundRobinGroupChat)中，针对诗歌创作任务使用max_turns并将其设置为最多1轮的示例：

完整代码

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 创建智能体。
model_client = OpenAIChatCompletionClient(model="gpt-3.5-turbo")
assistant = AssistantAgent("assistant", model_client=model_client)

# 创建团队，并将最大轮数设置为1。
team = RoundRobinGroupChat([assistant], max_turns=1)


async def main() -> None:
    task = "Write a 4-line poem about the ocean."
    while True:
        # 运行对话，并将内容输出至控制台。
        stream = team.run_stream(task=task)

        await Console(stream)
        # 获取用户的回复。
        task = input("Enter your feedback (type 'exit' to leave): ")
        if task.lower().strip() == "exit":
            break

asyncio.run(main())

运行结果

---------- user ----------
Write a 4-line poem about the ocean.
---------- assistant ----------
The ocean's vastness, blue and deep,
Where mysteries and secrets sleep,
Waves whisper tales of ancient lore,
A boundless world we can't ignore.

Terminate.
Enter your feedback (type 'exit' to leave): Can you make it about a person and its relationship with the ocean
---------- user ----------
Can you make it about a person and its relationship with the ocean
---------- assistant ----------
In the ocean's embrace, a soul finds peace,
Bonded with waves that never cease,
A dance of ebb and flow, a deep connection,
A human heart and the sea's affection.

Terminate.
Enter your feedback (type 'exit' to leave): exit

进程已结束，退出代码为 0

我们可以看到，在一个智能体做出回应后，团队立即停止了运行。

使用终止条件

在之前的分享中，我们已经学到了一些关于终止条件的示例。本次分享我们重点介绍“交接终止”(HandoffTermination)，当某个智能体发送“交接消息”(HandoffMessage)时，该条件会使团队停止运行。

让我们创建一个仅包含一个“助手智能体”(AssistantAgent)且设置了交接功能的团队，然后给这个团队安排一项任务。由于该智能体没有相关工具来继续处理任务，所以这项任务需要用户提供额外输入。

博主笔记：与助手智能体(AssistantAgent)配合使用的模型必须支持工具调用，才能使用交接功能。

完整代码

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.base import Handoff
from autogen_agentchat.conditions import HandoffTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 创建一个OpenAI模型客户端。
model_client = OpenAIChatCompletionClient(
    model="gpt-3.5-turbo",
    # api_key="sk-...", # 如果已设置“OPENAI_API_KEY”环境变量，此步骤可选。
)

# 创建一个总是将任务交接给用户的“偷懒型”助手智能体。
lazy_agent = AssistantAgent(
    "lazy_assistant",
    model_client=model_client,
    handoffs=[Handoff(target="user", message="Transfer to user.")],
    system_message="If you cannot complete the task, transfer to user. Otherwise, when finished, respond with "
                   "'TERMINATE'.",
)

# 定义一个用于检查交接消息的终止条件。
handoff_termination = HandoffTermination(target="user")

# 定义一个用于检查特定文本提及情况的终止条件。
text_termination = TextMentionTermination("TERMINATE")

# 创建一个由这个 “偷懒型” 助手智能体以及两种终止条件构成的单智能体团队。
lazy_agent_team = RoundRobinGroupChat([lazy_agent], termination_condition=handoff_termination | text_termination)


# 运行该团队，并将输出流式传输到控制台。
async def main() -> None:
    task = "What is the weather in Beijing?"
    await Console(lazy_agent_team.run_stream(task=task), output_stats=True)


asyncio.run(main())

运行结果

---------- user ----------
What is the weather in Beijing?
---------- lazy_assistant ----------
[FunctionCall(id='call_EAcMgrLGHdLw0e7iJGoMgxuu', arguments='{}', name='transfer_to_user')]
[Prompt tokens: 68, Completion tokens: 11]
---------- lazy_assistant ----------
[FunctionExecutionResult(content='Transfer to user.', call_id='call_EAcMgrLGHdLw0e7iJGoMgxuu')]
---------- lazy_assistant ----------
Transfer to user.
---------- Summary ----------
Number of messages: 4
Finish reason: Handoff to user from lazy_assistant detected.
Total prompt tokens: 68
Total completion tokens: 11
Duration: 34.85 seconds

进程已结束，退出代码为 0

我们可以看到，由于检测到交接消息，团队停止运行。现在，让我们提供智能体所需的信息，以便继续推进团队任务。

完整代码

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.base import Handoff
from autogen_agentchat.conditions import HandoffTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 创建一个OpenAI模型客户端。
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    # api_key="sk-...", # 如果已设置“OPENAI_API_KEY”环境变量，此步骤可选。
)

# 创建一个总是将任务交接给用户的“偷懒型”助手智能体。
lazy_agent = AssistantAgent(
    "lazy_assistant",
    model_client=model_client,
    handoffs=[Handoff(target="user", message="Transfer to user.")],
    system_message="If you cannot complete the task, transfer to user. Otherwise, when finished, respond with "
                   "'TERMINATE'.",
)

# 定义一个用于检查交接消息的终止条件。
handoff_termination = HandoffTermination(target="user")

# 定义一个用于检查特定文本提及情况的终止条件。
text_termination = TextMentionTermination("TERMINATE")

# 创建一个由这个 “偷懒型” 助手智能体以及两种终止条件构成的单智能体团队。
lazy_agent_team = RoundRobinGroupChat([lazy_agent], termination_condition=handoff_termination | text_termination)


# 运行该团队，并将输出流式传输到控制台。
async def main() -> None:
    task = "What is the weather in Beijing?"
    await Console(lazy_agent_team.run_stream(task=task), output_stats=True)
    await Console(lazy_agent_team.run_stream(task="The weather in Beijing is sunny."))

asyncio.run(main())

运行结果

---------- user ----------
What is the weather in Beijing?
---------- lazy_assistant ----------
[FunctionCall(id='call_ZrldEvdtwiR6K3z4NWQMdCmQ', arguments='{}', name='transfer_to_user')]
[Prompt tokens: 68, Completion tokens: 11]
---------- lazy_assistant ----------
[FunctionExecutionResult(content='Transfer to user.', call_id='call_ZrldEvdtwiR6K3z4NWQMdCmQ')]
---------- lazy_assistant ----------
Transfer to user.
---------- Summary ----------
Number of messages: 4
Finish reason: Handoff to user from lazy_assistant detected.
Total prompt tokens: 68
Total completion tokens: 11
Duration: 1.14 seconds
---------- user ----------
The weather in Beijing is sunny.
---------- lazy_assistant ----------
Great! If you have any more questions, feel free to ask.
---------- lazy_assistant ----------
TERMINATE

进程已结束，退出代码为 0

我们可以看到，在用户提供信息后，团队继续运行。

说明

如果大家在运行上述代码的时候有AutoGen相关的提示或报错(例如：该参数不存在，没有此类方法等)，请尝试更新一下AutoGen，博主在分享这篇博文的时候AutoGen的版本是0.4.5稳定版。

安装或更新命令

pip install -U "autogen-agentchat" "autogen-ext[openai,azure]"

另外大家要根据业务需求设置使用的LLM，不一定要按照我给大家分享代码中的设置来，如果只是为了测试并看运行结果可直接复制粘贴代码。

结束

好了，以上就是本次分享的全部内容。总体上看，微软AutoGen框架中的Human-in-the-Loop主要是指在智能体协作过程中引入用户参与和反馈的机制，通过Human-in-the-Loop机制，AutoGen框架能让用户在智能体系统运行的关键环节进行干预和提供反馈，使智能体的决策和任务执行更符合用户的需求和实际情况，提升系统的灵活性、准确性和可靠性，解决智能体可能出现的局限性问题。那么本次博主分享的内容大家理解了吗? 请大家多去大胆尝试和使用，博主也希望本次分享能够对大家有所帮助。如果大家对博主分享的内容感兴趣或有帮助，请点赞和关注，博主会持续更新关于微软AutoGen更多和更有趣的内容。