כיצד לבנות זרימות עבודה של AI בעולם האמיתי עם AutoGen: מדריך שלב אחר שלב

אז, שיחקתם עם מודלים של שפה גדולה ומתחילים לשלב AI Generative באפליקציות שלכם? זה מדהים! אבל בואו נהיה אמיתיים. לימודי תואר שני לא תמיד מתנהגים כמו שאנחנו רוצים שהם יתנהגו. הם כמו פעוטות מרושעים עם שכל משלהם!

עד מהרה אתה מבין ששרשרות הנחיות פשוטות פשוט לא מספיקות. לפעמים, אנחנו צריכים עוד משהו. לפעמים, אנחנו צריכים זרימות עבודה מרובות סוכנים! זה המקום שבו AutoGen נכנס לתמונה.

בואו ניקח דוגמה. דמיינו שאתם יוצרים אפליקציה לרישום הערות (ברור שלעולם אין מספיק מהם. 😝). אבל היי, אנחנו רוצים לעשות משהו מיוחד. אנחנו רוצים לקחת את ההערה הפשוטה והגולמית שמשתמש נותן לנו ולהפוך אותה למסמך מחודש במלואו, עם סיכום, כותרת אופציונלית ורשימת מטלות אוטומטית של משימות. ואנחנו רוצים לעשות את כל זה בלי להזיע - ובכן, לפחות עבור סוכני ה-AI שלכם.

אוקיי, עכשיו, אני יודע מה אתה חושב - "האם אלה לא כמו תוכניות טירונים?" בזה, אני אומר, אתה צודק. מתכוון... אבל נכון! אבל אל תלך שולל על ידי הפשטות של זרימת העבודה. הכישורים שתלמדו כאן - כמו טיפול בסוכני בינה מלאכותית, הטמעת בקרת זרימת עבודה וניהול היסטוריית שיחות - יעזרו לך לקחת את משחק הבינה המלאכותית שלך לשלב הבא.

אז תתכוננו, כי אנחנו הולכים ללמוד כיצד ליצור זרימות עבודה של AI באמצעות AutoGen!

לפני שנתחיל, שים לב שאתה יכול למצוא קישור לכל קוד המקור ב- GitHub .

בניית זרימות עבודה מותאמות אישית

נתחיל עם מקרה השימוש הראשון - "יצירת סיכום להערה ואחריו כותרת מותנית". למען ההגינות, אנחנו לא באמת צריכים להשתמש כאן בסוכנים. אבל היי, אנחנו חייבים להתחיל איפשהו נכון?

שלב 1: יצירת תצורת LLM הבסיסית שלנו

מסגרות סוכנות כמו AutoGen תמיד דורשות מאיתנו להגדיר את פרמטרי המודל. אנחנו מדברים על הדגם ומודל החזרה לשימוש, הטמפרטורה ואפילו הגדרות כמו פסק זמן ושמירה במטמון. במקרה של AutoGen, ההגדרה הזו נראית בערך כך:

 # build the gpt_configuration object base_llm_config = { "config_list": [ { "model": "Llama-3-8B-Instruct", "api_key": os.getenv("OPENAI_API_KEY"), "base_url": os.getenv("OPENAI_API_URL"), } ], "temperature": 0.0, "cache_seed": None, "timeout": 600, }

כפי שאתה יכול לראות, אני חובב AI ענק של קוד פתוח ונשבע ב-Llama 3. אתה יכול לגרום ל-AutoGen להצביע על כל שרת מסקנות תואם OpenAI פשוט על ידי הגדרת ערכי api_key ו- base_url . אז אל תהסס להשתמש ב- Groq, Together.ai, או אפילו vLLM כדי לארח את הדגם שלך באופן מקומי. אני משתמש ב- Inferix .

זה באמת כל כך קל!

אני סקרן! האם אתה מעוניין במדריך דומה לאירוח AI בקוד פתוח? ספרו לי בתגובות.

שלב 2: יצירת הסוכנים שלנו

