一文学会LangChain搭建Agent工具

Agent（智能代理）是一种能够感知环境、自主决策并执行任务的实体（可以是软件、机器人或系统）。它的核心能力在于自动化、智能化和适应性，广泛应用于多个领域。以下是Agent能做的事情，按场景分类说明：

1. 自动化任务处理

● 日常重复性工作：自动填写表格、整理数据、发送邮件。

● 流程自动化：替代人工完成订单处理、库存管理、报销审批等企业流程。

● 案例：企业中的RPA（机器人流程自动化）代理可自动处理财务对账。

2. 个性化服务

● 推荐系统：根据用户行为推荐商品（如电商）、内容（如短视频）或音乐（如Spotify）。

● 私人助理：管理日程、提醒事项（如Siri、Alexa），甚至预订餐厅。

● 案例：Netflix的推荐算法通过分析观看历史推荐影视剧。

3. 实时分析与决策

● 金融交易：高频交易Agent根据市场数据自动买卖股票。

● 风险控制：检测欺诈交易（如银行风控系统实时拦截异常刷卡）。

● 案例：天气预报Agent整合卫星数据，实时预测灾害并触发警报。

本文将从实际应用角度，带你学会Agent工具的编写。

工欲善其事，必先利其器。

本文项目都是在Python环境下使用的，使用到的插件有：

● LangChain

● LangSmith

● 通义千问DashScope

其中LangChain是我们用到的大语言模型的主要Python插件。

● LangChain 是一个用于开发由大型语言模型 (LLMs) 驱动的应用程序的框架。它简化了应用程序的开发流程，通过LangChain我们可以调用各个LLM大模型构建我们的应用。

● LangSmith是评估大模型能力好坏的评估工具，能够量化评估基于大模型的系统的效果。它允许您调试、测试、评估和监控基于任何 LLM 框架构建的链和智能代理，并无缝集成 LangChain（用于构建 LLM 的首选开源框架）。

● 安装Anaconda

自己下载安装。

● 创建Python 3.10环境

电脑开始菜单，打开Anaconda Prompt:

创建虚拟环境:

打开Anaconda Prompt窗口，输入：

conda create -n langchain python=3.10

进入langchain的Python 3.10环境:

conda activate langchain

安装依赖包：

pip install langchain-openai -i http://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com

LangSmith密钥申请

1.  登录网址：https://www.langchain.com/langsmith

这样就获得了LangChain的密钥

通义千问DashScope密钥申请

1.  登录网址:https://bailian.console.aliyun.com/?tab=model#/api-key

使用阿里云百炼平台，可支付宝扫码登录。

2.  进入APIKEY管理平台:https://bailian.console.aliyun.com/?tab=model#/api-key

这样就获得了通义千问的密钥。

在langchain-demo中新建demo04.py文件，翻译英文，代码如下:

import os


from langchain_community.llms import Tongyi
from langchain_core.messages import HumanMessage, SystemMessage

os.environ["LANGCHAIN_TRACING_V2"]="true"
os.environ["LANGCHAIN_API_KEY"]="LANGCHAIN_API_KEY"# 此处为langsmith的API KEY
os.environ["DASHSCOPE_API_KEY"]="DASHSCOPE_API_KEY"# 此处为通义千问的API KEY

# 使用语言模型 通义千问
llm =Tongyi()

messages =[
SystemMessage(content="Translate the following from English into chinese."),
HumanMessage(content="hi!"),
]
print(llm.invoke(messages))

将前面APIKEY在线申请的LangSmith和阿里通义千问的APIKEY分别填入项目：

打开Anaconda Prompt进入langchain-demo项目目录：

运行demo04.py:

python demo04.py

将Hi!翻译为你好！

在langchain-demo中新建demo05.py文件，引入了LangChain的StrOutputParser解析器：

import os


from langchain_community.llms import Tongyi
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.output_parsers import StrOutputParser

os.environ["LANGCHAIN_TRACING_V2"]="true"
os.environ["LANGCHAIN_API_KEY"]="LANGCHAIN_API_KEY"# 此处为langsmith的API KEY
os.environ["DASHSCOPE_API_KEY"]="DASHSCOPE_API_KEY"# 此处为通义千问的API KEY

# 输出解析器
llm =Tongyi()
parser =StrOutputParser()

messages =[
SystemMessage(content="Translate the following from English into chinese."),
HumanMessage(content="I am fine!"),
]

# | 运算符在 LangChain 中用于将两个元素组合在一起。
chain = llm | parser
print(chain.invoke(messages))

替换APIKEY密钥之后：

新建demo06.py文件，动态传入提示词：

import os


from langchain_community.llms import Tongyi
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

