Langchain模块Agent组件

 2025/12/23 

LangChain Agent 是一个让大型语言模型（LLM）自主决策、选择工具并执行任务的框架。它把 LLM 从一个”纯对话者”变成了一个”能动手的智能助手”。
用户问题 → LLM分析 → 选择工具 → 执行工具 → 观察结果 → 继续思考 → … → 最终答案

类型	特点	适用场景
ZERO_SHOT_REACT	无需示例，根据工具描述决策	通用任务，结构清晰
CONVERSATIONAL	记忆对话历史，上下文感知	多轮对话任务
STRUCTURED_CHAT	结构化输出，更可控	复杂多步骤任务
OPENAI_FUNCTIONS	利用OpenAI函数调用	需要精准参数传递

agent简单调用

pip install ddgs wikipedia-api requests numexpr

from langchain.agents import initialize_agent,AgentType,Tool
from langchain_deepseek import ChatDeepSeek
from ddgs import DDGS
from langchain.chains import LLMMathChain
from dotenv import load_dotenv
load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat")
search = DDGS()
llm_math = LLMMathChain.from_llm(llm)

tools = [
    Tool(name="Search", func=search.text, description="搜索信息"),
    Tool(name="Lookup",func=search.text,description="查找时使用"),
    Tool(name="Calculator", func=llm_math.run, description="数学计算"),
]
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    #verbose=True,
    handle_parsing_errors=True,
)

print(agent.invoke("北京奥运会召开的年份除以四，得到的结果开根是多少"))
print(agent.invoke("中华人民共和国是什么时候成立的"))

#**Thought:** 北京奥运会是2008年举办的。先除以4，再开平方根。  
# **Action:** Calculator  
# **Action Input:** 2008 ÷ 4 = 502，然后 sqrt(502)  

# **Observation:** 2008 ÷ 4 = 502，√502 ≈ 22.40535650240808  

# **Thought:** 现在得到最终答案。  
# **Final Answer:** 约等于 22.405

# {'input': '中华人民共和国是什么时候成立的', 'output': '** 中华人民共和国成立于1949年10月1日。'}

Conversational agent

专为对话优化的代理，一般会与memory相结合

使用ddgs（免费的网上搜索工具，但信息滞后

from langchain.agents import Tool,AgentType,initialize_agent
from langchain.memory import ConversationBufferMemory
from langchain_deepseek import ChatDeepSeek
from ddgs import DDGS
from dotenv import load_dotenv
load_dotenv()

llm = ChatDeepSeek(model="deepseek-chat")

# DDGS搜索返回的是历史数据，搜索结果来自2024年7月等历史时间
search = DDGS()

tools = [
    Tool(name="Search",func=search.text,description="useful for searching the internet"),
]
memory = ConversationBufferMemory(memory_key="chat_history")

agent_chain = initialize_agent(
    tools,
    llm, 
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION, 
    memory=memory, 
    verbose=True
)

agent_chain.invoke({"input": "中国的首都是哪里?"}) # 常识性问题，不需要用工具
agent_chain.invoke({"input": "今天是几月几号?"}) # DDGS数据滞后，如果要用专业api，需要花钱
agent_chain.invoke({"input": "小明今年2025年15岁"})
agent_chain.invoke({"input": "2015年小明几岁？"}) # 上下文记忆
agent_chain.invoke({"input": "北京和上海今天天气如何"}) # 用工具搜索

# 根据已知信息，小明在2025年是15岁，因此他出生于2010年。那么在2015年，他的年龄是2015减去2010，也就是5岁。

OpenAI multi function agent（略）

Self ask with search

自问自答的方式拆解问题，

我新使用了GoogleSerperAPIWrapper谷歌的网上检索工具，官网https://serper.dev/，新注册可以有2500次免费访问次数

我这里有报错，是因为AgentType.SELF_ASK_WITH_SEARCH对输出格式的要求很高，如果使用gpt就可以，但是我使用的是deepseek，所以最后报错，但中间推理过程是没问题的

from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain.agents import AgentType, initialize_agent, Tool

from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
load_dotenv()

llm = ChatDeepSeek(model="deepseek-chat")
search = GoogleSerperAPIWrapper()
tools = [Tool(
    name="Intermediate Answer", # 必须是这个名字
    func=search.run,
    description="useful for when you need to answer questions about current events",
)]

self_ask_agent_chain = initialize_agent(
    tools, 
    llm, 
    agent=AgentType.SELF_ASK_WITH_SEARCH,
    verbose=False,
    handle_parsing_errors=True
)

print(self_ask_agent_chain.run("“java之父”的故乡在哪里？"))

# **后续问题：** “java之父”是谁？
# **中间答案：** “Java之父”是詹姆斯·高斯林（James Gosling）。  

# **后续问题：** 詹姆斯·高斯林的故乡是哪里？  
# **中间答案：** 他出生在加拿大阿尔伯塔省的卡尔加里附近。  

# 所以最终答案是：**加拿大阿尔伯塔省卡尔加里附近**。

自定义的工具multi-input Tools

定义一个计算工具

from langchain_deepseek import ChatDeepSeek
from langchain.agents import initialize_agent,AgentType
from langchain.tools import StructuredTool
from dotenv import load_dotenv
load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat")

# 定义工具函数
def multipliter(a:float, b:float) -> float:
    """将两个数字相乘，返回乘积""" # 这个注释是必须给的，用于描述工具函数的功能
    return a * b
