7、langchain-在智能体中定义工具

官方文档：https://reference.langchain.com/python/langchain/tools/#langchain.tools.tool

# 工具的定义

工具是Agent调用执行动作的组件，他们通过让模型通过定义明确的输入和输出与交互来拓展模型的功能。

在构建Agent时，需要为其提供一个他可以使用的工具列表。除了实际可调用的函数之外，工具还需要有以下属性：

1. name：工具的名称
2. descrition：工具的说明，给模型提供指导
3. args_schema：使用 Pydantic 模块或 JSON 模块定义复杂的输入参数
4. parse_docstring：True/False 默认False，是否从 Google Style 函数文档中解析参数描述
5. 。。。更多详细看官方文档

# 定义工具的方式

# 1、写一个函数，使用@tool装饰器定义工具

使用@tool装饰器定义工具有个弊端：适合简单的、要么纯同步要么纯异步的函数。它无法同时包装两个函数。

两种方式

1、通过标准的参数

from langchain.tools import tool
from pydantic import BaseModel, Field

# 定义参数
class SearchArgs(BaseModel):
    query: str = Field(..., description="需要进行联网查询的信息关键词")

@tool("web_search", args_schema=SearchArgs, description="互联网搜索的工具，可以联网搜索任何公开信息")
def web_search(query: str) -> str:
    
    pass

2、通过谷歌标准的注释格式，同时必须将parse_docstring参数设为True

@tool("web_search_", parse_docstring=True)
def web_search_(query: str) -> str:
    """
    互联网搜索的工具，可以联网搜索任何公开信息
    Args:
        query: 需要进行联网查询的信息关键词

    Returns:
        返回搜索的结果信息，该信息是一个文本字符串
    """
    pass

# 完整示例：

src.tools.wab_search.py

from langchain_core.tools import tool, ToolException
from pydantic import BaseModel, Field
import aiohttp
import asyncio

# 1. 定义参数
class SearchArgs(BaseModel):
    query: str = Field(..., description="需要进行联网查询的信息关键词")


# 2. 定义异步工具
@tool("web_search", 
      args_schema=SearchArgs, 
      description="互联网搜索的工具，可以联网搜索任何公开信息") 
async def web_search(query: str) -> str:
    """异步工具"""
    url = 'https://www.searchapi.io/api/v1/search'
    params = {
        'api_key': 'TYYvR37',
        'engine': 'google',
        'q': query
    }
    
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, params=params) as response:
                if response.status != 200:
                    return f"API 请求失败，状态码: {response.status}"
                
                result = await response.text()
                return result
                
    except Exception as e:
        return f"网络请求发生异常: {str(e)}"

# @tool("web_search_", parse_docstring=True)
# def web_search_(query: str) -> str:
#     """
#     互联网搜索的工具，可以联网搜索任何公开信息
#     Args:
#         query: 需要进行联网查询的信息关键词

#     Returns:
#         返回搜索的结果信息，该信息是一个文本字符串
#     """
#     pass
# --- 测试代码 ---
if __name__ == "__main__":
    async def test():
        print(web_search.name)  # 工具名称
        print(web_search.description)  # 工具描述
        print(web_search.args)  # 工具的参数
        print(web_search.args_schema.model_json_schema())  # 工具的参数的描述(json)

        # 正常调用测试
        print("--- 测试 1: 正常调用 ---")
        res = await web_search.ainvoke({"query": "LangChain"})
        print(res)

    asyncio.run(test())

src.agent.py

from langchain.agents import create_agent
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os
from src.tools.web_search import web_search
load_dotenv('../.env')


llm = ChatOpenAI(
    base_url=os.getenv('OPENAI_BASE_URL'),
    api_key=os.getenv('OPENAI_API_KEY'),
    model=os.getenv('OPENAI_MODEL'),
)

# 创建agent时传入工具
agent = create_agent(
    model=llm,
    tools=[web_search],
    system_prompt="你是一个智能助手，尽可能回答用户问题"
)

# 2、继承BaseTool类

复杂的工具可以使用这种方式-自定义类