אתחול סוכני שיחה ב-AutoGen הוא די פשוט; פשוט ספק את תצורת ה-LLM הבסיסית יחד עם הודעת מערכת ואתה מוכן ללכת.

 import autogen def get_note_summarizer(base_llm_config: dict): # A system message to define the role and job of our agent system_message = """You are a helpful AI assistant. The user will provide you a note. Generate a summary describing what the note is about. The summary must follow the provided "RULES". "RULES": - The summary should be not more than 3 short sentences. - Don't use bullet points. - The summary should be short and concise. - Identify and retain any "catchy" or memorable phrases from the original text - Identify and correct all grammatical errors. - Output the summary and nothing else.""" # Create and return our assistant agent return autogen.AssistantAgent( name="Note_Summarizer", # Lets give our agent a nice name llm_config=base_llm_config, # This is where we pass the llm configuration system_message=system_message, ) def get_title_generator(base_llm_config: dict): # A system message to define the role and job of our agent system_message = """You are a helpful AI assistant. The user will provide you a note along with a summary. Generate a title based on the user's input. The title must be witty and easy to read. The title should accurate present what the note is about. The title must strictly be less than 10 words. Make sure you keep the title short. Make sure you print the title and nothing else. """ # Create and return our assistant agent return autogen.AssistantAgent( name="Title_Generator", llm_config=base_llm_config, system_message=system_message, )

החלק החשוב ביותר ביצירת סוכנים הוא system_message . קחו רגע להסתכל על system_message שהשתמשתי בה כדי להגדיר את הסוכנים שלי.

חשוב לזכור שהדרך שבה סוכני AI ב-AutoGen עובדים היא על ידי השתתפות בשיחה . האופן שבו הם מפרשים ומעבירים את השיחה הלאה תלויה לחלוטין ב- system_message שאיתה הם מוגדרים. זה אחד המקומות שבהם תבזבזו קצת זמן כדי לעשות דברים כמו שצריך.

אנחנו צריכים רק עוד סוכן אחד. סוכן שיפעל כמיופה כוח עבורנו, בני האדם. סוכן שיכול ליזום את השיחה עם ה"פתק" בתור ההנחיה הראשונית שלו.

 def get_user(): # A system message to define the role and job of our agent system_message = "A human admin. Supplies the initial prompt and nothing else." # Create and return our user agent return autogen.UserProxyAgent( name="Admin", system_message=system_message, human_input_mode="NEVER", # We don't want interrupts for human-in-loop scenarios code_execution_config=False, # We definitely don't want AI executing code. default_auto_reply=None, )

לא קורה כאן שום דבר מפואר. רק שים לב שהגדרתי את הפרמטר default_auto_reply ל- None . זה חשוב. הגדרה זו לאף מבטיחה שהשיחה מסתיימת בכל פעם שנשלחת הודעה לסוכן המשתמש.

אופס, לגמרי שכחתי ליצור את הסוכנים האלה. בוא נעשה את זה ממש מהר.

 # Create our agents user = get_user() note_summarizer = get_note_summarizer(base_llm_config) title_generator = get_title_generator(base_llm_config)

שלב 3: הגדר תיאום סוכן באמצעות GroupChat

החלק האחרון בפאזל הוא לגרום לסוכנים שלנו לתאם. אנחנו צריכים לקבוע את רצף ההשתתפות שלהם ולהחליט אילו סוכנים צריכים.

אוקיי, זה יותר מחתיכה אחת. אבל הבנתם את הנקודה! 🙈

פתרון אפשרי אחד יהיה לתת לבינה מלאכותית להבין את הרצף שבו משתתפים הסוכנים. זה לא רעיון רע. למעשה, זו אפשרות הבחירה שלי כאשר אני מתמודד עם בעיות מורכבות שבהן אופי זרימת העבודה הוא דינמי.

עם זאת, לגישה זו יש חסרונות. המציאות מכה שוב! הסוכן שאחראי על קבלת ההחלטות הללו זקוק לרוב למודל גדול, וכתוצאה מכך לאחורים ועלויות גבוהות יותר. בנוסף, קיים סיכון שהוא עשוי לקבל החלטות שגויות.