os.environ["LANGCHAIN_TRACING_V2"]="true"
os.environ["LANGCHAIN_API_KEY"]="LANGCHAIN_API_KEY"# 此处为langsmith的API KEY
os.environ["DASHSCOPE_API_KEY"]="DASHSCOPE_API_KEY"# 此处为通义千问的API KEY

llm =Tongyi()
parser =StrOutputParser()

# 1.创建一个字符串，我们将格式化为系统消息
system_template ="Translate the following into {language}:"

# 2.创建 PromptTemplate。这将是 system_template 和一个更简单的模板的组合，用于放置要翻译的文本
prompt_template = ChatPromptTemplate.from_messages(
[("system", system_template),("user","{text}")]
)

chain = prompt_template | llm | parser

result = chain.invoke({"language":"chinese","text":"hello world"})
print(result)

替换APIKEY之后：

运行demo06.py：

python demo06.py

根据动态传入的语言:chinese，以及英文单词hello world翻译结果：

使用LangServe， LangServe 帮助开发者将 LangChain 链部署为 REST WEB服务。

首先安装LangServe

打开Anaconda Prompt安装LangServe:

pip install "langserve[all]"

编写服务器serve.py文件

import os


from fastapi import FastAPI
from langchain_community.llms import Tongyi
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langserve import add_routes

os.environ["LANGCHAIN_TRACING_V2"]="true"
os.environ["LANGCHAIN_API_KEY"]="LANGCHAIN_API_KEY"# 此处为langsmith的API KEY
os.environ["DASHSCOPE_API_KEY"]="DASHSCOPE_API_KEY"# 此处为通义千问的API KEY

llm =Tongyi()
parser =StrOutputParser()

system_template ="Translate the following into {language}:"

prompt_template = ChatPromptTemplate.from_messages(
[("system", system_template),("user","{text}")]
)

chain = prompt_template | llm | parser

app =FastAPI(
    title="花宇心坊翻译机器Server",
    version="1.0",
    description="A simple API server using LangChain's Runnable interfaces",
)

add_routes(
    app,
    chain,
    path="/chain",
)

if __name__ =="__main__":
    import uvicorn

    uvicorn.run(app, host="localhost", port=8000)

将启动一个WEB服务，更新APIKEY:

运行WEB程序：

python serve.py

出现了类似SpringBoot项目启动的模板，都是WEB服务。

打开浏览器，输入:

http://localhost:8000/chain/playground/

输入我们想翻译的语言德语：

German language

以及翻译内容：

我想有个家

客户端client.py调用服务端

编写client.py文件调用服务器端接口获取数据:

from langserve import RemoteRunnable


 
remote_chain =RemoteRunnable("http://localhost:8000/chain/")


result = remote_chain.invoke({"language":"chinese","text":"hi"})


print(result)

从服务端进行远程调用，查看运行结果，再开一个Anaconda Prompt:

python client.py

成功将hi翻译为中文你好，以上是LangChain接入大语言模型的用法，接下来进入Agent部分。

还是在Anaconda Prompt里的langchain环境下安装相关依赖：

pip install -U langchain-community langgraph langchain-anthropic tavily-python

编写agent01.py文件，调用Tavily使用Tool查询深圳天气：

import os


from langchain_community.tools.tavily_search import TavilySearchResults

# 使用Tavily 搜索引擎:https://app.tavily.com/home
os.environ["TAVILY_API_KEY"]="TAVILY_API_KEY"
os.environ["LANGCHAIN_API_KEY"]="LANGCHAIN_API_KEY"

search =TavilySearchResults(max_results=2)
search_results = search.invoke("深圳天气怎么样？")
print(search_results)
# If we want, we can create other tools.
# Once we have all the tools we want, we can put them in a list that we will reference later.
tools =[search]

将TAVILT KEY和LangChain KEY替换后：

查询了深圳市的天气预报，返回格式是富文本，需要做处理后才可正常显示。

根据前面申请的通义千问KEY和LangChain调用LLMChain获取Agent模型调用。

新建agent02.py文件:

import os


from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import create_react_agent
from langchain_community.llms import Tongyi
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# 使用Tavily 搜索引擎:https://app.tavily.com/home
os.environ["TAVILY_API_KEY"]="TAVILY_API_KEY"
os.environ["LANGCHAIN_API_KEY"]="LANGCHAIN_API_KEY"
os.environ["DASHSCOPE_API_KEY"]="DASHSCOPE_API_KEY"

llm=Tongyi(temperature=1)
template='''
        你的名字是小黑子,当人问问题的时候,你都会在开头加上'唱,跳,rap,篮球!',然后再回答{question}
    '''
prompt=PromptTemplate(
        template=template,
        input_variables=["question"]#这个question就是用户输入的内容,这行代码不可缺少
)
chain =LLMChain(#将llm与prompt联系起来
        llm=llm,
        prompt=prompt
)
question='你是谁'
res=chain.invoke(question)#运行
print(res['text'])#打印结果

