github/autogen
1
1
Fork 0
mirror of https://github.com/microsoft/autogen.git synced 2026-04-20 03:02:16 -04:00
Code Issues Packages Projects Releases Wiki Activity

Textual Example; Tool Approver (#81)

Browse Source 
* Update example to use textual app. Add tool approver setting to chat completion agent.

* dep

* fix
This commit is contained in:
Eric Zhu
2024-06-15 18:04:24 -07:00
committed by GitHub
parent 8dad8b0536
commit 40701a5a00
6 changed files with 230 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
examples/software_consultancy.py
View File

@@ -10,46 +10,34 @@ import argparse
import asyncio
import base64
import logging
import os
import sys
sys.path.append(os.path.abspath(os.path.dirname(__file__)))
import aiofiles
import aiohttp
import openai
from agnext.application import SingleThreadedAgentRuntime
from agnext.chat.agents import ChatCompletionAgent, UserProxyAgent
from agnext.chat.agents import ChatCompletionAgent
from agnext.chat.memory import HeadAndTailChatMemory
from agnext.chat.patterns.group_chat_manager import GroupChatManager
from agnext.chat.types import PublishNow
from agnext.components.models import OpenAI, SystemMessage
from agnext.components.tools import FunctionTool
from agnext.core import AgentRuntime
from agnext.core import Agent, AgentRuntime
from markdownify import markdownify # type: ignore
from tqdm import tqdm
from typing_extensions import Annotated
sep = "+----------------------------------------------------------+"
async def get_user_input(prompt: str) -> Annotated[str, "The user input."]:
return await asyncio.get_event_loop().run_in_executor(None, input, prompt)
async def confirm(message: str) -> None:
user_input = await get_user_input(f"{message} (yes/no): ")
if user_input.lower() not in ["yes", "y"]:
raise ValueError(f"Operation cancelled: reason: {user_input}")
from utils import TextualChatApp, TextualUserAgent, start_runtime
async def write_file(filename: str, content: str) -> str:
# Ask for confirmation first.
await confirm(f"Are you sure you want to write to {filename}?")
async with aiofiles.open(filename, "w") as file:
await file.write(content)
return f"Content written to {filename}."
async def execute_command(command: str) -> Annotated[str, "The standard output and error of the executed command."]:
# Ask for confirmation first.
await confirm(f"Are you sure you want to execute {command}?")
process = await asyncio.subprocess.create_subprocess_shell(
command,
stdout=asyncio.subprocess.PIPE,
@@ -60,15 +48,11 @@ async def execute_command(command: str) -> Annotated[str, "The standard output a
async def read_file(filename: str) -> Annotated[str, "The content of the file."]:
# Ask for confirmation first.
await confirm(f"Are you sure you want to read {filename}?")
async with aiofiles.open(filename, "r") as file:
return await file.read()
async def remove_file(filename: str) -> str:
# Ask for confirmation first.
await confirm(f"Are you sure you want to remove {filename}?")
process = await asyncio.subprocess.create_subprocess_exec("rm", filename)
await process.wait()
if process.returncode != 0:
@@ -92,8 +76,6 @@ async def list_files(directory: str) -> Annotated[str, "The list of files in the
async def browse_web(url: str) -> Annotated[str, "The content of the web page in Markdown format."]:
# Ask for confirmation first.
await confirm(f"Are you sure you want to browse {url}?")
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
html = await response.text()
@@ -107,8 +89,6 @@ async def create_image(
description: Annotated[str, "Describe the image to create"],
filename: Annotated[str, "The path to save the created image"],
) -> str:
# Ask for confirmation first.
await confirm(f"Are you sure you want to create an image with description: {description}?")
# Use Dalle to generate an image from the description.
with tqdm(desc="Generating image...", leave=False) as pbar:
client = openai.AsyncClient()
@@ -122,7 +102,7 @@ async def create_image(
return f"Image created and saved to {filename}."
def software_consultancy(runtime: AgentRuntime) -> UserProxyAgent: # type: ignore
def software_consultancy(runtime: AgentRuntime, user_agent: Agent) -> None: # type: ignore
developer = ChatCompletionAgent(
name="Developer",
description="A Python software developer.",
@@ -159,6 +139,7 @@ def software_consultancy(runtime: AgentRuntime) -> UserProxyAgent: # type: igno
FunctionTool(list_files, name="list_files", description="List files in a directory."),
FunctionTool(browse_web, name="browse_web", description="Browse a web page."),
],
tool_approver=user_agent,
)
product_manager = ChatCompletionAgent(
name="ProductManager",
@@ -186,6 +167,7 @@ def software_consultancy(runtime: AgentRuntime) -> UserProxyAgent: # type: igno
FunctionTool(list_files, name="list_files", description="List files in a directory."),
FunctionTool(browse_web, name="browse_web", description="Browse a web page."),
],
tool_approver=user_agent,
)
ux_designer = ChatCompletionAgent(
name="UserExperienceDesigner",
@@ -216,6 +198,7 @@ def software_consultancy(runtime: AgentRuntime) -> UserProxyAgent: # type: igno
),
FunctionTool(list_files, name="list_files", description="List files in a directory."),
],
tool_approver=user_agent,
)
illustrator = ChatCompletionAgent(
name="Illustrator",
@@ -239,38 +222,19 @@ def software_consultancy(runtime: AgentRuntime) -> UserProxyAgent: # type: igno
description="Create an image from a description.",
),
],
)
customer = UserProxyAgent(
name="Customer",
description="A customer requesting for help.",
runtime=runtime,
user_input_prompt=f"{sep}\nYou:\n",
tool_approver=user_agent,
)
_ = GroupChatManager(
name="GroupChatManager",
description="A group chat manager.",
runtime=runtime,
memory=HeadAndTailChatMemory(head_size=1, tail_size=10),
model_client=OpenAI(model="gpt-4-turbo"),
participants=[developer, product_manager, ux_designer, illustrator, customer],
on_message_received=lambda message: print(f"{sep}\n{message.source}: {message.content}"),
# model_client=OpenAI(model="gpt-4-turbo"),
participants=[developer, product_manager, ux_designer, illustrator, user_agent],
)
return customer
async def main() -> None:
runtime = SingleThreadedAgentRuntime()
user_proxy = software_consultancy(runtime)
# Request the user to start the conversation.
runtime.send_message(PublishNow(), user_proxy)
while True:
# TODO: Add a way to stop the loop.
await runtime.process_next()
await asyncio.sleep(1)
if __name__ == "__main__":
description = "Work with a software development consultancy to create your own Python application."
art = r"""
+----------------------------------------------------------+
| ____ __ _ |
@@ -292,12 +256,27 @@ if __name__ == "__main__":
| the team! |
+----------------------------------------------------------+
"""
runtime = SingleThreadedAgentRuntime()
app = TextualChatApp(runtime, welcoming_notice=art, user_name="You")
user_agent = TextualUserAgent(
name="Customer",
description="A customer looking for help.",
runtime=runtime,
app=app,
)
software_consultancy(runtime, user_agent)
# Start the runtime.
asyncio.create_task(start_runtime(runtime))
# Start the app.
await app.run_async()
if __name__ == "__main__":
description = "Work with a software development consultancy to create your own Python application."
parser = argparse.ArgumentParser(description="Software consultancy demo.")
parser.add_argument("--verbose", action="store_true", help="Enable verbose logging.")
args = parser.parse_args()
if args.verbose:
logging.basicConfig(level=logging.WARNING)
logging.getLogger("agnext").setLevel(logging.DEBUG)
print(art)
asyncio.run(main())