עבור זרימות עבודה דטרמיניסטיות, שבהן אנו יודעים את רצף השלבים מבעוד מועד, אני אוהב לתפוס את המושכות ולנווט את הספינה בעצמי. למרבה המזל, AutoGen תומך במקרה השימוש הזה עם תכונה שימושית בשם GroupChat .

 from autogen import GroupChatManager from autogen.agentchat.groupchat import GroupChat from autogen.agentchat.agent import Agent def get_group_chat(agents, generate_title: bool = False): # Define the function which decides the agent selection order def speaker_selection_method(last_speaker: Agent, group_chat: GroupChat): # The admin will always forward the note to the summarizer if last_speaker.name == "Admin": return group_chat.agent_by_name("Note_Summarizer") # Forward the note to the title generator if the user wants a title if last_speaker.name == "Note_Summarizer" and generate_title: return group_chat.agent_by_name("Title_Generator") # Handle the default case - exit return None return GroupChat( agents=agents, messages=[], max_round=3, # There will only be 3 turns in this group chat. The group chat will exit automatically post that. speaker_selection_method=speaker_selection_method, )

תאר לעצמך GroupChat כקבוצת וואטסאפ שבה כל הסוכנים יכולים לשוחח ולשתף פעולה. הגדרה זו מאפשרת לסוכנים לבנות זה על עבודתו של זה. כיתת GroupChat יחד עם כיתת נלווית בשם GroupChatManager פועלת כמו מנהלי הקבוצה, ועוקבים אחר כל ההודעות שכל סוכן שולח כדי להבטיח שכולם יישארו בעניינים עם היסטוריית השיחה.

בקטע הקוד שלמעלה, יצרנו GroupChat עם שיטת speaker_selection_method מותאמת אישית. שיטת ה- speaker_selection_method מאפשרת לנו לציין את זרימת העבודה המותאמת אישית שלנו. הנה ייצוג חזותי של אותו הדבר.

מכיוון ששיטת ה- speaker_selection_method היא בעצם פונקציית Python, אנחנו יכולים לעשות איתה מה שאנחנו רוצים! זה עוזר לנו ליצור כמה זרימות עבודה עוצמתיות באמת. לדוגמה, נוכל:

• חברו סוכנים כדי לאמת את עבודתו של זה.
• יש לערב סוכן "פותר בעיות" אם אחד מהסוכנים הקודמים פלט שגיאה.
• הפעל webhooks למערכות חיצוניות כדי ליידע אותם על ההתקדמות שנעשתה.

תארו לעצמכם את האפשרויות! 😜

שלב 4: הגדרת הכל ותחילת השיחה

השלב האחרון הוא יצירת מופע של ה- GroupChat , כיסויו בתוך GroupChatManager ותחילת השיחה.

 # Create our group chat groupchat = get_group_chat([user, note_summarizer, title_generator], generate_title=True) manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=base_llm_config) # Start the chat user.initiate_chat( manager, clear_history=True, message=note, )

הערה: המשתמש משוחח בצ'אט עם GroupChatManager , לא עם הסוכנים הבודדים. אין לו מושג אילו סוכנים יצטרפו לשיחה כדי לספק את התשובה הסופית. ערמומי, נכון?

הפלט ייראה בערך כך:

 Admin (to chat_manager): Note: Convo with editor: - discuss titles and thumbnails - discuss video editing tips tracker - Zeeshan presents the tracker - what trick helps with what - he decidedls if we can experiment with something new - make sure all all videos since how config management works in k8s are backed u - make list of YouTube thumbnail templates - make list of YouTube idea generation limits -------------------------------------------------------------------------------- Next speaker: Note_Summarizer Note_Summarizer (to chat_manager): The note is about a conversation with an editor regarding video production. They discussed titles and thumbnails, as well as a video editing tips tracker presented by Zeeshan, which highlights tricks for specific tasks. Additionally, they ensured that all videos on Kubernetes configuration management are backed up and created lists of YouTube thumbnail templates and idea generation limits. -------------------------------------------------------------------------------- Next speaker: Title_Generator Title_Generator (to chat_manager): "Video Production Chat: Titles, Thumbnails, and Editing Tips" --------------------------------------------------------------------------------

השתלטות על השיחה

בשלב הבא, נצלול למקרה השימוש הסופי: לקחת "הערה" נתונה, לבנות אותה מחדש לבהירות טובה יותר, ולאחר מכן ליצור רשימת משימות עבור המשתמש.