将TAVILY和LangChain以及通义千问DashScope的APIKEY复制粘贴进来：

根据Template指令，询问模型你是谁，看看结果:

python agent02.py

下面是一个示例工具类 MyAgentTool，其中定义了一个用于执行基本数学计算的工具 math。

from langchain.agents import Tool
from typing import Any

class MyAgentTool:
    def __init__(self)->None:
# 初始化工具，可以在此添加更多本地工具
        pass

    def tools(self):
return[
Tool(
                name="math",
                description="用于执行基本的数学计算，比如加减乘除。",
                func=self.math_tool,
),
Tool(
                name="compare_numbers",
                description="比较两个数的大小，格式：'num1,num2'。",
                func=self.compare_numbers,
)
]

    def math_tool(self,input: str)->str:
"""
        简单的数学计算工具，解析输入的数学表达式并返回结果。
        """
try:
# 安全地计算数学表达式
# 仅允许数字和基本运算符
            allowed_chars ="0123456789+-*/(). "
ifnotall(char in allowed_chars for char in input):
return"输入包含不允许的字符。"
            result =eval(input)
returnstr(result)
        except Exception as e:
return f"计算错误: {e}"

    def compare_numbers(self,input: str)->str:
"""
        比较两个数的大小，输入格式为 'num1,num2'。
        返回比较结果：num1 > num2, num1 < num2 或 num1 = num2。
        """
try:
# 解析输入字符串为两个数字
            num1_str, num2_str = input.split(',')
            num1 =float(num1_str.strip())
            num2 =float(num2_str.strip())

# 比较两个数字的大小
if num1 >num2:
return f"{num1} > {num2}"
            elif num1 <num2:
return f"{num1} < {num2}"
else:
return f"{num1} = {num2}"
        except ValueError:
return"输入格式错误，请使用 'num1,num2' 形式。"

# 实例化工具类
tools =MyAgentTool()

# 调用 math 工具
print(tools.math_tool("3 + 5 * 2"))# 输出: 13
print(tools.math_tool("10 / 2 - 3"))# 输出: 2.0
print(tools.math_tool("import os"))# 输出: 输入包含不允许的字符。

方法说明了它是数学工具，定义了两个函数：

math:用于执行基本的数学计算，比如加减乘除。

compare_numbers:比较两个数的大小，格式：'num1,num2。

并注册为Tool工具

先运行一下：

python MyAgentTool.py

根据

编写一个agent03.py调用MyAgentTool工具：

import os


from langchain.agents import AgentExecutor, LLMSingleActionAgent, AgentOutputParser
from langchain.prompts import StringPromptTemplate
from langchain.chains import LLMChain
from typing import List, Union, Any
from langchain.schema import AgentAction, AgentFinish, OutputParserException
import re
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import Tool

from MyAgentTool import MyAgentTool

# 使用Tavily 搜索引擎:https://app.tavily.com/home
os.environ["TAVILY_API_KEY"]="TAVILY_API_KEY"
os.environ["LANGCHAIN_API_KEY"]="LANGCHAIN_API_KEY"
os.environ["DASHSCOPE_API_KEY"]="DASHSCOPE_API_KEY"
class MyAgent:
    def __init__(self)->None:
# Agent 的提示词模板
        self.template ="""请尽可能详细地回答下面的问题，你将始终用中文回答。当需要时，你可以使用以下工具:
                        {tools}
                        请按照以下格式回答:
                        Question: {input}
                        Thought: 你应该思考下一步该做什么
                        Action: 选择一个操作，必须是以下工具之一 [{tool_names}]
                        Action Input: 该操作的输入
                        Observation: 该操作的结果
                        ...（此 Thought/Action/Action Input/Observation 可以重复多次）
                        Thought: 我现在知道最终的答案了
                        Final Answer: {final_answer}
                        现在开始! 记住使用中文回答，如果使用英文回答将受到惩罚。
                        Question: {input}
                        {agent_scratchpad}"""

# 定义语言模型（LLM）
        self.llm =ChatTongyi(model='qwen-plus')

# 初始化工具列表
        self.tools =MyAgentTool().tools()

# 创建 Agent 的提示词
        self.prompt = self.MyTemplate(
            template=self.template,
            tools=self.tools,
            input_variables=["input","intermediate_steps"],
)

# 定义 LLMChain
        self.llm_chain =LLMChain(
            llm=self.llm,
            prompt=self.prompt
)

# 获取工具名称列表
        self.toolnames =[tool.name for tool in self.tools]

# 定义一个 LLMSingleActionAgent
        self.agent =LLMSingleActionAgent(
            llm_chain=self.llm_chain,
            allowed_tools=self.toolnames,
            output_parser=self.MyOutputParser(),
            stop=["\nObservation:"],
)