tool = [StructuredTool.from_function(multipliter)]

# 初始化agent
agent_executor = initialize_agent(
    tools=tool,
    llm=llm,
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, # 使用STRUCTURED
    verbose=True
)

print(agent_executor.invoke("10乘以10是多少"))

# Thought:用户想知道10乘以10的结果，我可以使用乘法工具来计算。

# Action:
# ```
# {
#   "action": "multipliter",
#   "action_input": {"a": 10, "b": 10}
# }
# ```
# Observation: 100.0
# Thought:我已经通过乘法工具计算出了10乘以10的结果，现在可以给出最终答案。

# Action:
# ```
# {
#   "action": "Final Answer",
#   "action_input": "10乘以10等于100。"
# }
# ```

# > Finished chain.
# {'input': '10乘以10是多少', 'output': '10乘以10等于100。'}

工具输入规范Tool input Schema

定义工具输入参数结构的组件，是给LLM看的，不是给人看的

使用 Pydantic 模型来指定工具接受什么参数、每个参数的类型、描述等，

有了输入模式，LLM 可以：

理解每个参数的含义
知道需要提供哪些参数
自动从用户问题中提取参数值

pip install tldextract

from pydantic import BaseModel, Field, model_validator
from langchain_deepseek import ChatDeepSeek
from langchain_community.tools import RequestsGetTool
from langchain_community.utilities.requests import TextRequestsWrapper
from langchain.agents import initialize_agent, AgentType
import tldextract
from dotenv import load_dotenv
from urllib.parse import urlparse

load_dotenv()

llm = ChatDeepSeek(model="deepseek-chat")

# 定义域名访问白名单，只允许访问这些域名的网站
_APPRES_DOMAINS = {"google", "wikipedia", "github", "stackoverflow","baidu"}

# 定义工具输入的数据模型，使用Pydantic进行数据验证
class Toolinput(BaseModel):
    # url字段，必填
    url: str = Field(..., description="The URL to access")
    
    # 模型验证器，在实例化后执行，用于验证URL
    @model_validator(mode='after')
    def validate_url(self) -> 'Toolinput':
        # 解析URL，检查是否有协议头（如http://或https://）
        parsed = urlparse(self.url)
        if not parsed.scheme:
            # 如果没有协议头，自动添加https://
            self.url = "https://" + self.url

        # 使用tldextract提取域名
        domain = tldextract.extract(self.url).domain
        # 检查域名是否在白名单中
        if domain not in _APPRES_DOMAINS:
            raise ValueError(f"Access to {self.url} is not allowed")
        print(f"Accessing {self.url}")
        return self

# 创建请求工具实例
tools = RequestsGetTool(
    # 指定输入参数的数据模型
    args_schema=Toolinput,
    # 允许危险请求（网络请求可能带来安全风险，需要显式启用）
    allow_dangerous_requests=True,
    # 请求包装器，用于处理HTTP请求
    requests_wrapper=TextRequestsWrapper()
)

# 初始化代理（智能体），将工具和语言模型结合起来
agent = initialize_agent(
    [tools],
    llm,
    # 代理类型：零样本反应描述，适用于简单任务
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
)

print(agent.run("www.google.com的主题是什么"))

人为参与验证Human-in-the-loop Tool Validation

人为参与验证是一种开发与测试方法，即在工具或系统（尤其是AI或自动化工具）的关键环节中，主动引入人类判断来检查、验证或修正输出结果，以确保其正确性、可靠性及安全性。

from langchain.callbacks import HumanApprovalCallbackHandler
from langchain_community.tools import ShellTool

# 使用ShellTool进行人工判断
tool = ShellTool(callbacks=[HumanApprovalCallbackHandler()])
print(tool.run("pwd"))

# Do you approve of the following input? Anything except 'Y'/'Yes' (case-insensitive) will be treated as a no.
# y
# Executing command:
#  pwd
# /root/python-langchain

def _should_check(serialized_obj: dict) -> bool:
    # 表示仅对Shell工具进行验证
    return serialized_obj.get("name") == "terminal"

def _approve(_input: str) -> bool:
    # 白名单机制：自动批准安全命令pwd
    if _input == "pwd":
        return True
    msg = (
        "你是否批准以下输入？"
        "除了Y和Yes之外的任何内容都将视为不批准"
    )
    msg += "\n\n" + _input + "\n"
    resp = input(msg)
    # 返回True仅当用户明确输入"y"或"yes"（不区分大小写）
    return resp.lower() in ("yes","y")

callbacks = [
    HumanApprovalCallbackHandler(
        should_check=_should_check,  # 绑定条件检查函数
        approve=_approve             # 绑定批准逻辑函数
    )
]

from langchain_deepseek import ChatDeepSeek
from langchain.agents import initialize_agent,AgentType
from langchain_community.agent_toolkits.load_tools import load_tools
from dotenv import load_dotenv
load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat")
tools = load_tools(
    ["llm-math", "terminal"],  # 工具列表：数学工具和Shell终端
    llm=llm,
    allow_dangerous_tools=True  # 必须显式启用才能使用危险工具
)
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, # 代理类型：零样本推理
)

print(agent.invoke("echo 'hello world' in terminal",callbacks=callbacks))

# {'input': "echo 'hello world' in terminal", 'output': 'The command `echo \'hello world\'` was executed and the output was "hello world".'}