需要重写BaseTool类的run函数，但是我们在重写时需要定义成_run 私有化函数 _arun异步调用的函数也是要写的，不然调用 .ainvoke时还是调用 _run

工具的参数有两种实现方式

src.tools.web_search_class.py

from langchain.tools import BaseTool
from typing import Type
from pydantic import Field, BaseModel, create_model
import asyncio
class SearchArgs(BaseModel):
    query: str = Field(..., description="需要进行联网查询的信息关键词")

class WebSearch(BaseTool):
    name: str = 'web_search_class'  # 工具的名称
    description: str = '互联网搜索的工具，可以联网搜索任何公开信息'  # 工具的描述
    # 工具参数第一种写法  使用这个方式不需要重写init方法
    # args_schema: Type[BaseModel] = SearchArgs  # 工具的参数  
    
    # 工具参数第二种写法（如果要用这种，需要注释掉上面的 args_schema）
    def __init__(self):
        super().__init__()
        self.args_schema = create_model('SearchInput', query=(str, Field(..., description='互联网搜索的工具，可以联网搜索任何公开信息')))
    
    def _run(self, query: str) -> str:
        try:
            import requests

            url = 'https://www.searchapi.io/api/v1/search'
            params = {
                'api_key':'TYYv2ijxp2aWTeqWCWbsVR37',
                'engine': 'google',
                'q': query
            }

            response = requests.get(url, params=params)
            print(response.text)
            return response.text
        except Exception as e:
            return str(e)
            
    def _arun(self, query: str) -> str:
        import aiohttp
        url = 'https://www.searchapi.io/api/v1/search'
        params = {
            'api_key':'TYYv2sVR37',
            'engine': 'google',
            'q': query
        }
        try:
            async with aiohttp.ClientSession as session:
                async with session.get(url, params=params) as response:
                    return await response.text()
        except Exception as e:
            return f"Error:{e}"

src.agent.py

from langchain.agents import create_agent
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os
# from src.tools.web_search import web_search
from src.tools.web_search_class import WebSearch
load_dotenv('../.env')

WebSearchTool = WebSearch()
llm = ChatOpenAI(
    base_url=os.getenv('OPENAI_BASE_URL'),
    api_key=os.getenv('OPENAI_API_KEY'),
    model=os.getenv('OPENAI_MODEL'),
)

# 创建agent时传入工具
agent = create_agent(
    model=llm,
    tools=[WebSearchTool],
    system_prompt="你是一个智能助手，尽可能回答用户问题"
)

# 3、将Runnable对象定义成工具

Runnable对象.as_tool

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, SystemMessage
from dotenv import load_dotenv
from pydantic import BaseModel, Field
import os
load_dotenv('../../../.env')
llm = ChatOpenAI(
    base_url=os.getenv('OPENAI_BASE_URL'),
    api_key=os.getenv('OPENAI_API_KEY'),
    model=os.getenv('OPENAI_MODEL'),
)
prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content='你是一个诗人'),
    ('human', '帮我写一个{topic}主题{language}语言的诗句')
])
class ToolArgs(BaseModel):
    topic: str = Field(description='诗句的主题')
    language: str = Field(description='诗句采用的语言')
chain = prompt | llm | StrOutputParser()
potry_tool = chain.as_tool(
    name='potry_tool',
    description="写诗句的工具",
    args_schema=ToolArgs,
)
print(potry_tool.args_schema.model_json_schema())
print(potry_tool.name)
print(potry_tool.description)
print(potry_tool.invoke({'topic': '春节', 'language':'中文'}))

  potry_tool = chain.as_tool(
{'properties': {'topic': {'description': '诗句的主题', 'title': 'Topic', 'type': 'string'}, 'language': {'description': '诗句采用的语言', 'title': 'Language', 'type': 'string'}}, 'required': ['topic', 'language'], 'title': 'ToolArgs', 'type': 'object'}
potry_tool
写诗句的工具
爆竹声声辞旧岁，  
灯火辉煌迎新春。  
家家户户笑语暖，  
福满华堂喜气浓。

# 4、从MCP服务端获取工具

以后再说

# 案例

runnable.as_tool实现的写诗句的工具

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, SystemMessage
from dotenv import load_dotenv
from pydantic import BaseModel, Field
import os
load_dotenv('../.env')
llm = ChatOpenAI(
    base_url=os.getenv('OPENAI_BASE_URL'),
    api_key=os.getenv('OPENAI_API_KEY'),
    model=os.getenv('OPENAI_MODEL'),
)
prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content='你是一个诗人'),
    ('human', '帮我写一个{topic}主题{language}语言的诗句')
])
class ToolArgs(BaseModel):
    topic: str = Field(description='诗句的主题')
    language: str = Field(description='诗句采用的语言')