# 运行 Agent 的方法
    def run(self,input: str)->str:
        agent_executor = AgentExecutor.from_agent_and_tools(
            agent=self.agent,
            tools=self.tools,
            handle_parsing_errors=True,
            verbose=True
)
return agent_executor.run(input=input)

# 自定义模板渲染类
class MyTemplate(StringPromptTemplate):
template: str
tools: List[Tool]

        def format(self,**kwargs: Any)->str:
# 获取中间步骤
            intermediate_steps = kwargs.pop("intermediate_steps")
            thoughts =""
for action, observation inintermediate_steps:
                thoughts += action.log
                thoughts += f"\nObservation: {observation}\nThought: "

# 将 agent_scratchpad 设置为该值
            kwargs["agent_scratchpad"]= thoughts

# 从提供的工具列表中创建一个名为 tools 的变量
            kwargs["tools"]="\n".join([f"{tool.name}: {tool.description}"for tool in self.tools])

# 创建一个提供的工具名称列表
            kwargs["tool_names"]=", ".join([tool.name for tool in self.tools])

# 确保传递 final_answer 变量
if"final_answer"notinkwargs:
                kwargs["final_answer"]="暂时没有答案"

return self.template.format(**kwargs)

# 自定义输出解析类
class MyOutputParser(AgentOutputParser):
        def parse(self,output: str)-> Union[AgentAction, AgentFinish]:
if"Final Answer:"inoutput:
return AgentFinish(
                    return_values={"output": output.split("Final Answer:")[-1].strip()},
                    log=output,
)
# 使用正则解析出动作和动作输入
            regex = r"Action\s*\d*\s*:(.*?)\nAction\s*\d*\s*Input\s*\s*:(.*)"
            match = re.search(regex, output, re.DOTALL)
ifnotmatch:
                raise OutputParserException(f"无法解析 LLM 输出: `{output}`")
            action = match.group(1).strip()
            action_input = match.group(2).strip(" ").strip('"')
# 返回操作和操作输入
return AgentAction(tool=action, tool_input=action_input, log=output)

if __name__ =="__main__":
    myagent =MyAgent()
    question ="请计算以下表达式的结果：2 + 3 * 4 - 5 / 2"
    result = myagent.run(question)
print("Agent 的回答:", result)
    question ="请比较以下两个数字的大小：10, 20"
    result = myagent.run(question)
print("Agent 的回答:", result)
 

更新Tavily、LangChain和通义DashScope里面的APIKEY:

提示词模板 (self.template)

提示词模板定义了 Agent 如何组织其思考和行动流程。这是与语言模型交互的关键部分。

占位符解释：

{tools}：列出所有可用工具及其描述。

{input}：用户输入的问题。

{tool_names}：所有工具的名称列表。

{agent_scratchpad}：中间步骤记录，包括思考、行动、观察结果。

语言模型（LLM）

使用 ChatTongyi 作为语言模型。

self.llm = ChatTongyi(model='qwen-plus')

说明：

ChatTongyi 是来自 langchain_community 的自定义语言模型。请根据实际使用的模型调整。

工具集成

初始化工具列表，并获取所有工具的名称。

self.tools = MyAgentTool().tools()

self.toolnames = [tool.name for tool in self.tools]

LLMChain

将语言模型和提示词模板结合，创建一个 LLMChain 实例。

self.prompt = self.MyTemplate( template=self.template, tools=self.tools, input_variables=["input", "intermediate_steps"], )

self.llm_chain = LLMChain( llm=self.llm, prompt=self.prompt )

说明：

LLMChain 负责将提示词模板传递给语言模型，并获取生成的响应。

Agent 初始化

使用 LLMSingleActionAgent 定义 Agent 的行为。

self.agent = LLMSingleActionAgent( llm_chain=self.llm_chain, allowed_tools=self.toolnames, output_parser=self.MyOutputParser(), stop=["\nObservation:"], )

参数说明：

llm_chain：与语言模型和提示词模板关联的链。

allowed_tools：Agent 允许调用的工具名称列表。

output_parser：自定义的输出解析器，用于解析语言模型的输出。

stop：停止词，指示语言模型在生成特定内容后停止。

运行 Agent

通过实例化 MyAgent 类并调用 run 方法，可以让 Agent 处理输入问题并返回回答。

if __name__ =="__main__":
    myagent =MyAgent()
    question ="请计算以下表达式的结果：2 + 3 * 4 - 5 / 2"
    result = myagent.run(question)
print("Agent 的回答:", result)
    question ="请比较以下两个数字的大小：10, 20"
    result = myagent.run(question)
print("Agent 的回答:", result)

查看结果：

python agent03.py