Basic Concepts

A Custom Agent refers to an intelligent agent program customized by users based on specific requirements, which can execute specific tasks or workflows. Custom functions can be bound as tools callable by the Agent using the @tool decorator.

Steps to Create a Custom Agent

1. Define Agent Class

from typing import Optional
from langchain.agents import AgentExecutor
from langchain.agents import tool

class CustomAgent:
    def __init__(self, name: str, description: Optional[str] = None):
        self.name = name
        self.description = description or f"{name} custom agent"
        self.tools = []

2. Use @tool Decorator to Bind Tools

@tool
def search_internet(query: str) -> str:
    """Perform internet search and return results

    Args:
        query: Search keywords or questions

    Returns:
        Search result summary
    """
    # Actual search implementation code
    return f"Search results for '{query}': ..."

3. Register Tools to Agent

def register_tool(self, tool_func):
    """Register tool function to Agent"""
    self.tools.append(tool_func)
    return tool_func

Complete Example

from typing import List, Dict, Any

class DataAnalysisAgent(CustomAgent):
    def __init__(self):
        super().__init__("Data Analysis Assistant", "Professional data processing and analysis Agent")

    @tool
    def data_cleaning(self, raw_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Clean raw data

        Args:
            raw_data: List of raw data, each element is a dictionary

        Returns:
            Cleaned data list
        """
        # Implement data cleaning logic
        cleaned_data = []
        for item in raw_data:
            cleaned_item = {k: v.strip() if isinstance(v, str) else v
                          for k, v in item.items()}
            cleaned_data.append(cleaned_item)
        return cleaned_data

    @tool
    def generate_report(self, data: List[Dict[str, Any]], metrics: List[str]) -> str:
        """Generate analysis report based on data

        Args:
            data: Input data
            metrics: List of metrics to analyze

        Returns:
            Formatted analysis report
        """
        # Implement report generation logic
        report = f"Data Analysis Report\n\nAnalyzed {len(data)} records\n"
        for metric in metrics:
            values = [d.get(metric, 0) for d in data]
            report += f"{metric}: Average={sum(values)/len(values):.2f}\n"
        return report

Advanced Tool Binding Usage

1. Dynamic Tool Binding

def external_api_call(api_endpoint: str, params: dict) -> dict:
    """Call external API"""
    # Actual API call code
    return {"status": "success", "data": {}}

# Dynamically bind external tools
agent = CustomAgent("API Agent")
agent.register_tool(external_api_call)

2. Tools with Configuration

def create_calculator_tool(unit: str = "dollars"):
    @tool
    def currency_calculator(amount: float) -> str:
        """Currency calculator"""
        return f"{amount} {unit}"
    return currency_calculator

# Use tools with configuration
agent.register_tool(create_calculator_tool("euros"))

Application Scenarios

  1. Customer Service Agent: Bind tools for FAQ queries, ticket creation, customer information lookup
  2. Data Analysis Agent: Bind tools for data cleaning, visualization, report generation
  3. Content Creation Agent: Bind tools for text generation, grammar checking, SEO optimization
  4. IT Operations Agent: Bind tools for server monitoring, log querying, fault diagnosis

Install Dependencies

pip install -qU langchain-core langchain-openai

Write Code

Define a tool:

# Define tool
@tool
def get_word_length(word: str) -> int:
    """Returns the length of a word."""
    return len(word)

Create an Agent:

# Create Agent
agent = (
    {
        "input": lambda x: x["input"],
        "agent_scratchpad": lambda x: format_to_openai_tool_messages(
            x["intermediate_steps"]
        ),
    }
    | prompt
    | llm_with_tools
    | OpenAIToolsAgentOutputParser()
)

It is recommended to use GPT-4, as GPT-3.5 does not perform well on tasks.

Complete Code

from langchain_openai import ChatOpenAI
from langchain.agents import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents.format_scratchpad.openai_tools import (
    format_to_openai_tool_messages,
)
from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser
from langchain.agents import AgentExecutor


llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0)
# llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# Define tool
@tool
def get_word_length(word: str) -> int:
    """Returns the length of a word."""
    return len(word)


# print(get_word_length.invoke("abc"))
# Define a tool set
tools = [get_word_length]
# Prompt
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are very powerful assistant, but don't know current events",
        ),
        (
            "user",
            "{input}"
        ),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
)

# Bind tool set
llm_with_tools = llm.bind_tools(tools)

# Create Agent
agent = (
    {
        "input": lambda x: x["input"],
        "agent_scratchpad": lambda x: format_to_openai_tool_messages(
            x["intermediate_steps"]
        ),
    }
    | prompt
    | llm_with_tools
    | OpenAIToolsAgentOutputParser()
)
# Executor
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
agent_executor_stream = list(agent_executor.stream({"input": "How many letters in the word eudca"}))
print(f"agent_executor_stream: {agent_executor_stream}")

Execution Results

➜ python3 test27.py


> Entering new AgentExecutor chain...

Invoking: `get_word_length` with `{'word': 'eudca'}`


5The word "eudca" has 5 letters.

> Finished chain.