py_agent_search

Description

py_agent_search is an AI agent based on LangGraph that allows you to perform web searches (via DuckDuckGo), Wikipedia lookups, and YouTube searches, with a streaming interface for generating responses. Logs are collected and sent to a Loki instance, viewable via Grafana. The agent’s memory is persisted in Redis, enabling it to maintain conversational context.

Prerequisites

• Docker and Docker Compose installed.
• Python 3.8 or higher.
• A .env file with the necessary environment variables (e.g., Redis, Loki configuration, etc.).
• If you use OpenAI as the LLM, you must define OPEN_API_KEY in the .env file or as an environment variable.

Docker Configuration

The project provides a docker-compose.yml file that configures the following services:

1. redis: Redis instance for the agent’s persistent memory.
2. loki: Loki instance to collect logs generated by the agent.
3. promtail: Promtail agent to ship Docker logs to Loki.
4. grafana: Grafana interface to visualize logs from Loki.

Starting Services

docker-compose up -d

After starting, Grafana will be available at http://localhot:3000 To view the agent’s logs, use the following query in Grafana (with Loki datasource preconfigured): ** {application="agent_search"} |= "" | json **

Note: If you encounter connection issues, verify that the hostname is correct (e.g., localhost:3000).

Environment Variables

Key environment variables should be defined in a .env file at the project root. Example contents:

# Redis configuration for persistent memory
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0

# Loki URL for sending logs (default: localhost:3100)
LOKI_URL=http://localhost:3100/loki/api/v1/push

# Application environment (development | production)
APP_ENV=development

# (Optional) OpenAI API key, if using ChatOpenAI
OPEN_API_KEY=your_openai_api_key

Logging

The log.py module configures a logger named agent_search that sends logs to:

• Loki (via LokiLoggerHandler) at the endpoint defined by LOKI_URL.
• Console (only in non-production environments) for local development.

The main function is:

from log import send_log

send_log(message: str, metadata: dict = {})

Whenever the agent generates an event (e.g., chat start, end of generation, errors), send_log is called with the message and metadata to store in Loki.

StreamingCallbackHandler

In stream.py, the StreamingCallbackHandler class is defined, extending LangChain’s BaseCallbackHandler. It is responsible for:

• Receiving streaming tokens generated by the LLM.
• Accumulating tokens and invoking send_log once generation completes.
• Providing callbacks for both the LLM and the tools (e.g., DuckDuckGo, Wikipedia, YouTube).

from langchain.callbacks.base import BaseCallbackHandler
from log import send_log

class StreamingCallbackHandler(BaseCallbackHandler):
    """
    Handler for streaming tokens from the LLM and logging to Loki.
    """
    metadata_logger = {"loki_metadata": {}}

    # ... implementation of on_llm_start, on_llm_new_token, on_llm_end ...

Persistent Memory

The memory.py file implements the agent’s persistent memory using Redis:

• AsyncRedisSaver (and its CheckpointSaver) allows saving conversation checkpoints in Redis.
• On each new chat, the agent can retrieve previous context from Redis to maintain conversational continuity.

Example configuration in main.py:

redis_conf = {
    "host": os.getenv("REDIS_HOST", "localhost"),
    "port": os.getenv("REDIS_PORT", 6379),
    "db": os.getenv("REDIS_DB", 0)
}

agent = AgentSearch(redis_persistence_config=redis_conf)

Search Tools

In tools.py, the available tools for the agent are defined:

1. DuckDuckGoSearchRun (search_tool): Performs generic web searches via DuckDuckGo.
2. WikipediaQueryRun (wikipedia_tool): Queries Wikipedia for information on a topic.
3. YouTubeSearchTool (youtube_tool): Searches for videos on YouTube.

All tools use StreamingCallbackHandler to track streaming tokens and generate logs.

from langchain_community.tools import WikipediaQueryRun, YouTubeSearchTool, DuckDuckGoSearchRun
from stream import StreamingCallbackHandler

handler = StreamingCallbackHandler()