150
examples/utils.py Normal file
View File

@@ -0,0 +1,150 @@
import asyncio
from asyncio import Future
from agnext.application import SingleThreadedAgentRuntime
from agnext.chat.types import PublishNow, RespondNow, TextMessage, ToolApprovalRequest, ToolApprovalResponse
from agnext.components import TypeRoutedAgent, message_handler
from agnext.core import AgentRuntime, CancellationToken
from textual.app import App, ComposeResult
from textual.containers import ScrollableContainer
from textual.widgets import Button, Footer, Header, Input, Markdown, Static
class UserMessage(Markdown):
def on_mount(self) -> None:
self.styles.margin = 1
self.styles.padding = 1
self.styles.border = ("solid", "green")
class AssistantMessage(Markdown):
def on_mount(self) -> None:
self.styles.margin = 1
self.styles.padding = 1
self.styles.border = ("solid", "blue")
class WelcomeMessage(Static):
def on_mount(self) -> None:
self.styles.margin = 1
self.styles.padding = 1
self.styles.border = ("solid", "blue")
class ChatInput(Input):
def on_mount(self) -> None:
self.focus()
def on_input_submitted(self, event: Input.Submitted) -> None:
self.clear()
class ToolApprovalRequestNotice(Static):
def __init__(self, request: ToolApprovalRequest, response_future: Future[ToolApprovalResponse]) -> None: # type: ignore
self._tool_call = request.tool_call
self._future = response_future
super().__init__()
def compose(self) -> ComposeResult:
yield Static(f"Tool call: {self._tool_call.name}, arguments: {self._tool_call.arguments[:50]}")
yield Button("Approve", id="approve", variant="warning")
yield Button("Deny", id="deny", variant="default")
def on_mount(self) -> None:
self.styles.margin = 1
self.styles.padding = 1
self.styles.border = ("solid", "red")
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id
assert button_id is not None
if button_id == "approve":
self._future.set_result(ToolApprovalResponse(tool_call_id=self._tool_call.id, approved=True, reason=""))
else:
self._future.set_result(ToolApprovalResponse(tool_call_id=self._tool_call.id, approved=False, reason=""))
self.remove()
class TextualChatApp(App): # type: ignore
"""A Textual app for a chat interface."""
def __init__(self, runtime: AgentRuntime, welcoming_notice: str, user_name: str) -> None: # type: ignore
self._runtime = runtime
self._welcoming_notice = welcoming_notice
self._user_name = user_name
super().__init__()
def compose(self) -> ComposeResult:
yield Header()
yield Footer()
yield ScrollableContainer(id="chat-messages")
yield ChatInput()
def on_mount(self) -> None:
chat_messages = self.query_one("#chat-messages")
notice = WelcomeMessage(self._welcoming_notice, id="welcome")
chat_messages.mount(notice)
async def on_input_submitted(self, event: Input.Submitted) -> None:
user_input = event.value
await self.publish_user_message(user_input)
async def post_request_user_input_notice(self) -> None:
chat_messages = self.query_one("#chat-messages")
notice = Static("Please enter your input.", id="typing")
chat_messages.mount(notice)
notice.scroll_visible()
async def publish_user_message(self, user_input: str) -> None:
chat_messages = self.query_one("#chat-messages")
# Remove all typing messages.
chat_messages.query("#typing").remove()
# Publish the user message to the runtime.
await self._runtime.publish_message(TextMessage(source=self._user_name, content=user_input))
async def post_runtime_message(self, message: TextMessage) -> None: # type: ignore
"""Post a message from the agent runtime to the message list."""
chat_messages = self.query_one("#chat-messages")
msg = AssistantMessage(f"{message.source}: {message.content}")
chat_messages.mount(msg)
msg.scroll_visible()
async def handle_tool_approval_request(self, message: ToolApprovalRequest) -> ToolApprovalResponse: # type: ignore
chat_messages = self.query_one("#chat-messages")
future: Future[ToolApprovalResponse] = asyncio.get_event_loop().create_future()
tool_call_approval_notice = ToolApprovalRequestNotice(message, future)
chat_messages.mount(tool_call_approval_notice)
tool_call_approval_notice.scroll_visible()
return await future
class TextualUserAgent(TypeRoutedAgent): # type: ignore
"""An agent that is used to receive messages from the runtime."""
def __init__(self, name: str, description: str, runtime: AgentRuntime, app: TextualChatApp) -> None: # type: ignore
super().__init__(name, description, runtime)
self._app = app
@message_handler # type: ignore
async def on_text_message(self, message: TextMessage, cancellation_token: CancellationToken) -> None: # type: ignore
await self._app.post_runtime_message(message)
@message_handler # type: ignore
async def on_respond_now(self, message: RespondNow, cancellation_token: CancellationToken) -> None: # type: ignore
await self._app.post_request_user_input_notice()
@message_handler # type: ignore
async def on_publish_now(self, message: PublishNow, cancellation_token: CancellationToken) -> None: # type: ignore
await self._app.post_request_user_input_notice()
@message_handler # type: ignore
async def on_tool_approval_request(
self, message: ToolApprovalRequest, cancellation_token: CancellationToken
) -> ToolApprovalResponse:
return await self._app.handle_tool_approval_request(message)
async def start_runtime(runtime: SingleThreadedAgentRuntime) -> None: # type: ignore
"""Run the runtime in a loop."""
while True:
await runtime.process_next()