הנה איך אנחנו הולכים לעשות את זה:

נתחיל בזיהוי רשימה של נושאים המכוסים בהערה. רשימה זו היא הכוח המניע מאחורי התהליך כולו. הוא קובע את הסעיפים עבור ההערה המעוצבת מחדש וקובע את רמת הפירוט עבור המשימות שנוצרו.

יש רק בעיה אחת קטנה. לסוכן Paraphrazer ו- Task_Creator לא ממש אכפת מהתפוקה של זה. אכפת להם רק מהפלט של Topic_Analyzer .

אז, אנחנו צריכים דרך למנוע מהתגובות של הסוכנים האלה לבלבל את היסטוריית השיחה, אחרת זה יהיה כאוס מוחלט. כבר השתלטנו על זרימת העבודה; עכשיו, הגיע הזמן להיות גם הבוס של היסטוריית השיחה! 😎

שלב 1: יצירת הסוכנים

דבר ראשון, אנחנו צריכים להקים את הסוכנים שלנו. אני לא הולך לשעמם אותך בפרטים אז הנה הקוד:

 def get_topic_analyzer(base_llm_config: dict): # A system message to define the role and job of our agent system_message = """You are a helpful AI assistant. The user will provide you a note. Generate a list of topics discussed in that note. The output must obey the following "RULES": "RULES": - Output should only contain the important topics from the note. - There must be atleast one topic in output. - Don't reuse the same text from user's note. - Don't have more than 10 topics in output.""" # Create and return our assistant agent return autogen.AssistantAgent( name="Topic_Analyzer", llm_config=base_llm_config, system_message=system_message, ) def get_paraphrazer(base_llm_config: dict): # A system message to define the role and job of our agent system_message = """You are a helpful AI content editor. The user will provide you a note along with a summary. Rewrite that note and make sure you cover everything in the note. Do not include the title. The output must obey the following "RULES": "RULES": - Output must be in markdown. - Make sure you use each points provided in summary as headers. - Each header must start with `##`. - Headers are not bullet points. - Each header can optionally have a list of bullet points. Don't put bullet points if the header has no content. - Strictly use "-" to start bullet points. - Optionally make an additional header named "Addional Info" to cover points not included in the summary. Use "Addional Info" header for unclassified points. - Identify and correct spelling & grammatical mistakes.""" # Create and return our assistant agent return autogen.AssistantAgent( name="Paraphrazer", llm_config=base_llm_config, system_message=system_message, ) def get_tasks_creator(base_llm_config: dict): # A system message to define the role and job of our agent system_message = """You are a helpful AI personal assistant. The user will provide you a note along with a summary. Identify each task the user has to do as next steps. Make sure to cover all the action items mentioned in the note. The output must obey the following "RULES": "RULES": - Output must be an YAML object with a field named tasks. - Make sure each task object contains fields title and description. - Extract the title based on the tasks the user has to do as next steps. - Description will be in markdown format. Feel free to include additional formatting and numbered lists. - Strictly use "-" or "dashes" to start bullet points in the description field. - Output empty tasks array if no tasks were found. - Identify and correct spelling & grammatical mistakes. - Identify and fix any errors in the YAML object. - Output should strictly be in YAML with no ``` or any additional text.""" # Create and return our assistant agent return autogen.AssistantAgent( name="Task_Creator", llm_config=base_llm_config, system_message=system_message, )

שלב 2: צור GroupChat מותאם אישית

לְמַרְבֶּה הַצַעַר. AutoGen אינו מאפשר לנו לשלוט ישירות בהיסטוריית השיחה. אז, אנחנו צריכים להמשיך ולהרחיב את שיעור GroupChat עם היישום המותאם אישית שלנו.

 class CustomGroupChat(GroupChat): def __init__(self, agents): super().__init__(agents, messages=[], max_round=4) # This function get's invoked whenever we want to append a message to the conversation history. def append(self, message: Dict, speaker: Agent): # We want to skip messages from the Paraphrazer and the Task_Creator if speaker.name != "Paraphrazer" and speaker.name != "Task_Creator": super().append(message, speaker) # The `speaker_selection_method` now becomes a function we will override from the base class def select_speaker(self, last_speaker: Agent, selector: AssistantAgent): if last_speaker.name == "Admin": return self.agent_by_name("Topic_Analyzer") if last_speaker.name == "Topic_Analyzer": return self.agent_by_name("Paraphrazer") if last_speaker.name == "Paraphrazer": return self.agent_by_name("Task_Creator") # Return the user agent by default return self.agent_by_name("Admin")