chain = prompt | llm | StrOutputParser()
potry_tool = chain.as_tool(
    name='potry_tool',
    description="写诗句的工具",
    args_schema=ToolArgs,
)
# print(potry_tool.args_schema.model_json_schema())
# print(potry_tool.name)
# print(potry_tool.description)
# print(potry_tool.invoke({'topic': '春节', 'language':'中文'}))

BaseTool实现的联网搜索的工具

from langchain.tools import BaseTool
from typing import Type
from pydantic import Field, BaseModel, create_model
import asyncio
class SearchArgs(BaseModel):
    query: str = Field(..., description="需要进行联网查询的信息关键词")

class WebSearch(BaseTool):
    name: str = 'web_search_class'  # 工具的名称
    description: str = '互联网搜索的工具，可以联网搜索任何公开信息'  # 工具的描述
    return_direct: bool = False
    # 工具参数第一种写法
    # args_schema: Type[BaseModel] = SearchArgs  # 工具的参数
    
    # 工具参数第二种写法（如果要用这种，需要注释掉上面的 args_schema）
    def __init__(self):
        super().__init__()
        self.args_schema = create_model('SearchInput', query=(str, Field(..., description='互联网搜索的工具，可以联网搜索任何公开信息')))
    
    def _run(self, query: str) -> str:
        try:
            import requests

            url = 'https://www.searchapi.io/api/v1/search'
            params = {
                'api_key':'TYYv2ijxp2aWTeqWCWbsVR37',
                'engine': 'google',
                'q': query
            }

            response = requests.get(url, params=params)
            print(response.text)
            return response.text
        except Exception as e:
            return str(e)
            
    async def _arun(self, query: str) -> str:
        import aiohttp
        url = 'https://www.searchapi.io/api/v1/search'
        params = {
            'api_key':'TYYv2ijxp2aWTeqWCWbsVR37',
            'engine': 'google',
            'q': query
        }
        try:
            async with aiohttp.ClientSession as session:
                async with session.get(url, params=params) as response:
                    return await response.text()
        except Exception as e:
            return f"Error:{e}"

agent

ps:应该是用langgraph dev 启动，但是忘了账号了

from langchain.agents import create_agent
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os
# from src.tools.web_search import web_search
from src.tools.web_search_class import WebSearch
from src.tools.runnable_tool import potry_tool
load_dotenv('../.env')

WebSearchTool = WebSearch()
llm = ChatOpenAI(
    base_url=os.getenv('OPENAI_BASE_URL'),
    api_key=os.getenv('OPENAI_API_KEY'),
    model=os.getenv('OPENAI_MODEL'),
)

# 创建agent时传入工具
agent = create_agent(
    model=llm,
    tools=[WebSearchTool, potry_tool],
    system_prompt="你是一个智能助手，尽可能回答用户问题"
)

# # 调用agent来处理这个query
# response = agent.runs(query)


# print(agent.invoke({"messages": [("user", "帮我写个中文的关于雪的诗")]}))
# print('---------------------')
# print(agent.invoke({"messages": [("user", "帮我从网上找一篇关于雪的诗")]}))
import asyncio

async def main():
    result = await agent.ainvoke({"messages": [("user", "帮我写个中文的关于雪的诗")]})
    print(result)

asyncio.run(main())