search_tool = DuckDuckGoSearchRun(
    name="duckduckgo_search",
    description="Search for information on the web via DuckDuckGo.",
    callbacks=[handler],
    return_direct=False,
    response_format="content"
)

wikipedia_tool = WikipediaQueryRun(
    name="wikipedia_search",
    description="Search for information on Wikipedia.",
    callbacks=[handler],
    return_direct=False,
    response_format="content"
)

youtube_tool = YouTubeSearchTool(
    name="youtube_search",
    description="Search for videos on YouTube.",
    callbacks=[handler],
    return_direct=False,
    response_format="content"
)

Installation

The package can be installed via pip:

pip install py_agent_search

This command will install all necessary dependencies, including LangChain, LangGraph, Redis, Loki Logger Handler, and the community search tools.

Framework Used

The project leverages LangGraph as the main framework for orchestrating processing nodes, checkpoint memory management, and tool handling. Specifically:

• create_react_agent from LangGraph to create a “ReAct” style agent.
• AsyncRedisSaver from LangGraph for persistent memory.
• StreamingCallbackHandler from LangChain for streaming callbacks.

Usage Example

An example of initializing the agent and streaming a response is provided in main.py:

import asyncio
import uuid
import os
from py_agent_search import AgentSearch
from log import send_log

redis_conf = {
    "host": os.getenv("REDIS_HOST", "localhost"),
    "port": os.getenv("REDIS_PORT", 6379),
    "db": os.getenv("REDIS_DB", 0)
}

async def main():
    """
    Initialize the agent and start a sample conversation.
    """
    try:
        # Generate a unique thread_id for the conversation
        thread_id = str(uuid.uuid4())
        question = input("Enter your question: ")

        # Send an initial log message
        send_log(message="Starting chat interaction", metadata={"question": question})
        print(f"Question: {question}\nWaiting for response...")

        # Instantiate the agent with Redis persistence
        agent = AgentSearch(redis_persistence_config=redis_conf)

        # Stream tokens as they are generated
        async for token in agent.stream(input=question, thread_id=thread_id):
            print(token, end="", flush=True)

    except Exception as e:
        print(f"An error occurred: {e}")
        send_log(message="Error during chat interaction", metadata={"error": str(e)})

if __name__ == "__main__":
    asyncio.run(main())

Details on AgentSearch

• Defined in agent.py.
• Uses LangGraph’s create_react_agent and LangChain’s ChatOpenAI for response generation.
• Integrates the tools (search_tool, wikipedia_tool, youtube_tool) to extend information retrieval capabilities.
• Relies on AsyncRedisSaver for persistent memory.
• Each phase (LLM call, tool execution, end of streaming) sends logs to Loki via send_log.

Project Structure

py_agent_search/
├── agent.py            # Definition of the AgentSearch class, agent orchestration
├── stream.py           # StreamingCallbackHandler for streaming logging
├── memory.py           # Implementation of persistent memory with Redis
├── tools.py            # Definition of search tools (DuckDuckGo, Wikipedia, YouTube)
├── log.py              # Logger configuration and log shipping to Loki
├── main.py             # Example of agent initialization and usage
├── Dockerfile          # (Optional) Dockerfile for building a custom Python container
├── docker-compose.yml  # Configuration for services (Redis, Loki, Promtail, Grafana)
├── .env                # Environment variable definitions (not included in the repository)
└── README.md           # This descriptive file

Starting the Agent

1. Ensure you have created the .env file with the correct variables.

2. Start the supporting services:

   docker-compose up -d
   

3. Install the Python package (if not already installed):

   pip install py_agent_search
   

Conclusions

This project provides a fully equipped “ReAct” AI agent including:

• Search tools (web, Wikipedia, YouTube).
• Token-by-token streaming of responses.
• Context persistence in Redis.
• Centralized logging to Loki, viewable in Grafana.
• Easy installation via pip and configuration through Docker Compose.

For more details, refer to the individual implementation files (agent.py, stream.py, memory.py, tools.py, log.py, main.py) and the official documentation of LangGraph and LangChain.