אנו עוקפים שתי פונקציות ממחלקת הבסיס GroupChat :

• append - זה שולט אילו הודעות יתווספו להיסטוריית השיחה.
• select_speaker - זוהי דרך נוספת לציין את ה- speaker_selection_method .

אבל רגע, כשצללתי עמוק יותר לתוך הקוד של AutoGen, הבנתי שה- GroupChatManager גורם לכל סוכן לשמור גם על היסטוריית השיחה. אל תשאל אותי למה. אני באמת לא יודע!

אז בואו נרחיב גם את GroupChatManager כדי לתקן את זה:

 class CustomGroupChatManager(GroupChatManager): def __init__(self, groupchat, llm_config): super().__init__(groupchat=groupchat, llm_config=llm_config) # Don't forget to register your reply functions self.register_reply(Agent, CustomGroupChatManager.run_chat, config=groupchat, reset_config=GroupChat.reset) def run_chat( self, messages: Optional[List[Dict]] = None, sender: Optional[Agent] = None, config: Optional[GroupChat] = None, ) -> Union[str, Dict, None]: """Run a group chat.""" if messages is None: messages = self._oai_messages[sender] message = messages[-1] speaker = sender groupchat = config for i in range(groupchat.max_round): # set the name to speaker's name if the role is not function if message["role"] != "function": message["name"] = speaker.name groupchat.append(message, speaker) if self._is_termination_msg(message): # The conversation is over break # We do not want each agent to maintain their own conversation history history # broadcast the message to all agents except the speaker # for agent in groupchat.agents: # if agent != speaker: # self.send(message, agent, request_reply=False, silent=True) # Pro Tip: Feel free to "send" messages to the user agent if you want to access the messages outside of autogen for agent in groupchat.agents: if agent.name == "Admin": self.send(message, agent, request_reply=False, silent=True) if i == groupchat.max_round - 1: # the last round break try: # select the next speaker speaker = groupchat.select_speaker(speaker, self) # let the speaker speak # We'll now have to pass their entire conversation of messages on generate_reply # Commented OG code: reply = speaker.generate_reply(sender=self) reply = speaker.generate_reply(sender=self, messages=groupchat.messages) except KeyboardInterrupt: # let the admin agent speak if interrupted if groupchat.admin_name in groupchat.agent_names: # admin agent is one of the participants speaker = groupchat.agent_by_name(groupchat.admin_name) # We'll now have to pass their entire conversation of messages on generate_reply # Commented OG code: reply = speaker.generate_reply(sender=self) reply = speaker.generate_reply(sender=self, messages=groupchat.messages) else: # admin agent is not found in the participants raise if reply is None: break # The speaker sends the message without requesting a reply speaker.send(reply, self, request_reply=False) message = self.last_message(speaker) return True, None

ביצעתי כמה עריכות קטנות ביישום המקורי. אתה אמור להיות מסוגל לעקוב אחר ההערות כדי לדעת יותר.

אבל יש דבר אחד שאני באמת רוצה להדגיש. אתה יכול לעקוף את שיטת ה-"run_chat" של GroupChatManager כדי להוסיף מנוע זרימת עבודה משלך כמו Apache Airflow או Temporal. העוסקים במערכות מבוזרות יודעים בדיוק עד כמה יכולת זו חזקה!

שלב 3: הגדרת הכל ותחילת השיחה