1
pyproject.toml
View File

@@ -37,6 +37,7 @@ dev = [
"aiofiles",
"types-aiofiles",
"colorama",
"textual",
]
docs = ["sphinx", "furo", "sphinxcontrib-apidoc", "myst-parser"]

39
src/agnext/chat/agents/chat_completion_agent.py
View File

@@ -2,8 +2,6 @@ import asyncio
import json
from typing import Any, Coroutine, Dict, List, Mapping, Sequence, Tuple
from tqdm.asyncio import tqdm
from ...components import (
FunctionCall,
TypeRoutedAgent,
@@ -16,7 +14,7 @@ from ...components.models import (
SystemMessage,
)
from ...components.tools import Tool
from ...core import AgentRuntime, CancellationToken
from ...core import Agent, AgentRuntime, CancellationToken
from ..memory import ChatMemory
from ..types import (
FunctionCallMessage,
@@ -26,6 +24,8 @@ from ..types import (
RespondNow,
ResponseFormat,
TextMessage,
ToolApprovalRequest,
ToolApprovalResponse,
)
from ..utils import convert_messages_to_llm_messages
@@ -49,6 +49,11 @@ class ChatCompletionAgent(TypeRoutedAgent):
tool calls, the agent will call itselfs with the tool calls until it
gets a response that is not a list of tool calls, and then use that
response as the final response.
tool_approver (Agent | None, optional): The agent that approves tool
calls. Defaults to None. If no tool approver is provided, the agent
will execute the tools without approval. If a tool approver is
provided, the agent will send a request to the tool approver before
executing the tools.
"""
def __init__(
@@ -60,6 +65,7 @@ class ChatCompletionAgent(TypeRoutedAgent):
memory: ChatMemory,
model_client: ChatCompletionClient,
tools: Sequence[Tool] = [],
tool_approver: Agent | None = None,
) -> None:
super().__init__(name, description, runtime)
self._description = description
@@ -67,6 +73,7 @@ class ChatCompletionAgent(TypeRoutedAgent):
self._client = model_client
self._memory = memory
self._tools = tools
self._tool_approver = tool_approver
@message_handler()
async def on_text_message(self, message: TextMessage, cancellation_token: CancellationToken) -> None:
@@ -88,9 +95,7 @@ class ChatCompletionAgent(TypeRoutedAgent):
"""Handle a respond now message. This method generates a response and
returns it to the sender."""
# Generate a response.
with tqdm(desc=f"{self.name} is thinking...", bar_format="{desc}: {elapsed_s}") as pbar:
response = await self._generate_response(message.response_format, cancellation_token)
pbar.close()
response = await self._generate_response(message.response_format, cancellation_token)
# Return the response.
return response
@@ -100,10 +105,7 @@ class ChatCompletionAgent(TypeRoutedAgent):
"""Handle a publish now message. This method generates a response and
publishes it."""
# Generate a response.
# TODO: refactor this to use message_handler decorator.
with tqdm(desc=f"{self.name} is thinking...", bar_format="{desc}: {elapsed_s}", leave=False) as pbar:
response = await self._generate_response(message.response_format, cancellation_token)
pbar.close()
response = await self._generate_response(message.response_format, cancellation_token)
# Publish the response.
await self._publish_message(response)
@@ -221,6 +223,23 @@ class ChatCompletionAgent(TypeRoutedAgent):
tool = next((t for t in self._tools if t.name == name), None)
if tool is None:
return (f"Error: tool {name} not found.", call_id)
# Check if the tool needs approval
if self._tool_approver is not None:
# Send a tool approval request.
approval_request = ToolApprovalRequest(
tool_call=FunctionCall(id=call_id, arguments=json.dumps(args), name=name)
)
approval_response = await self._send_message(
message=approval_request,
recipient=self._tool_approver,
cancellation_token=cancellation_token,
)
if not isinstance(approval_response, ToolApprovalResponse):
raise ValueError(f"Expecting {ToolApprovalResponse.__name__}, received: {type(approval_response)}")
if not approval_response.approved:
return (f"Error: tool {name} approved, reason: {approval_response.reason}", call_id)
try:
result = await tool.run_json(args, cancellation_token)
result_as_str = tool.return_value_as_string(result)

18
src/agnext/chat/types.py
View File

@@ -54,3 +54,21 @@ class PublishNow:
class Reset: ...
@dataclass
class ToolApprovalRequest:
"""A message to request approval for a tool call. The sender expects a
response upon sending and waits for it synchronously."""
tool_call: FunctionCall
@dataclass
class ToolApprovalResponse:
"""A message to respond to a tool approval request. The response is sent
synchronously."""
tool_call_id: str
approved: bool
reason: str

2
src/agnext/components/_type_routed_agent.py
View File

@@ -120,7 +120,7 @@ def message_handler(
if target_types is None:
raise AssertionError("Message type not found")
print(type_hints)
# print(type_hints)
return_types = get_types(type_hints["return"])
if return_types is None: