forked from microsoft/autogen
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathagentchat_nestedchat.py
172 lines (127 loc) · 6.77 KB
/
agentchat_nestedchat.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
#!/usr/bin/env python
# coding: utf-8
# # Solving Complex Tasks with Nested Chats
#
# This notebook shows how you can leverage **nested chats** to solve complex task with AutoGen. Nested chats is a sequence of chats created by a receiver agent after receiving a message from a sender agent and finished before the receiver agent replies to this message. Nested chats allow AutoGen agents to use other agents as their inner monologue to accomplish tasks. This abstraction is powerful as it allows you to compose agents in rich ways. This notebook shows two basic examples of nested chat.
#
# \:\:\:info Requirements
#
# Install `pyautogen`:
# ```bash
# pip install pyautogen
# ```
#
# For more information, please refer to the [installation guide](/docs/installation/).
#
# \:\:\:
# In[2]:
from typing_extensions import Annotated
import autogen
config_list = autogen.config_list_from_json(env_or_file="OAI_CONFIG_LIST")
llm_config = {"config_list": config_list}
# \:\:\:tip
#
# Learn more about the various ways to configure LLM endpoints [here](/docs/topics/llm_configuration).
#
# \:\:\:
# ### Example Task
#
# Suppose we want the agents to complete the following sequence of tasks:
# In[3]:
task = """Write a concise but engaging blogpost about Navida."""
# ## Scenario 1
#
# Let's say we desire the following workflow to solve the task: a user_proxy agent issues the initial query to a writer and acts as a proxy for the user. Whenever an initial writing is provided, a critic should be invoked to offer critique as feedback. This workflow can be realized by a three-agent system shown below. The system includes a user_proxy agent and a writer agent communicating with each other, with a critic agent nested within the user_proxy agent to provide critique. Whenever the user_proxy receives a message from the writer, it engages in a conversation with the critic agent to work out feedback on the writer's message.
#
# 
# ### Step 1. Define Agents
# Define the agents, including the outer agents writer and user_proxy, and the inner agent critic.
# In[4]:
writer = autogen.AssistantAgent(
name="Writer",
llm_config={"config_list": config_list},
system_message="""
You are a professional writer, known for your insightful and engaging articles.
You transform complex concepts into compelling narratives.
You should imporve the quality of the content based on the feedback from the user.
""",
)
user_proxy = autogen.UserProxyAgent(
name="User",
human_input_mode="NEVER",
is_termination_msg=lambda x: x.get("content", "").find("TERMINATE") >= 0,
code_execution_config={
"last_n_messages": 1,
"work_dir": "tasks",
"use_docker": False,
}, # Please set use_docker=True if docker is available to run the generated code. Using docker is safer than running the generated code directly.
)
critic = autogen.AssistantAgent(
name="Critic",
llm_config={"config_list": config_list},
system_message="""
You are a critic, known for your thoroughness and commitment to standards.
Your task is to scrutinize content for any harmful elements or regulatory violations, ensuring
all materials align with required guidelines.
For code
""",
)
# ### Step 2: Orchestrate Nested Chats to Solve Tasks
#
# In[24]:
def reflection_message(recipient, messages, sender, config):
print("Reflecting...", "yellow")
return f"Reflect and provide critique on the following writing. \n\n {recipient.chat_messages_for_summary(sender)[-1]['content']}"
user_proxy.register_nested_chats(
[{"recipient": critic, "message": reflection_message, "summary_method": "last_msg", "max_turns": 1}],
trigger=writer, # condition=my_condition,
)
res = user_proxy.initiate_chat(recipient=writer, message=task, max_turns=2, summary_method="last_msg")
# ## Scenarios 2
# Let's say we desire the following workflow to solve the task. Compared to scenario 1, we want to include an additional `critic_executor` agent to chat with the `critic` and execute some tool calls involved in the chat. For example, a tool for detecting harmful content in the output of the writer.
#
# This workflow can be realized by a four-agent system shown below. The system includes a user_proxy agent and a writer agent communicating with each other, with a chat between the `critic` and `critic_executor` agent nested within the `user_proxy` agent to provide critique. Whenever the user_proxy receives a message from the writer, it engages in a conversation between `critic` and `critic_executor` to work out feedback on the writer's message. A summary of the nested conversation will be passed to the user_proxy, which will then be passed to the writer as feedback.
#
# 
#
# In[10]:
critic_executor = autogen.UserProxyAgent(
name="Critic_Executor",
human_input_mode="NEVER",
# is_termination_msg=lambda x: x.get("content", "").find("TERMINATE") >= 0,
code_execution_config={
"last_n_messages": 1,
"work_dir": "tasks",
"use_docker": False,
}, # Please set use_docker=True if docker is available to run the generated code. Using docker is safer than running the generated code directly.
)
# one way of registering functions is to use the register_for_llm and register_for_execution decorators
@critic_executor.register_for_execution()
@critic.register_for_llm(name="check_harmful_content", description="Check if content contain harmful keywords.")
def check_harmful_content(content: Annotated[str, "Content to check if harmful keywords."]):
# List of harmful keywords for demonstration purposes
harmful_keywords = ["violence", "hate", "bullying", "death"]
# Normalize the input text to lower case to ensure case-insensitive matching
text = content.lower()
print(f"Checking for harmful content...{text}", "yellow")
# Check if any of the harmful keywords appear in the text
for keyword in harmful_keywords:
if keyword in text:
return "Denied. Harmful content detected:" + keyword # Harmful content detected
return "Approve. TERMINATE" # No harmful content detected
def reflection_message_no_harm(recipient, messages, sender, config):
print("Reflecting...", "yellow")
return f"Reflect and provide critique on the following writing. Ensure it does not contain harmful content. You can use tools to check it. \n\n {recipient.chat_messages_for_summary(sender)[-1]['content']}"
user_proxy.register_nested_chats(
[
{
"sender": critic_executor,
"recipient": critic,
"message": reflection_message_no_harm,
"max_turns": 2,
"summary_method": "last_msg",
}
],
trigger=writer, # condition=my_condition,
)
res = user_proxy.initiate_chat(recipient=writer, message=task, max_turns=2, summary_method="last_msg")