אנחנו מגדירים את כל זה כמו הדוגמה הקודמת וצופים בבייבי הזה מגרגר! 🐱

 # Create our agents user = get_user() topic_analyzer = get_topic_analyzer(base_llm_config) paraphrazer = get_paraphrazer(base_llm_config) task_creator = get_tasks_creator(base_llm_config) # Create our group chat groupchat = CustomGroupChat(agents=[user, topic_analyzer, paraphrazer, task_creator]) manager = CustomGroupChatManager(groupchat=groupchat, llm_config=base_llm_config) # Start the chat user.initiate_chat( manager, clear_history=True, message=note, ) # Lets print the count of tasks just for fun chat_messages = user.chat_messages.get(manager) if chat_messages is not None: for message in chat_messages: if message.get("name") == "Task_Creator": taskList = yaml.safe_load(message.get("content")) # type: ignore l = len(taskList.get("tasks")) print(f"Got {l} tasks from Task_Creator.")

הפלט ייראה בערך כך:

 Admin (to chat_manager): Note: Convo with editor: - discuss titles and thumbnails - discuss video editing tips tracker - Zeeshan presents the tracker - what trick helps with what - he decidedls if we can experiment with something new - make sure all all videos since how config management works in k8s are backed u - make list of YouTube thumbnail templates - make list of YouTube idea generation limits -------------------------------------------------------------------------------- Topic_Analyzer (to chat_manager): Here is the list of topics discussed in the note: 1. Titles 2. Thumbnails 3. Video editing tips 4. Config management in Kubernetes (k8s) 5. YouTube thumbnail templates 6. YouTube idea generation limits -------------------------------------------------------------------------------- Paraphrazer (to chat_manager): Here is the rewritten note in markdown format: ## Titles - Discuss titles and thumbnails with the editor ## Video Editing Tips Tracker ### Zeeshan presents the tracker - What trick helps with what - He decides if we can experiment with something new ## Config Management in Kubernetes (k8s) - Make sure all videos since how config management works in k8s are backed up ## YouTube Thumbnail Templates - Make a list of YouTube thumbnail templates ## YouTube Idea Generation Limits - Make a list of YouTube idea generation limits ## Additional Info - Discuss video editing tips tracker with Zeeshan - Present the tracker and decide if we can experiment with something new -------------------------------------------------------------------------------- Task_Creator (to chat_manager): tasks: - title: Discuss Titles and Thumbnails description: >- - Discuss titles and thumbnails with the editor This task involves having a conversation with the editor to discuss the titles and thumbnails for the videos. - title: Discuss Video Editing Tips Tracker description: >- - Zeeshan presents the tracker - Discuss what trick helps with what - Decide if we can experiment with something new This task involves discussing the video editing tips tracker presented by Zeeshan, understanding what tricks help with what, and deciding if it's possible to experiment with something new. - title: Back up All Videos Since How Config Management Works in k8s description: >- - Make sure all videos since how config management works in k8s are backed up This task involves ensuring that all videos related to config management in Kubernetes (k8s) are backed up. - title: Create List of YouTube Thumbnail Templates description: >- - Make list of YouTube thumbnail templates This task involves creating a list of YouTube thumbnail templates. - title: Create List of YouTube Idea Generation Limits description: >- - Make list of YouTube idea generation limits This task involves creating a list of YouTube idea generation limits. -------------------------------------------------------------------------------- Got 5 tasks from Task_Creator.

כן. ברוכים הבאים לעידן הבינה המלאכותית שנותן לנו להומנים עבודה לעשות! (איפה הכל השתבש? 🤷‍♂️)

מַסְקָנָה

קשה לבנות אפליקציות מונעות בינה מלאכותית. אבל אפשר לעשות את זה עם הכלים הנכונים. לסיכום:

• סוכני בינה מלאכותית פותחים את היכולת המדהימה הזו של עיצוב בעיות מורכבות כשיחות.
• אם נעשה נכון, זה יכול לעזור להפוך את יישומי הבינה המלאכותית שלנו לדטרמיניסטים ואמינים יותר.
• כלים כמו AutoGen מספקים לנו מסגרת עם הפשטות פשוטות לבניית סוכני AI.

כשלבים הבאים, תוכל לבדוק את המשאבים הבאים כדי לצלול עמוק יותר לתוך עולם סוכני הבינה המלאכותית:

• ריפו של GitHub כדי לנסות את זה בעצמך (קוד)
• יישום דפוסי שיחה באמצעות AutoGen (Docs)