基本概念
Custom Agent 是指用户根据特定需求自定义的智能代理程序，它可以执行特定的任务或工作流程。通过 @tool 装饰器可以将自定义功能绑定为Agent可调用的工具。
创建 Custom Agent 的步骤
1. 定义Agent类
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. 使用@tool装饰器绑定工具
@tool
def search_internet(self, query: str) -> str:
"""执行互联网搜索并返回结果
Args:
query: 搜索关键词或问题
Returns:
搜索结果摘要
"""
# 实际搜索实现代码
return f"关于'{query}'的搜索结果: ..."3. 注册工具到Agent
def register_tool(self, tool_func):
"""将工具函数注册到Agent"""
self.tools.append(tool_func)
return tool_func完整示例
from typing import List, Dict, Any
class DataAnalysisAgent(CustomAgent):
def __init__(self):
super().__init__("数据分析助手", "专业的数据处理和分析Agent")
@tool
def data_cleaning(self, raw_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""对原始数据进行清洗处理
Args:
raw_data: 原始数据列表，每个元素是一个字典
Returns:
清洗后的数据列表
"""
# 实现数据清洗逻辑
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:
"""基于数据生成分析报告
Args:
data: 输入数据
metrics: 需要分析的指标列表
Returns:
格式化分析报告
"""
# 实现报告生成逻辑
report = f"数据分析报告\n\n共分析{len(data)}条记录\n"
for metric in metrics:
values = [d.get(metric, 0) for d in data]
report += f"{metric}: 平均值={sum(values)/len(values):.2f}\n"
return report工具绑定高级用法
1. 动态工具绑定
def external_api_call(api_endpoint: str, params: dict) -> dict:
"""调用外部API"""
# 实际API调用代码
return {"status": "success", "data": {}}
# 动态绑定外部工具
agent = CustomAgent("API代理")
agent.register_tool(external_api_call)2. 带配置的工具
def create_calculator_tool(unit: str = "dollars"):
@tool
def currency_calculator(amount: float) -> str:
"""货币计算器"""
return f"{amount} {unit}"
return currency_calculator
# 使用带配置的工具
agent.register_tool(create_calculator_tool("euros"))应用场景
- 客户服务Agent：绑定FAQ查询、工单创建、客户信息查询等工具
- 数据分析Agent：绑定数据清洗、可视化、报告生成等工具
- 内容创作Agent：绑定文本生成、语法检查、SEO优化等工具
- IT运维Agent：绑定服务器监控、日志查询、故障诊断等工具
安装依赖
pip install -qU langchain-core langchain-openai编写代码
定义一个工具：
# 定义工具
@tool
def get_word_length(word: str) -> int:
"""Returns the length of a word."""
return len(word)创建一个Agent：
# 创建Agent
agent = (
{
"input": lambda x: x["input"],
"agent_scratchpad": lambda x: format_to_openai_tool_messages(
x["intermediate_steps"]
),
}
| prompt
| llm_with_tools
| OpenAIToolsAgentOutputParser()
)推荐使用GPT-4，GPT3.5任务表现上并不是很好。
完整代码
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)
# 定义工具
@tool
def get_word_length(word: str) -> int:
"""Returns the length of a word."""
return len(word)
# print(get_word_length.invoke("abc"))
# 定义一个工具集
tools = [get_word_length]
# 提示词
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are very powerful assistant, but don't know current events",
),
(
"user",
"{input}"
),
MessagesPlaceholder(variable_name="agent_scratchpad"),
]
)
# 绑定工具集
llm_with_tools = llm.bind_tools(tools)
# 创建Agent
agent = (
{
"input": lambda x: x["input"],
"agent_scratchpad": lambda x: format_to_openai_tool_messages(
x["intermediate_steps"]
),
}
| prompt
| llm_with_tools
| OpenAIToolsAgentOutputParser()
)
# 执行器
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}")执行结果
➜ python3 test27.py
> Entering new AgentExecutor chain...
Invoking: `get_word_length` with `{'word': 'eudca'}`
5The word "eudca" has 5 letters.
> Finished chain.