Reflexion#
Reflexion ist ein Entwurfsmuster, bei dem einer LLM-Generierung eine Reflexion folgt, die selbst eine weitere LLM-Generierung ist, die auf der Ausgabe der ersten basiert. Zum Beispiel, wenn die Aufgabe darin besteht, Code zu schreiben, kann die erste LLM einen Code-Schnipsel generieren, und die zweite LLM kann eine Kritik des Code-Schnipsels generieren.
Im Kontext von AutoGen und Agenten kann Reflexion als ein Paar von Agenten implementiert werden, wobei der erste Agent eine Nachricht generiert und der zweite Agent eine Antwort auf die Nachricht generiert. Die beiden Agenten interagieren weiter, bis sie eine Abbruchbedingung erreichen, wie z. B. eine maximale Anzahl von Iterationen oder eine Genehmigung durch den zweiten Agenten.
Lassen Sie uns ein einfaches Reflexionsentwurfsmuster mit AutoGen-Agenten implementieren. Es wird zwei Agenten geben: einen Coder-Agenten und einen Reviewer-Agenten. Der Coder-Agent wird einen Code-Schnipsel generieren, und der Reviewer-Agent wird eine Kritik des Code-Schnipsels generieren.
Nachrichtenprotokoll#
Bevor wir die Agenten definieren, müssen wir zuerst das Nachrichtenprotokoll für die Agenten definieren.
from dataclasses import dataclass
@dataclass
class CodeWritingTask:
task: str
@dataclass
class CodeWritingResult:
task: str
code: str
review: str
@dataclass
class CodeReviewTask:
session_id: str
code_writing_task: str
code_writing_scratchpad: str
code: str
@dataclass
class CodeReviewResult:
review: str
session_id: str
approved: bool
Die obige Nachrichtensammlung definiert das Protokoll für unser Beispiel-Reflexionsentwurfsmuster.
Die Anwendung sendet eine
CodeWritingTask-Nachricht an den Coder-Agenten.Der Coder-Agent generiert eine
CodeReviewTask-Nachricht, die an den Reviewer-Agenten gesendet wird.Der Reviewer-Agent generiert eine
CodeReviewResult-Nachricht, die an den Coder-Agenten zurückgesendet wird.Abhängig von der
CodeReviewResult-Nachricht sendet der Coder-Agent, wenn der Code genehmigt wird, eineCodeWritingResult-Nachricht an die Anwendung zurück. Andernfalls sendet der Coder-Agent eine weitereCodeReviewTask-Nachricht an den Reviewer-Agenten, und der Prozess wird fortgesetzt.
Wir können das Nachrichtenprotokoll mit einem Datenflussdiagramm visualisieren.
Agenten#
Nun definieren wir die Agenten für das Reflexionsentwurfsmuster.
import json
import re
import uuid
from typing import Dict, List, Union
from autogen_core import MessageContext, RoutedAgent, TopicId, default_subscription, message_handler
from autogen_core.models import (
AssistantMessage,
ChatCompletionClient,
LLMMessage,
SystemMessage,
UserMessage,
)
Wir verwenden die Broadcast-API, um das Entwurfsmuster zu implementieren. Die Agenten implementieren das Pub/Sub-Modell. Der Coder-Agent abonniert die Nachrichten CodeWritingTask und CodeReviewResult und veröffentlicht die Nachrichten CodeReviewTask und CodeWritingResult.
@default_subscription
class CoderAgent(RoutedAgent):
"""An agent that performs code writing tasks."""
def __init__(self, model_client: ChatCompletionClient) -> None:
super().__init__("A code writing agent.")
self._system_messages: List[LLMMessage] = [
SystemMessage(
content="""You are a proficient coder. You write code to solve problems.
Work with the reviewer to improve your code.
Always put all finished code in a single Markdown code block.
For example:
```python
def hello_world():
print("Hello, World!")
```
Respond using the following format:
Thoughts: <Your comments>
Code: <Your code>
""",
)
]
self._model_client = model_client
self._session_memory: Dict[str, List[CodeWritingTask | CodeReviewTask | CodeReviewResult]] = {}
@message_handler
async def handle_code_writing_task(self, message: CodeWritingTask, ctx: MessageContext) -> None:
# Store the messages in a temporary memory for this request only.
session_id = str(uuid.uuid4())
self._session_memory.setdefault(session_id, []).append(message)
# Generate a response using the chat completion API.
response = await self._model_client.create(
self._system_messages + [UserMessage(content=message.task, source=self.metadata["type"])],
cancellation_token=ctx.cancellation_token,
)
assert isinstance(response.content, str)
# Extract the code block from the response.
code_block = self._extract_code_block(response.content)
if code_block is None:
raise ValueError("Code block not found.")
# Create a code review task.
code_review_task = CodeReviewTask(
session_id=session_id,
code_writing_task=message.task,
code_writing_scratchpad=response.content,
code=code_block,
)
# Store the code review task in the session memory.
self._session_memory[session_id].append(code_review_task)
# Publish a code review task.
await self.publish_message(code_review_task, topic_id=TopicId("default", self.id.key))
@message_handler
async def handle_code_review_result(self, message: CodeReviewResult, ctx: MessageContext) -> None:
# Store the review result in the session memory.
self._session_memory[message.session_id].append(message)
# Obtain the request from previous messages.
review_request = next(
m for m in reversed(self._session_memory[message.session_id]) if isinstance(m, CodeReviewTask)
)
assert review_request is not None
# Check if the code is approved.
if message.approved:
# Publish the code writing result.
await self.publish_message(
CodeWritingResult(
code=review_request.code,
task=review_request.code_writing_task,
review=message.review,
),
topic_id=TopicId("default", self.id.key),
)
print("Code Writing Result:")
print("-" * 80)
print(f"Task:\n{review_request.code_writing_task}")
print("-" * 80)
print(f"Code:\n{review_request.code}")
print("-" * 80)
print(f"Review:\n{message.review}")
print("-" * 80)
else:
# Create a list of LLM messages to send to the model.
messages: List[LLMMessage] = [*self._system_messages]
for m in self._session_memory[message.session_id]:
if isinstance(m, CodeReviewResult):
messages.append(UserMessage(content=m.review, source="Reviewer"))
elif isinstance(m, CodeReviewTask):
messages.append(AssistantMessage(content=m.code_writing_scratchpad, source="Coder"))
elif isinstance(m, CodeWritingTask):
messages.append(UserMessage(content=m.task, source="User"))
else:
raise ValueError(f"Unexpected message type: {m}")
# Generate a revision using the chat completion API.
response = await self._model_client.create(messages, cancellation_token=ctx.cancellation_token)
assert isinstance(response.content, str)
# Extract the code block from the response.
code_block = self._extract_code_block(response.content)
if code_block is None:
raise ValueError("Code block not found.")
# Create a new code review task.
code_review_task = CodeReviewTask(
session_id=message.session_id,
code_writing_task=review_request.code_writing_task,
code_writing_scratchpad=response.content,
code=code_block,
)
# Store the code review task in the session memory.
self._session_memory[message.session_id].append(code_review_task)
# Publish a new code review task.
await self.publish_message(code_review_task, topic_id=TopicId("default", self.id.key))
def _extract_code_block(self, markdown_text: str) -> Union[str, None]:
pattern = r"```(\w+)\n(.*?)\n```"
# Search for the pattern in the markdown text
match = re.search(pattern, markdown_text, re.DOTALL)
# Extract the language and code block if a match is found
if match:
return match.group(2)
return None
Ein paar Dinge, die man über CoderAgent beachten sollte:
Er verwendet Chain-of-Thought-Prompting in seiner Systemnachricht.
Er speichert Nachrichtenverläufe für verschiedene
CodeWritingTaskin einem Wörterbuch, sodass jede Aufgabe ihren eigenen Verlauf hat.Wenn eine LLM-Inferenzanfrage über seinen Modellclient durchgeführt wird, transformiert er den Nachrichtenverlauf in eine Liste von
autogen_core.models.LLMMessage-Objekten, die an den Modellclient übergeben werden.
Der Reviewer-Agent abonniert die CodeReviewTask-Nachricht und veröffentlicht die CodeReviewResult-Nachricht.
@default_subscription
class ReviewerAgent(RoutedAgent):
"""An agent that performs code review tasks."""
def __init__(self, model_client: ChatCompletionClient) -> None:
super().__init__("A code reviewer agent.")
self._system_messages: List[LLMMessage] = [
SystemMessage(
content="""You are a code reviewer. You focus on correctness, efficiency and safety of the code.
Respond using the following JSON format:
{
"correctness": "<Your comments>",
"efficiency": "<Your comments>",
"safety": "<Your comments>",
"approval": "<APPROVE or REVISE>",
"suggested_changes": "<Your comments>"
}
""",
)
]
self._session_memory: Dict[str, List[CodeReviewTask | CodeReviewResult]] = {}
self._model_client = model_client
@message_handler
async def handle_code_review_task(self, message: CodeReviewTask, ctx: MessageContext) -> None:
# Format the prompt for the code review.
# Gather the previous feedback if available.
previous_feedback = ""
if message.session_id in self._session_memory:
previous_review = next(
(m for m in reversed(self._session_memory[message.session_id]) if isinstance(m, CodeReviewResult)),
None,
)
if previous_review is not None:
previous_feedback = previous_review.review
# Store the messages in a temporary memory for this request only.
self._session_memory.setdefault(message.session_id, []).append(message)
prompt = f"""The problem statement is: {message.code_writing_task}
The code is:
```
{message.code}
```
Previous feedback:
{previous_feedback}
Please review the code. If previous feedback was provided, see if it was addressed.
"""
# Generate a response using the chat completion API.
response = await self._model_client.create(
self._system_messages + [UserMessage(content=prompt, source=self.metadata["type"])],
cancellation_token=ctx.cancellation_token,
json_output=True,
)
assert isinstance(response.content, str)
# TODO: use structured generation library e.g. guidance to ensure the response is in the expected format.
# Parse the response JSON.
review = json.loads(response.content)
# Construct the review text.
review_text = "Code review:\n" + "\n".join([f"{k}: {v}" for k, v in review.items()])
approved = review["approval"].lower().strip() == "approve"
result = CodeReviewResult(
review=review_text,
session_id=message.session_id,
approved=approved,
)
# Store the review result in the session memory.
self._session_memory[message.session_id].append(result)
# Publish the review result.
await self.publish_message(result, topic_id=TopicId("default", self.id.key))
Der ReviewerAgent verwendet den JSON-Modus bei der Durchführung einer LLM-Inferenzanfrage und verwendet ebenfalls Chain-of-Thought-Prompting in seiner Systemnachricht.
Protokollierung#
Schalten Sie die Protokollierung ein, um die zwischen den Agenten ausgetauschten Nachrichten zu sehen.
import logging
logging.basicConfig(level=logging.WARNING)
logging.getLogger("autogen_core").setLevel(logging.DEBUG)
Ausführung des Entwurfsmusters#
Testen wir das Entwurfsmuster mit einer Coding-Aufgabe. Da alle Agenten mit dem Klassen-Decorator default_subscription() dekoriert sind, abonnieren die Agenten bei ihrer Erstellung automatisch das Standard-Thema. Wir veröffentlichen eine CodeWritingTask-Nachricht an das Standard-Thema, um den Reflexionsprozess zu starten.
from autogen_core import DefaultTopicId, SingleThreadedAgentRuntime
from autogen_ext.models.openai import OpenAIChatCompletionClient
runtime = SingleThreadedAgentRuntime()
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
await ReviewerAgent.register(runtime, "ReviewerAgent", lambda: ReviewerAgent(model_client=model_client))
await CoderAgent.register(runtime, "CoderAgent", lambda: CoderAgent(model_client=model_client))
runtime.start()
await runtime.publish_message(
message=CodeWritingTask(task="Write a function to find the sum of all even numbers in a list."),
topic_id=DefaultTopicId(),
)
# Keep processing messages until idle.
await runtime.stop_when_idle()
# Close the model client.
await model_client.close()
INFO:autogen_core:Publishing message of type CodeWritingTask to all subscribers: {'task': 'Write a function to find the sum of all even numbers in a list.'}
INFO:autogen_core:Calling message handler for ReviewerAgent with message type CodeWritingTask published by Unknown
INFO:autogen_core:Calling message handler for CoderAgent with message type CodeWritingTask published by Unknown
INFO:autogen_core:Unhandled message: CodeWritingTask(task='Write a function to find the sum of all even numbers in a list.')
INFO:autogen_core.events:{"prompt_tokens": 101, "completion_tokens": 88, "type": "LLMCall"}
INFO:autogen_core:Publishing message of type CodeReviewTask to all subscribers: {'session_id': '51db93d5-3e29-4b7f-9f96-77be7bb02a5e', 'code_writing_task': 'Write a function to find the sum of all even numbers in a list.', 'code_writing_scratchpad': 'Thoughts: To find the sum of all even numbers in a list, we can use a list comprehension to filter out the even numbers and then use the `sum()` function to calculate their total. The implementation should handle edge cases like an empty list or a list with no even numbers.\n\nCode:\n```python\ndef sum_of_even_numbers(numbers):\n return sum(num for num in numbers if num % 2 == 0)\n```', 'code': 'def sum_of_even_numbers(numbers):\n return sum(num for num in numbers if num % 2 == 0)'}
INFO:autogen_core:Calling message handler for ReviewerAgent with message type CodeReviewTask published by CoderAgent:default
INFO:autogen_core.events:{"prompt_tokens": 163, "completion_tokens": 235, "type": "LLMCall"}
INFO:autogen_core:Publishing message of type CodeReviewResult to all subscribers: {'review': "Code review:\ncorrectness: The function correctly identifies and sums all even numbers in the provided list. The use of a generator expression ensures that only even numbers are processed, which is correct.\nefficiency: The function is efficient as it utilizes a generator expression that avoids creating an intermediate list, therefore using less memory. The time complexity is O(n) where n is the number of elements in the input list, which is optimal for this task.\nsafety: The function does not include checks for input types. If a non-iterable or a list containing non-integer types is passed, it could lead to unexpected behavior or errors. It’s advisable to handle such cases.\napproval: REVISE\nsuggested_changes: Consider adding input validation to ensure that 'numbers' is a list and contains only integers. You could raise a ValueError if the input is invalid. Example: 'if not isinstance(numbers, list) or not all(isinstance(num, int) for num in numbers): raise ValueError('Input must be a list of integers')'. This will make the function more robust.", 'session_id': '51db93d5-3e29-4b7f-9f96-77be7bb02a5e', 'approved': False}
INFO:autogen_core:Calling message handler for CoderAgent with message type CodeReviewResult published by ReviewerAgent:default
INFO:autogen_core.events:{"prompt_tokens": 421, "completion_tokens": 119, "type": "LLMCall"}
INFO:autogen_core:Publishing message of type CodeReviewTask to all subscribers: {'session_id': '51db93d5-3e29-4b7f-9f96-77be7bb02a5e', 'code_writing_task': 'Write a function to find the sum of all even numbers in a list.', 'code_writing_scratchpad': "Thoughts: I appreciate the reviewer's feedback on input validation. Adding type checks ensures that the function can handle unexpected inputs gracefully. I will implement the suggested changes and include checks for both the input type and the elements within the list to confirm that they are integers.\n\nCode:\n```python\ndef sum_of_even_numbers(numbers):\n if not isinstance(numbers, list) or not all(isinstance(num, int) for num in numbers):\n raise ValueError('Input must be a list of integers')\n \n return sum(num for num in numbers if num % 2 == 0)\n```", 'code': "def sum_of_even_numbers(numbers):\n if not isinstance(numbers, list) or not all(isinstance(num, int) for num in numbers):\n raise ValueError('Input must be a list of integers')\n \n return sum(num for num in numbers if num % 2 == 0)"}
INFO:autogen_core:Calling message handler for ReviewerAgent with message type CodeReviewTask published by CoderAgent:default
INFO:autogen_core.events:{"prompt_tokens": 420, "completion_tokens": 153, "type": "LLMCall"}
INFO:autogen_core:Publishing message of type CodeReviewResult to all subscribers: {'review': 'Code review:\ncorrectness: The function correctly sums all even numbers in the provided list. It raises a ValueError if the input is not a list of integers, which is a necessary check for correctness.\nefficiency: The function remains efficient with a time complexity of O(n) due to the use of a generator expression. There are no unnecessary intermediate lists created, so memory usage is optimal.\nsafety: The function includes input validation, which enhances safety by preventing incorrect input types. It raises a ValueError for invalid inputs, making the function more robust against unexpected data.\napproval: APPROVE\nsuggested_changes: No further changes are necessary as the previous feedback has been adequately addressed.', 'session_id': '51db93d5-3e29-4b7f-9f96-77be7bb02a5e', 'approved': True}
INFO:autogen_core:Calling message handler for CoderAgent with message type CodeReviewResult published by ReviewerAgent:default
INFO:autogen_core:Publishing message of type CodeWritingResult to all subscribers: {'task': 'Write a function to find the sum of all even numbers in a list.', 'code': "def sum_of_even_numbers(numbers):\n if not isinstance(numbers, list) or not all(isinstance(num, int) for num in numbers):\n raise ValueError('Input must be a list of integers')\n \n return sum(num for num in numbers if num % 2 == 0)", 'review': 'Code review:\ncorrectness: The function correctly sums all even numbers in the provided list. It raises a ValueError if the input is not a list of integers, which is a necessary check for correctness.\nefficiency: The function remains efficient with a time complexity of O(n) due to the use of a generator expression. There are no unnecessary intermediate lists created, so memory usage is optimal.\nsafety: The function includes input validation, which enhances safety by preventing incorrect input types. It raises a ValueError for invalid inputs, making the function more robust against unexpected data.\napproval: APPROVE\nsuggested_changes: No further changes are necessary as the previous feedback has been adequately addressed.'}
INFO:autogen_core:Calling message handler for ReviewerAgent with message type CodeWritingResult published by CoderAgent:default
INFO:autogen_core:Unhandled message: CodeWritingResult(task='Write a function to find the sum of all even numbers in a list.', code="def sum_of_even_numbers(numbers):\n if not isinstance(numbers, list) or not all(isinstance(num, int) for num in numbers):\n raise ValueError('Input must be a list of integers')\n \n return sum(num for num in numbers if num % 2 == 0)", review='Code review:\ncorrectness: The function correctly sums all even numbers in the provided list. It raises a ValueError if the input is not a list of integers, which is a necessary check for correctness.\nefficiency: The function remains efficient with a time complexity of O(n) due to the use of a generator expression. There are no unnecessary intermediate lists created, so memory usage is optimal.\nsafety: The function includes input validation, which enhances safety by preventing incorrect input types. It raises a ValueError for invalid inputs, making the function more robust against unexpected data.\napproval: APPROVE\nsuggested_changes: No further changes are necessary as the previous feedback has been adequately addressed.')
Code Writing Result:
--------------------------------------------------------------------------------
Task:
Write a function to find the sum of all even numbers in a list.
--------------------------------------------------------------------------------
Code:
def sum_of_even_numbers(numbers):
if not isinstance(numbers, list) or not all(isinstance(num, int) for num in numbers):
raise ValueError('Input must be a list of integers')
return sum(num for num in numbers if num % 2 == 0)
--------------------------------------------------------------------------------
Review:
Code review:
correctness: The function correctly sums all even numbers in the provided list. It raises a ValueError if the input is not a list of integers, which is a necessary check for correctness.
efficiency: The function remains efficient with a time complexity of O(n) due to the use of a generator expression. There are no unnecessary intermediate lists created, so memory usage is optimal.
safety: The function includes input validation, which enhances safety by preventing incorrect input types. It raises a ValueError for invalid inputs, making the function more robust against unexpected data.
approval: APPROVE
suggested_changes: No further changes are necessary as the previous feedback has been adequately addressed.
--------------------------------------------------------------------------------
Die Log-Nachrichten zeigen die Interaktion zwischen dem Coder- und dem Reviewer-Agenten. Die endgültige Ausgabe zeigt den vom Coder-Agenten generierten Code-Schnipsel und die vom Reviewer-Agenten generierte Kritik.