Introduction: The Power of AI-Assisted Communication in Webex
In today's fast-paced digital workplace, effective communication is key. Imagine having an intelligent assistant in your Cisco Webex Teams that can search the web, retrieve user information, and perform calculations - all while maintaining context across conversations. That's exactly what we're going to build in this tutorial!
By the end of this guide, you'll have created a powerful AI-powered Webex bot using LangGraph's stateful large language model (LLM) agents. This bot will enhance your team's interactions within Webex by providing intelligent responses and performing complex tasks efficiently.
Project Overview: What We're Building and Why
Our Webex Bot AI Assistant will leverage the following key technologies:
- 🧠 LangChain & LangGraph: For building stateful, multi-actor AI applications
- 🚀 OpenAI: Providing the underlying language model
- 🤖 Webex Bot lib: For seamless integration with Webex Teams leveraging WebSockets
- 📊 SQLite: Lightweight database for maintaining conversation history
Key Features:
- 🔍 Web search capabilities with Tavily Search
- 👤 Retrieval of Webex user information
- 🧮 Mathematical operations (e.g., calculating a number's power)
- 💾 Persistent conversation history for improved context awareness
Prerequisites:
- Basic knowledge of Python
- Understanding of bot frameworks (helpful, but not required)
- Understanding of LangGraph agents (helpful, but not required)
Let's dive in and start building!
Setting Up Your Development Environment
Before we start coding, let's set up our environment:
1. Clone the project repository:
git clone https://github.com/lieranderl/webexbot-langgraph-assistant-template
2. Create a .env file in the project root with the following variables:
# LLM ENVIRONMENT
OPENAI_API_BASE=<OpenAI API base URL>
OPENAI_API_KEY=<Your OpenAI API key>
LLM_MODEL=<Name of llm model>
# LANGCHAIN ENVIRONMENT
LANGCHAIN_ENDPOINT=<LangSmith endpoint URL>
LANGCHAIN_API_KEY=<Your LangSmith API key>
LANGCHAIN_PROJECT=<Name of your LangSmith project>
# Tools API
TAVILY_API_KEY=<Your Tavily API key for web search>
# SQLite Database for Conversation History
SQL_CONNECTION_STR=checkpoints.db
# Webex Bot Configuration
WEBEX_TEAMS_ACCESS_TOKEN=<Your Webex Bot Access Token>
WEBEX_TEAMS_DOMAIN=<Your Webex Domain>
3. Install dependencies using Poetry:
poetry install
Now that our environment is set up, let's start building our AI assistant!
Project Structure and SOLID Principles
Our project follows a structure that promotes better organization, maintainability, and adheres to SOLID principles:
project_root/
│
├── src/
│ ├── graph_reactagent/
│ │ ├── __init__.py
│ │ ├── interfaces.py
│ │ ├── graph.py
│ │ ├── invoker.py
│ │ ├── messages_filter.py
│ │ ├── prompt_formatter.py
│ │ └── tools.py
│ │
│ └── webexbot/
│ ├── __init__.py
│ ├── commands.py
│ └── webexbot.py
│
├── tests/
│ ├── __init__.py
│ ├── test_graph_reactagent.py
│ ├── test_messages_filter.py
│ ├── test_tools.py
│ └── test_webexbot.py
│
├── .env
├── .coveragerc
├── pyproject.toml
└── README.md
This structure separates concerns and allows for easier extension and modification of individual components. Let's dive into some key files and how they embody SOLID principles:
interfaces.py
The interfaces.py file defines the interfaces (protocols in Python) that our classes will implement. This adheres to the Interface Segregation Principle (the 'I' in SOLID) by providing specific interfaces for different functionalities.
from typing import Protocol, Any, Dict, List
from datetime import datetime
from langchain_core.messages import AnyMessage
class IMessageFilter(Protocol):
def filter_messages(self, messages: List[AnyMessage]) -> List[AnyMessage]: ...
class IPromptFormatter(Protocol):
def format_prompt(
self, messages: List[AnyMessage], system_time: datetime
) -> Any: ...
class IGraphInvoker(Protocol):
def invoke(self, message: str, **kwargs: Any) -> Dict[str, Any]: ...
These interfaces allow us to define the contract that classes must follow without specifying the implementation. This promotes loose coupling and makes our system more modular and easier to extend.
messages_filter.py
The messages_filter.py file contains the implementation of our message filtering logic. This adheres to the Single Responsibility Principle (the 'S' in SOLID) by focusing solely on the task of filtering messages.
from typing import List
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, AnyMessage
from graph_reactagent.interfaces import IMessageFilter
class DefaultMessageFilter(IMessageFilter):
def filter_messages(self, messages: List[AnyMessage]) -> List[AnyMessage]:
message_count = min(6, len(messages))
filtered_messages = messages[-message_count:]
while message_count > 1:
if isinstance(filtered_messages[0], HumanMessage):
break
message_count -= 1
while message_count > 0 and isinstance(
messages[-message_count], (AIMessage, ToolMessage)
):
# we cannot have an assistant message at the start of the chat history
# if after removal of the first, we have an assistant message,
# we need to remove the assistant message too
# all tool messages should be preceded by an assistant message
# if we remove a tool message, we need to remove the assistant message too
message_count -= 1
filtered_messages = messages[-message_count:]
return filtered_messages
This implementation ensures that we maintain a manageable context window for our AI model by limiting the number of messages passed to it. It also makes sure that tool-related messages are properly contextualized by including the message that triggered the tool use.
prompt_formatter.py
The prompt_formatter.py file handles the formatting of our prompts. This adheres to the Open/Closed Principle (the 'O' in SOLID) by allowing for easy extension of formatting behavior without modifying existing code.
from typing import Any, List
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from .interfaces import IPromptFormatter
from langchain_core.messages import AnyMessage
class DefaultPromptFormatter(IPromptFormatter):
def __init__(self, prompt: ChatPromptTemplate):
self.prompt = prompt
def format_prompt(self, messages: List[AnyMessage], system_time: datetime) -> Any:
return self.prompt.invoke(
{
"messages": messages,
"system_time": system_time.isoformat(),
}
)
This formatter takes a ChatPromptTemplate and uses it to format the messages and system time into a prompt suitable for our AI model. By using a template, we can easily modify the prompt structure without changing the formatting logic.
Crafting the Brain: Building the ReAct Agent with LangGraph
Now that we've covered our supporting files, let's take a deeper look at how we build our ReAct agent in the graph.py file:
from typing import Any, Dict
from datetime import datetime, timezone
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent, ToolNode
from langchain_core.prompts import ChatPromptTemplate
from .interfaces import IMessageFilter, IPromptFormatter
from .messages_filter import DefaultMessageFilter
from .prompt_formatter import DefaultPromptFormatter
from .tools import search, get_webex_user_info, power
SYSTEM_PROMPT = """You are a helpful AI assistant in a Webex chat.
Your primary tasks include:
1. Searching the web to provide relevant information.
2. Retrieving personal information from Webex.
3. Performing power calculations.
4. Interacting with users to assist with their queries.
Ensure your responses are clear, accurate, and concise. Always aim to provide the most helpful information based on the user's request.
System time: {system_time}"""
class Graph:
def __init__(
self,
model: str,
tools: list,
message_filter: IMessageFilter,
prompt_formatter: IPromptFormatter,
name: str = "ReAct Agent",
):
self.model = ChatOpenAI(model=model)
self.tools = ToolNode(tools)
self.message_filter = message_filter
self.prompt_formatter = prompt_formatter
self.name = name
self.graph = self._create_graph()
def _create_graph(self):
graph = create_react_agent(
self.model, self.tools, state_modifier=self._format_for_model
)
graph.name = self.name
return graph
def _format_for_model(self, state: Dict[str, Any]) -> Any:
messages = state["messages"]
filtered_messages = self.message_filter.filter_messages(messages)
return self.prompt_formatter.format_prompt(
filtered_messages, datetime.now(timezone.utc).astimezone()
)
def create_default_graph():
prompt = ChatPromptTemplate.from_messages(
[
("system", SYSTEM_PROMPT),
("placeholder", "{messages}"),
]
)
tools = [get_webex_user_info, search, power]
return Graph(
model="gpt-4o",
tools=tools,
message_filter=DefaultMessageFilter(),
prompt_formatter=DefaultPromptFormatter(prompt),
name="Webex Bot Demo ReAct Agent",
)
This Graph class encapsulates the creation and configuration of our ReAct agent. It uses dependency injection (adhering to the Dependency Inversion Principle, the 'D' in SOLID) to allow for flexible message filtering and prompt formatting.
The _format_for_model method ties together our message filtering and prompt formatting. It first applies the message filter to limit the context window, then uses the prompt formatter to prepare the input for our AI model.
By structuring our code this way, we've created a flexible and extensible system that can easily accommodate changes or additions to our bot's capabilities. For example, if we wanted to implement a different message filtering strategy or change our prompt format, we could do so by creating new classes that implement the IMessageFilter or IPromptFormatter interfaces, respectively, without having to modify the existing Graph class.
This setup creates a powerful ReAct agent that can understand context, use tools, and generate appropriate responses.
Empowering Your Bot: Implementing Custom Tools
Our bot's capabilities are enhanced by custom tools. Let's look at the key tools in our tools.py file:
1. Web Search Tool:
def search(
query: str, *, config: Annotated[RunnableConfig, InjectedToolArg]
) -> Optional[List[Dict[str, Any]]]:
"""Search for general web results.
Search for real-time information using the Tavily search engine.
This function performs a search using the Tavily search engine, which is designed
to provide comprehensive, accurate, and trusted results. It's particularly useful
for answering questions about current events.
"""
max_results: int = config.get("configurable", {}).get("max_results") or 5
wrapped = TavilySearchResults(max_results=max_results)
result = wrapped.invoke({"query": query})
return cast(List[Dict[str, Any]], result)
2. Power Calculation Tool:
def power(a: int, b: int) -> int:
"""Calculate power of a number."""
return a**b
3. Webex User Info Retrieval Tool:
def get_webex_user_info(
config: Annotated[RunnableConfig, InjectedToolArg],
) -> Optional[Dict[str, str]]:
"""Get user information: email and user name/display name from Webex SDK"""
displayName: str = config.get("configurable", {}).get("displayName") or ""
email: str = config.get("configurable", {}).get("email") or ""
return {
"displayName": displayName,
"email": email,
}
These tools allow our bot to perform web searches, mathematical calculations, and retrieve user information from Webex.
Maintaining Context: Checkpointing with SQLite
To ensure our bot maintains context across conversations, we use SQLite for checkpointing. Here's how we set it up in our invoker.py file:
class GraphInvoker(IGraphInvoker):
def __init__(
self,
graph: Graph,
connection_str: str = "",
llm_model: str = "",
temperature: float = 0.1,
):
self.graph = graph
self.connection_str = connection_str or os.getenv("SQL_CONNECTION_STR")
self.llm_model = llm_model or os.getenv("LLM_MODEL")
self.temperature = temperature
def invoke(self, message: str, **kwargs: Any) -> Dict[str, Any]:
if not self.connection_str:
raise ValueError("No database connection string provided")
if not self.llm_model:
raise ValueError("No LLM model provided")
with SqliteSaver.from_conn_string(self.connection_str) as saver:
self.graph.graph.checkpointer = saver
run_id = uuid4()
config = RunnableConfig(
configurable={
"model": self.llm_model,
"temperature": self.temperature,
**kwargs,
},
run_id=run_id,
)
result = self.graph.graph.invoke(
input={"messages": [HumanMessage(content=message)]},
config=config,
)
return result
This setup ensures that the conversation state persists between interactions.
Bringing It All Together: Integrating with Webex Teams
Now, let's integrate our AI assistant with Webex Teams.
First, we need to create and register the bot on Webex. For instructions on how to create and register your bot, please follow the Webex Bot guide
In our webexbot.py file:
def create_bot():
load_dotenv()
graph = create_default_graph()
invoker = GraphInvoker(graph)
openai_command = OpenAI(invoker)
return WebexBot(
teams_bot_token=os.getenv("WEBEX_TEAMS_ACCESS_TOKEN"),
approved_domains=[os.getenv("WEBEX_TEAMS_DOMAIN")],
bot_name="AI-Assistant",
bot_help_subtitle="",
threads=False,
help_command=openai_command,
)
if __name__ == "__main__":
bot = create_bot()
bot.run()
And in our commands.py file:
class OpenAI(Command):
def __init__(self, invoker: IGraphInvoker):
super().__init__()
self.invoker = invoker
def execute(self, message, attachment_actions, activity):
response = self.invoker.invoke(
message,
thread_id=activity["target"]["globalId"],
email=activity["actor"]["id"],
displayName=activity["actor"]["displayName"],
max_results=5,
)
return response["messages"][-1].content
This setup allows our bot to process all text inputs using our ReAct agent and respond within Webex Teams.
🚀 Running your AI Assistant
This project uses Poetry for dependency management and Poe the Poet for task running. Read more in readme
To start your Webex Bot AI Assistant, use the following command:
poe start
or
poetry run python -m webexbot.webexbot
Your bot is now live and ready to assist users in Webex Teams!
Example Interaction
Supercharging Development: Tracing and Analyzing with LangSmith
To gain insights into your bot's performance and behavior, we've integrated LangSmith for tracing. This allows you to visualize the decision-making process of your AI assistant and optimize its performance.
Conclusion: Your Webex Bot AI Assistant in Action
Congratulations! You've successfully built a powerful Webex Bot AI Assistant using LangGraph's stateful LLM agents. Your bot can now:
- Understand and maintain context in conversations
- Perform web searches for up-to-date information
- Retrieve Webex user information
- Calculate a number's power
- Provide intelligent responses to user queries
This AI-powered assistant will significantly enhance communication and productivity within your Webex Teams environment.
Customizing Your Webex Bot AI Assistant
While the Webex Bot AI Assistant we've built provides a solid foundation, you may want to adapt it for your specific use cases. Here are some ways you can customize the bot to better suit your needs:
1. Adding New Tools
One of the most straightforward ways to extend your bot's capabilities is by adding new tools. Here's how you can do this:
1. Define a new tool function in tools.py:
def new_custom_tool(query: str, config: Annotated[RunnableConfig, InjectedToolArg]) -> str:
# Implement your custom tool logic here
return f"Result of custom tool for query: {query}"
2. Add the new tool to the create_default_graph function in graph.py:
def create_default_graph():
# ... existing code ...
tools = [get_webex_user_info, search, power, new_custom_tool]
return Graph(
model="gpt-4o",
tools=tools,
message_filter=DefaultMessageFilter(),
prompt_formatter=DefaultPromptFormatter(prompt),
name="Webex Bot Demo ReAct Agent",
)
3. Update the system prompt in graph.py to inform the AI about the new tool:
SYSTEM_PROMPT = """You are a helpful AI assistant in a Webex chat.
Your primary tasks include:
1. Searching the web to provide relevant information.
2. Retrieving personal information from Webex.
3. Performing power calculations.
4. [Description of your new custom tool]
5. Interacting with users to assist with their queries.
Ensure your responses are clear, accurate, and concise. Always aim to provide the most helpful information based on the user's request.
System time: {system_time}"""
2. Customizing Message Filtering
If you need a different message filtering strategy, you can create a new class that implements the IMessageFilter interface:
class CustomMessageFilter(IMessageFilter):
def filter_messages(self, messages: List[AnyMessage]) -> List[AnyMessage]:
# Implement your custom filtering logic here
return messages # This is just a placeholder
Then, update the create_default_graph function to use your new filter:
def create_default_graph():
# ... existing code ...
return Graph(
model="gpt-4o",
tools=tools,
message_filter=CustomMessageFilter(),
prompt_formatter=DefaultPromptFormatter(prompt),
name="Webex Bot Demo ReAct Agent",
)
3. Modifying the Prompt Format
To change how the AI interprets messages, you can modify the prompt format. Create a new class implementing the IPromptFormatter interface:
class CustomPromptFormatter(IPromptFormatter):
def __init__(self, prompt: ChatPromptTemplate):
self.prompt = prompt
def format_prompt(self, messages: List[AnyMessage], system_time: datetime) -> Any:
# Implement your custom prompt formatting logic here
return self.prompt.invoke(
{
"messages": messages,
"system_time": system_time.isoformat(),
# Add any additional context you want to include
}
)
Update the create_default_graph function to use your new formatter:
def create_default_graph():
# ... existing code ...
custom_prompt = ChatPromptTemplate.from_messages(
[
("system", YOUR_CUSTOM_SYSTEM_PROMPT),
("placeholder", "{messages}"),
]
)
return Graph(
model="gpt-4o",
tools=tools,
message_filter=DefaultMessageFilter(),
prompt_formatter=CustomPromptFormatter(custom_prompt),
name="Webex Bot Demo ReAct Agent",
)
4. Integrating with External Services
You can enhance your bot by integrating it with external services relevant to your organization. For example, you might add a tool to query a company database or interact with a project management system.
Here's a simple example of how you might create a tool to interact with a hypothetical API:
import requests
def query_company_database(query: str, config: Annotated[RunnableConfig, InjectedToolArg]) -> str:
api_key = config.get("configurable", {}).get("company_db_api_key")
if not api_key:
return "Error: API key for company database not provided"
response = requests.get(f"https://api.company-database.com/query?q={query}", headers={"Authorization": f"Bearer {api_key}"})
if response.status_code == 200:
return response.json()["result"]
else:
return f"Error querying company database: {response.status_code}"
Remember to add the new API key to your .env file and update your bot's configuration to pass it to the tool.
By leveraging these customization techniques, you can adapt the Webex Bot AI Assistant to better fit your organization's specific needs and workflows. Whether you're adding new capabilities, changing how the bot processes messages, or integrating with your existing systems, the modular design of the bot makes these customizations straightforward to implement.
Troubleshooting
If you encounter any issues while setting up or running your bot, check these common problems:
- API Key Issues: Ensure all API keys in your
.envfile are correct and up-to-date. - Dependency Conflicts: Make sure you're using the correct versions of all libraries. Check the
pyproject.tomlfile for version specifications. - Webex Integration: If your bot isn't responding in Webex, verify that your bot's access token is correct and that it has the necessary permissions.
For more specific issues, consult the documentation of the relevant libraries or reach out to the community on dev.to or GitHub.
Resources
Remember, building AI-powered bots is an iterative process. Don't be afraid to experiment, learn, and improve your bot over time. Good luck with your project!
Top comments (0)