Add AWS CloudWatch logs agent toolkit

libs/experimental/extended_testing_deps.txt

@@ -6,3 +6,4 @@ sentence-transformers>=2,<3
 jinja2>=3,<4
 pandas>=2.0.1,<3
 tabulate>=0.9.0,<1
+boto3>=1.35.84,<2

libs/experimental/langchain_experimental/agents/__init__.py

@@ -9,6 +9,7 @@
 """
 
 from langchain_experimental.agents.agent_toolkits import (
+    create_cloudwatch_logs_agent,
     create_csv_agent,
     create_pandas_dataframe_agent,
     create_spark_dataframe_agent,
@@ -20,4 +21,5 @@
     "create_pandas_dataframe_agent",
     "create_spark_dataframe_agent",
     "create_xorbits_agent",
+    "create_cloudwatch_logs_agent",
 ]

libs/experimental/langchain_experimental/agents/agent_toolkits/__init__.py

@@ -1,3 +1,6 @@
+from langchain_experimental.agents.agent_toolkits.cloudwatch.base import (
+    create_cloudwatch_logs_agent,
+)
 from langchain_experimental.agents.agent_toolkits.csv.base import create_csv_agent
 from langchain_experimental.agents.agent_toolkits.pandas.base import (
     create_pandas_dataframe_agent,
@@ -16,4 +19,5 @@
     "create_spark_dataframe_agent",
     "create_python_agent",
     "create_csv_agent",
+    "create_cloudwatch_logs_agent",
 ]

libs/experimental/langchain_experimental/agents/agent_toolkits/cloudwatch/__init__.py

@@ -0,0 +1 @@
+"""cloudwtach toolkit"""

libs/experimental/langchain_experimental/agents/agent_toolkits/cloudwatch/base.py

@@ -0,0 +1,129 @@
+import os
+from datetime import datetime, timedelta
+from typing import Any, Dict, List, Optional
+
+import boto3
+from botocore.exceptions import NoCredentialsError, PartialCredentialsError
+from langchain.agents.agent import AgentExecutor
+from langchain.agents.mrkl.base import ZeroShotAgent
+from langchain.chains.llm import LLMChain
+from langchain_core.callbacks.base import BaseCallbackManager
+from langchain_core.language_models import BaseLLM
+
+from langchain_experimental.agents.agent_toolkits.cloudwatch.prompt import (
+    PREFIX,
+    SUFFIX,
+)
+
+DEFAULT_REGION = "us-east-1"
+
+region_name = os.getenv("AWS_REGION", DEFAULT_REGION)
+
+# Initialize the boto3 client for CloudWatch Logs
+cloudwatch_logs_client = boto3.client("logs", region_name=region_name)
+
+
+def _validate_cloudwatch_client(client: Any) -> bool:
+    """Validates the CloudWatch client."""
+    try:
+        # Basic check to ensure the client is valid
+        return isinstance(client, type(client))
+    except Exception:
+        return False
+
+
+def _validate_aws_credentials() -> None:
+    try:
+        boto3.client("sts").get_caller_identity()
+    except (NoCredentialsError, PartialCredentialsError):
+        raise ValueError("AWS credentials are not configured properly.")
+
+
+def fetch_cloudwatch_logs(
+    cloudwatch_client: Any,
+    log_group_name: str,
+    start_time: datetime,
+    end_time: datetime,
+) -> str:
+    """Fetches logs from AWS CloudWatch."""
+    try:
+        response = cloudwatch_client.filter_log_events(
+            logGroupName=log_group_name,
+            startTime=int(start_time.timestamp() * 1000),
+            endTime=int(end_time.timestamp() * 1000),
+        )
+        error_keywords = ["ERROR", "Exception", "Traceback"]
+        logs = "\n".join(
+            event["message"]
+            for event in response.get("events", [])
+            if any(keyword in event["message"] for keyword in error_keywords)
+        )
+        return logs
+    except Exception as e:
+        raise ValueError(f"Error fetching logs: {e}")
+
+
+def create_cloudwatch_logs_agent(
+    llm: BaseLLM,
+    cloudwatch_client: Any,
+    log_group_name: str,
+    callback_manager: Optional[BaseCallbackManager] = None,
+    prefix: str = PREFIX,
+    suffix: str = SUFFIX,
+    input_variables: Optional[List[str]] = None,
+    verbose: bool = False,
+    return_intermediate_steps: bool = False,
+    max_iterations: Optional[int] = 15,
+    max_execution_time: Optional[float] = None,
+    early_stopping_method: str = "force",
+    agent_executor_kwargs: Optional[Dict[str, Any]] = None,
+    **kwargs: Any,
+) -> AgentExecutor:
+    """
+    Construct an AWS CloudWatch logs analyzer agent using an LLM.
+
+    """
+    if not _validate_cloudwatch_client(cloudwatch_client):
+        raise ValueError("Invalid CloudWatch client provided.")
+
+    _validate_aws_credentials()
+
+    if input_variables is None:
+        input_variables = ["logs", "input", "agent_scratchpad"]
+
+    # Fetch recent logs
+    now = datetime.utcnow()
+    ten_minutes_ago = now - timedelta(minutes=10)
+    logs = fetch_cloudwatch_logs(
+        cloudwatch_client, log_group_name, ten_minutes_ago, now
+    )
+
+    # Prompt and tools
+    tools: list = []
+    prompt = ZeroShotAgent.create_prompt(
+        tools, prefix=prefix, suffix=suffix, input_variables=input_variables
+    )
+    partial_prompt = prompt.partial(logs=logs)
+    llm_chain = LLMChain(
+        llm=llm,
+        prompt=partial_prompt,
+        callback_manager=callback_manager,
+    )
+    tool_names = [tool.name for tool in tools]
+    agent = ZeroShotAgent(  # type: ignore[call-arg]
+        llm_chain=llm_chain,
+        allowed_tools=tool_names,
+        callback_manager=callback_manager,
+        **kwargs,
+    )
+    return AgentExecutor.from_agent_and_tools(
+        agent=agent,
+        tools=tools,
+        callback_manager=callback_manager,
+        verbose=verbose,
+        return_intermediate_steps=return_intermediate_steps,
+        max_iterations=max_iterations,
+        max_execution_time=max_execution_time,
+        early_stopping_method=early_stopping_method,
+        **(agent_executor_kwargs or {}),
+    )

libs/experimental/langchain_experimental/agents/agent_toolkits/cloudwatch/prompt.py

@@ -0,0 +1,7 @@
+# flake8: noqa
+
+PREFIX = """You are an expert in log analysis for cloud-based systems."""
+
+SUFFIX = """Analyze the following logs and provide insights. Identify errors, potential issues, and recommend solutions:
+{logs}
+"""