大模型应用开发-Langchain(V1-最新版)-上https://mp.weixin.qq.com/s?__biz=MzI4NDk5NDAzNQ==&mid=2247486106&idx=1&sn=9bd024c1fed37f77c4e5d87728a96853&scene=21&poc_token=HH3WSGmjAdZ-xGSsAhWbnt4W4r_1jdhHPO-Kgq6_

大模型应用开发-Langchain(V1-最新版)-中https://mp.weixin.qq.com/s?__biz=MzI4NDk5NDAzNQ==&mid=2247486583&idx=1&sn=ef21b4fef9006dd1473276746ff3dbd8&scene=21&poc_token=HLbXSGmj-1j5diXRiu5zL2SeuQQNijWuNlQktTXE

七 工具

7.1 langchain_core中的Tools

7.2 @tool

这是最常用的方式，通过@tool装饰器命令将我们定义的普通函数包装成大模型可以调用的工具

import os
from typing import Any, Dict, List
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
 
# 加载环境变量
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
 
# ====== 1. 定义工具 ======
@tool
def multiply(a: int, b: int) -> int:
    """将两个整数相乘"""
    return a * b
 
@tool
def get_weather(city: str) -> str:
    """模拟获取指定城市的天气"""
    weather_map = {"北京": "晴", "上海": "多云", "广州": "雷阵雨"}
    return f"{city} 的天气是：{weather_map.get(city, '未知')}"
# 工具列表
tools = [multiply, get_weather]
 
# ====== 2. 初始化模型并绑定工具 ======
#绑定千问大模型 并开启工具调用功能
llm = ChatTongyi(
    model="qwen-max",
    temperature=0,
    model_kwargs={"tool_choice": "auto"}  # 启用自动工具选择
)
# 绑定工具（关键！）
llm_with_tools = llm.bind_tools(tools)
 
# ====== 3. 工具查找字典（用于执行）======
tool_map = {tool.name: tool for tool in tools}
 
# ====== 4. 手动实现一轮工具调用流程 ======
def call_with_tools(user_input: str) -> str:
    # 第一步：发送用户消息，获取模型响应（可能包含 tool_calls）
    messages = [HumanMessage(content=user_input)]
    ai_msg = llm_with_tools.invoke(messages)
 
    # 检查是否需要调用工具
    if hasattr(ai_msg, 'tool_calls') and ai_msg.tool_calls:
        print(f"🔧 模型请求调用工具: {ai_msg.tool_calls}")
 
        # 执行所有工具调用
        tool_messages = []
        for tool_call in ai_msg.tool_calls:
            #获取工具信息
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            tool_func = tool_map[tool_name]
            # 执行工具
            result = tool_func.invoke(tool_args)
            print(f"  → 执行 {tool_name}({tool_args}) = {result}")
 
            # 构造 ToolMessage（必须包含 tool_call_id）
            tool_messages.append(
                ToolMessage(
                    content=str(result),
                    tool_call_id=tool_call["id"]
                )
            )
 
        # 第二步：将工具结果发回模型，生成最终回答
        messages.extend([ai_msg] + tool_messages)
        final_response = llm.invoke(messages)
        return final_response.content
    else:
        # 不需要工具，直接返回
        return ai_msg.content
 
# ====== 5. 测试 ======
if __name__ == "__main__":
    test_inputs = [
        "3 乘以 8 等于多少？",
        "上海今天天气怎么样？",
        "你好！"
    ]
 
    for question in test_inputs:
        print(f"\n{'='*50}")
        print(f"👤 用户: {question}")
        answer = call_with_tools(question)
        print(f"🤖 AI: {answer}")

7.3 StructuredTool

对于输入过于复杂的函数，可以借助pydantic定义输入参数结构+StructuredTool的方式将函数封装为大模型可以调用的工具

import os
import asyncio
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
from dotenv import load_dotenv
from langchain_core.tools import StructuredTool
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
# 加载环境变量
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
# ====== 1. 定义 Pydantic 输入 Schema ======
class MultiplyInput(BaseModel):
    # Field(...)表示为必须字段 description为参数描述
    a: int = Field(..., description="第一个整数")
    b: int = Field(..., description="第二个整数")
class WeatherInput(BaseModel):
    city: str = Field(..., description="城市名称，例如 '北京'、'上海'")
# ====== 2. 定义同步/异步函数 ======
def multiply_sync(a: int, b: int) -> int:
    """同步乘法函数"""
    return a * b
async def get_weather_async(city: str) -> str:
    """异步天气查询（模拟网络请求）"""
    await asyncio.sleep(0.1)  # 模拟延迟
    weather_map = {"北京": "晴", "上海": "多云", "广州": "雷阵雨"}
    return f"{city} 的天气是：{weather_map.get(city, '未知')}"
# ====== 3. 创建 StructuredTool 结构化工具对象 ======
multiply_tool = StructuredTool.from_function(
    # 指定函数
    func=multiply_sync,
    name="multiply_numbers",
    description="将两个整数相乘",
    args_schema=MultiplyInput,  # ← 关键：自定义参数结构
)
weather_tool = StructuredTool.from_function(
    # 同步占位符，因为指定了coroutine此函数变为异步函数 此处占位即可 实际不会调用这里
    func=lambda x: x,
    # 指向自定义的异步函数 当此工具调用时调用此函数
    coroutine=get_weather_async,  # ← 异步实现
    name="get_weather",
    description="获取指定城市的当前天气",
    # 指定参数结构
    args_schema=WeatherInput,
)
# 工具列表
tools = [multiply_tool, weather_tool]
# 工具名称-工具映射
tool_map = {tool.name: tool for tool in tools}
# ====== 4. 初始化模型并绑定工具 ======
llm = ChatTongyi(
    model="qwen-max",
    temperature=0,
    model_kwargs={"tool_choice": "auto"}
)
llm_with_tools = llm.bind_tools(tools)
# ====== 5. 手动执行工具调用流程（支持异步）======
async def call_with_structured_tools(user_input: str) -> str:
    # 第一步：发送用户消息，获取模型响应
    messages = [HumanMessage(content=user_input)]
    ai_msg = llm_with_tools.invoke(messages)
    if hasattr(ai_msg, 'tool_calls') and ai_msg.tool_calls:
        print(f"🔧 模型请求调用工具: {ai_msg.tool_calls}")
        tool_messages = []
        for tool_call in ai_msg.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            tool_obj = tool_map[tool_name]
            # 判断是否为异步工具
            if tool_obj.coroutine:
                # 异步执行
                result = await tool_obj.arun(tool_args)
            else:
                result = tool_obj.run(tool_args)
            print(f"  → 执行 {tool_name}({tool_args}) = {result}")
            tool_messages.append(
                ToolMessage(
                    content=str(result),
                    tool_call_id=tool_call["id"]
                )
            )
        # 第二步：将工具结果返回模型，生成最终回答
        messages.extend([ai_msg] + tool_messages)
        final_response = llm.invoke(messages)
        return final_response.content
    else:
        return ai_msg.content
# ====== 6. 测试函数 ======
async def main():
    test_inputs = [
        "计算 6 乘以 9 等于多少？",
        "查一下上海的天气",
        "你好呀！"
    ]
    for question in test_inputs:
        print(f"\n{'=' * 50}")
        print(f"👤 用户: {question}")
        answer = await call_with_structured_tools(question)
        print(f"🤖 AI: {answer}")
# ====== 运行 ======
if __name__ == "__main__":
    asyncio.run(main())

7.4 BaseTool

如果追求高定制工具，可以继承最基础的工具类BaseTool，然后自定义工具

import os
import asyncio
from typing import Optional, Type
from pydantic import BaseModel, Field
from dotenv import load_dotenv
from langchain_core.tools import BaseTool
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
# 加载环境变量
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
# ====== 1. 定义输入 Schema ======
class UnitConvertInput(BaseModel):
    value: float = Field(..., description="要转换的数值")
    from_unit: str = Field(..., description="原始单位，如 'C'（摄氏度）或 'F'（华氏度）")
    to_unit: str = Field(..., description="目标单位，如 'C' 或 'F'")
# ====== 2. 自定义 BaseTool 工具 ======
class TemperatureConverter(BaseTool):
    name: str = "convert_temperature"
    description: str = "将温度在摄氏度（C）和华氏度（F）之间相互转换"
    args_schema: Type[BaseModel] = UnitConvertInput  # ← 关键：指定输入结构
    def _run(self, value: float, from_unit: str, to_unit: str) -> str:
        """同步执行温度转换"""
        from_unit = from_unit.upper()
        to_unit = to_unit.upper()
        if from_unit == to_unit:
            return f"{value}°{from_unit}"
        elif from_unit == "C" and to_unit == "F":
            result = value * 9 / 5 + 32
        elif from_unit == "F" and to_unit == "C":
            result = (value - 32) * 5 / 9
        else:
            return "仅支持 C 和 F 之间的转换。"
        return f"{value}°{from_unit} = {result:.2f}°{to_unit}"
    async def _arun(self, value: float, from_unit: str, to_unit: str) -> str:
        """异步版本（模拟耗时操作）"""
        await asyncio.sleep(0.05)  # 模拟 I/O 延迟
        return self._run(value, from_unit, to_unit)
# ====== 3. 创建工具实例 ======
converter_tool = TemperatureConverter()
tools = [converter_tool]
tool_map = {tool.name: tool for tool in tools}
# ====== 4. 初始化模型并绑定工具 ======
llm = ChatTongyi(
    model="qwen-max",
    temperature=0,
    model_kwargs={"tool_choice": "auto"}
)
llm_with_tools = llm.bind_tools(tools)
# ====== 5. 手动执行工具调用流程（支持异步）======
async def call_with_basetool(user_input: str) -> str:
    messages = [HumanMessage(content=user_input)]
    ai_msg = llm_with_tools.invoke(messages)
    if hasattr(ai_msg, 'tool_calls') and ai_msg.tool_calls:
        print(f"🔧 模型请求调用工具: {ai_msg.tool_calls}")
        tool_messages = []
        for tool_call in ai_msg.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            tool_obj = tool_map[tool_name]
            # 判断是否为异步工具（这里我们统一用 arun 演示）
            result = await tool_obj.arun(tool_input=tool_args)
            print(f"  → 执行 {tool_name}({tool_args}) = {result}")
            tool_messages.append(
                ToolMessage(
                    content=str(result),
                    tool_call_id=tool_call["id"]
                )
            )
        # 将工具结果返回模型，生成最终回答
        messages.extend([ai_msg] + tool_messages)
        final_response = llm.invoke(messages)
        return final_response.content
    else:
        return ai_msg.content
# ====== 6. 测试 ======
async def main():
    test_inputs = [
        "把 25 摄氏度转成华氏度",
        "100°F 等于多少摄氏度？",
        "今天真热啊！"
    ]
    for question in test_inputs:
        print(f"\n{'='*50}")
        print(f"👤 用户: {question}")
        answer = await call_with_basetool(question)
        print(f"🤖 AI: {answer}")
if __name__ == "__main__":
    asyncio.run(main())

八 Agent

在 LangChain 1.x（特别是 1.1+） 中，Agent 的构建范式已彻底转向 显式控制流 + 原生工具调用（tool calling），不再依赖“黑盒式”的 AgentExecutor。

最新版Langchain 1.x中推荐使用LangGraph构建高度自由、可控的智能体，后续会单独出讲解langGraph的文章。

8.1 LangGraph实现

这是目前最新版Langchain推荐的构建智能体的方式，借助LangGraph构建工作流形成智能体，而且最新版langchain也是默认将LangGraph作为运行时，自身运行在langgraph上

import os
from typing import Annotated, Sequence, TypedDict
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
# ====== 1. 定义工具 ======
@tool
def multiply(a: int, b: int) -> int:
    """将两个整数相乘"""
    return a * b
tools = [multiply]
# ====== 2. 初始化模型（绑定工具）======
llm = ChatTongyi(model="qwen-max", temperature=0)
llm_with_tools = llm.bind_tools(tools)
# ====== 3. 定义状态图 ======
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], "聊天消息历史"]
    # 用于存储中间结果
    intermediate_steps: Annotated[list, "中间步骤结果"]
# 决策函数：是否继续调用工具？
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END
# 调用模型节点
def call_model(state: AgentState):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}
# 工具调用节点
def call_tools(state: AgentState):
    # 获取最后一个消息中的工具调用
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls
    
    # 执行所有工具调用
    results = []
    for tool_call in tool_calls:
        # 执行工具
        tool_result = multiply.invoke(tool_call["args"])
        results.append({
            "name": tool_call["name"],
            "args": tool_call["args"],
            "result": tool_result
        })
    
    # 创建一个包含工具执行结果的AI消息
    result_str = "\n".join([f"工具 {r['name']}({r['args']}) 的结果是: {r['result']}" for r in results])
    response_message = AIMessage(content=result_str)
    
    return {"messages": [response_message], "intermediate_steps": results}
# ====== 4. 构建 Graph ======
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tools)
# 设置入口
workflow.set_entry_point("agent")
# 条件边：根据模型输出决定下一步
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        END: END,
    },
)
# 工具执行完后返回 agent
workflow.add_edge("tools", "agent")
# 编译为可运行应用
app = workflow.compile()
# ====== 5. 测试 ======
if __name__ == "__main__":
    inputs = {"messages": [HumanMessage(content="7 乘以 6 等于多少？")], "intermediate_steps": []}
    final_answer = None
    for event in app.stream(inputs, stream_mode="values"):
        if "messages" in event and event["messages"]:
            last_message = event["messages"][-1]
            print(f"消息类型: {type(last_message).__name__}")
            if hasattr(last_message, 'content'):
                print(f"内容: {last_message.content}")
            if hasattr(last_message, 'tool_calls'):
                print(f"工具调用: {last_message.tool_calls}")
        # 保存最后一个消息作为最终答案
        if "messages" in event:
            final_answer = event["messages"][-1]
    if final_answer and hasattr(final_answer, 'content'):
        print("🤖 最终回答:", final_answer.content)
    else:
        print("🤖 最终回答: 未能生成答案")

8.2 手动实现智能体循环

手动实现LangGraph的执行循环过程

import os
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage, ToolMessage
 
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
 
 
@tool
def add(a: int, b: int) -> int:
    """将两个整数相加"""
    return a + b
 
 
tools = [add]
tool_map = {t.name: t for t in tools}
 
llm = ChatTongyi(model="qwen-max")
llm_with_tools = llm.bind_tools(tools)
 
 
def run_agent(user_input: str, max_turns=3) -> str:
    messages = [HumanMessage(content=user_input)]
 
    for _ in range(max_turns):
        ai_msg = llm_with_tools.invoke(messages)
        messages.append(ai_msg)
 
        # 检查是否需要调用工具
        if not hasattr(ai_msg, 'tool_calls') or not ai_msg.tool_calls:
            return ai_msg.content  # 直接返回答案
 
        # 执行所有工具调用
        for tc in ai_msg.tool_calls:
            result = tool_map[tc["name"]].invoke(tc["args"])
            messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
 
    return "达到最大推理步数。"
 
 
# 测试
print(run_agent("3 加 9 等于多少？"))

8.3 基于Prompt方式构建ReAct智能体

这是以往构建智能体的方式，智能体的执行准确性、结果解析等都依赖人工编写的Prompt

import os
import time
import requests
from typing import List, Any, Dict, Tuple
from dotenv import load_dotenv
from urllib.parse import urljoin, urlparse
# —————— 1. 初始化模型 ——————
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
from langchain_community.chat_models import ChatTongyi
llm = ChatTongyi(
    model="qwen-max",
    temperature=0,
    model_kwargs={"tool_choice": "auto"}
)
# —————— 2. 安全可靠的 Bing 中国版搜索工具 ——————
from langchain_core.tools import Tool
from bs4 import BeautifulSoup
from langchain_core.output_parsers import BaseOutputParser
def enhanced_bing_search(query: str, max_results: int = 3) -> str:
    """
    Bing 搜索工具
    返回前 N 条搜索结果及对应链接的详细内容。
    """
    # 从输入的搜索关键词中提取关键词列表
    class KeywordExtractOutputParser(BaseOutputParser[List[str]]):
        """解析关键词抽取结果的输出解析器"""
        def parse(self, text: str) -> List[str]:
            # 清理文本，移除可能的前缀
            cleaned_text = text.strip()
            # 如果有"关键词："之类的前缀，移除它
            if "：" in cleaned_text:
                cleaned_text = cleaned_text.split("：", 1)[1].strip()
            # 处理多种分隔符
            # 替换中文逗号和顿号为英文逗号
            cleaned_text = cleaned_text.replace("，", ",").replace("、", ",").replace(";", ",").replace("；", ",")
            # 按逗号分割并清理每个关键词
            keywords = [kw.strip() for kw in cleaned_text.split(",") if kw.strip()]
            # 如果没有逗号分隔，尝试按行分割
            if len(keywords) <= 1 and "\n" in cleaned_text:
                keywords = [kw.strip() for kw in cleaned_text.split("\n") if kw.strip()]
            # 如果还是没有分割开，尝试按空格分割（但只在明显用空格分隔的情况下）
            if len(keywords) <= 1 and " " in cleaned_text and len(cleaned_text.split()) > 1:
                keywords = [kw.strip() for kw in cleaned_text.split(" ") if kw.strip()]
            return keywords
    # 利用大模型从输入的query中抽取出关键词
    def extract_keywords(query: str) -> List[str]:
        # 创建关键词抽取提示词
        keyword_prompt = ChatPromptTemplate.from_template(
            "请从以下内容中抽取关键词，用英文逗号分隔返回，不要包含其他内容：\n{query}"
        )
        # 创建关键词抽取链
        keyword_chain = keyword_prompt | llm | KeywordExtractOutputParser()
        # 执行关键词抽取
        try:
            keywords = keyword_chain.invoke({"query": query})
            return keywords if isinstance(keywords, list) else [keywords]
        except Exception as e:
            # 如果解析失败，回退到简单的处理方式
            print(f"关键词抽取失败: {e}")
            return [query]  # 返回原始查询作为唯一关键词
    #抽取搜索查询中包含的关键词
    query_keywords = extract_keywords(query)
    try:
        # 搜索工具地址
        url = "https://cn.bing.com/search"
        # 搜索参数
        params = {"q": query}
        # 伪造浏览器头,发起请求
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36",
            "Accept-Language": "zh-CN,zh;q=0.9",
        }
        # 执行初始搜索
        response = requests.get(url, params=params, headers=headers, timeout=8)
        # 检查返回状态 如果搜索失败会抛出异常 退出
        response.raise_for_status()
        # 解析搜索结果内容
        response.encoding = 'utf-8'
        soup = BeautifulSoup(response.text, "html.parser")
        results = []
        # 提取搜索结果
        for item in soup.select("li.b_algo")[:max_results]:
            title_tag = item.select_one("h2 a")
            desc_tag = item.select_one("div.b_caption p")
            link_tag = item.select_one("h2 a")
            if title_tag and desc_tag and link_tag:
                title = title_tag.get_text(strip=True)
                desc = desc_tag.get_text(strip=True)
                link = link_tag.get('href')
                if title and desc and link:
                    results.append({
                        "title": title,
                        "desc": desc,
                        "link": link
                    })
        # 对链接进行深度抓取
        detailed_results = []
        deep_fetch_count = 0
        for result in results:
            if deep_fetch_count >= max_results:  # 限制深度抓取的数量
                # 不进行深度抓取，直接使用搜索结果摘要中的信息
                detailed_results.append(f"【{result['title']}】{result['desc']}")
                continue
            try:
                # 检查链接是否值得深度抓取
                if any(keyword in result['title'] + result['desc'] for keyword in query_keywords):
                    # 获取页面内容
                    page_response = requests.get(result["link"], headers=headers, timeout=5)
                    page_response.raise_for_status()
                    page_response.encoding = 'utf-8'
                    page_soup = BeautifulSoup(page_response.text, "html.parser")
                    # 移除脚本和样式元素
                    for script in page_soup(["script", "style"]):
                        script.decompose()
                    # 提取主要内容
                    content = ""
                    # 尝试查找文章正文区域
                    content_areas = page_soup.select("article, .article, .content, .post-content, .entry-content, main")
                    if content_areas:
                        content = content_areas[0].get_text(strip=True)
                    else:
                        # 如果找不到明确的内容区域，就使用整个body的文本
                        body = page_soup.select_one("body")
                        if body:
                            content = body.get_text(strip=True)
                    # 清理和截取内容
                    content = ' '.join(content.split())[:1000]  # 限制长度并清理多余空白
                    detailed_results.append(f"【{result['title']}】{result['desc']}\n详细信息: {content}")
                    deep_fetch_count += 1
                else:
                    # 不进行深度抓取，直接使用摘要
                    detailed_results.append(f"【{result['title']}】{result['desc']}")
            except Exception:
                # 如果无法获取详细内容，就只使用摘要
                detailed_results.append(f"【{result['title']}】{result['desc']}")
        return "\n\n".join(detailed_results) if detailed_results else "未找到相关结果。"
    except Exception as e:
        return f"Bing 搜索失败：{str(e)}"
# 搜索工具定义
search_tool = Tool(
    name="必应搜索",
    func=enhanced_bing_search,
    description="用于在互联网上搜索最新信息，例如新闻、天气、事件、百科等。输入应为清晰的问题或关键词，如“今天 北京 天气”或“2024年诺贝尔奖得主都有谁？”。"
)
# —————— 3. 计算器工具 ——————
def safe_python_math(expr: str) -> str:
    """安全执行简单数学表达式"""
    try:
        allowed = set("0123456789+-*/(). ")
        if not all(c in allowed for c in expr):
            return "错误：表达式包含非法字符"
        # 使用eval计算expr表达式
        # 为了防止用户执行恶意代码，我们限制了__builtins__，情况所有内置函数
        # 最后{}清空所有局部变量
        result = eval(expr, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"计算出错：{e}"
math_tool = Tool(
    name="计算器",
    func=safe_python_math,
    description="用于解答数学问题。输入应为有效的算术表达式，例如 '2 + 3 * 4' 或 '(100 - 20) / 4'。"
)
# 生成智能体可用的工具列表
tools = [search_tool, math_tool]
# —————— 4. 中文 ReAct 提示词 ——————
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import BaseOutputParser
from typing import List
# 创建关键词抽取的输出解析器
class KeywordExtractOutputParser(BaseOutputParser[List[str]]):
    """解析关键词抽取结果的输出解析器"""
    def parse(self, text: str) -> List[str]:
        # 清理文本，移除可能的前缀
        cleaned_text = text.strip()
        # 如果有"关键词："之类的前缀，移除它
        if "：" in cleaned_text:
            cleaned_text = cleaned_text.split("：", 1)[1].strip()
        # 处理多种分隔符
        # 替换中文逗号和顿号为英文逗号
        cleaned_text = cleaned_text.replace("，", ",").replace("、", ",").replace(";", ",").replace("；", ",")
        # 按逗号分割并清理每个关键词
        keywords = [kw.strip() for kw in cleaned_text.split(",") if kw.strip()]
        # 如果没有逗号分隔，尝试按行分割
        if len(keywords) <= 1 and "\n" in cleaned_text:
            keywords = [kw.strip() for kw in cleaned_text.split("\n") if kw.strip()]
        # 如果还是没有分割开，尝试按空格分割（但只在明显用空格分隔的情况下）
        if len(keywords) <= 1 and " " in cleaned_text and len(cleaned_text.split()) > 1:
            keywords = [kw.strip() for kw in cleaned_text.split(" ") if kw.strip()]
        return keywords
# 定义一个查询分解为关键词的函数，利用大模型从输入的query中抽取出关键词
def extract_keywords(query: str) -> List[str]:
    # 创建关键词抽取提示词
    keyword_prompt = ChatPromptTemplate.from_template(
        "请从以下内容中抽取关键词，用英文逗号分隔返回，不要包含其他内容：\n{query}"
    )
    # 创建关键词抽取链
    keyword_chain = keyword_prompt | llm | KeywordExtractOutputParser()
    # 执行关键词抽取
    try:
        keywords = keyword_chain.invoke({"query": query})
        return keywords if isinstance(keywords, list) else [keywords]
    except Exception as e:
        # 如果解析失败，回退到简单的处理方式
        print(f"关键词抽取失败: {e}")
        return [query]  # 返回原始查询作为唯一关键词
# 构建智能体提示
react_prompt_template = (
    "你是一个智能助手，可以使用以下工具来回答问题：\n\n"
    "{tools}\n\n"
    "请严格按照以下格式进行推理和行动：\n\n"
    "问题：你需要回答的原始问题\n"
    "思考：你应该始终先思考下一步该做什么。对于实时信息查询（如新闻等），你应该考虑使用中文和英文关键词分别进行搜索，这样获取的信息会更全面。\n"
    "行动：要执行的操作，必须是以下之一：[{tool_names}]\n"
    "行动输入：传递给该操作的具体输入内容\n"
    "观察：操作返回的结果\n"
    "……（“思考/行动/行动输入/观察”可以重复多次”）\n"
    "思考：我现在已经知道最终答案了\n"
    "最终答案：对原始问题的完整回答\n\n"
    "特别注意：\n"
    "1. 对于实时信息查询，请尝试使用中英文关键词分别进行搜索,这样可以增加回答的准确性\n"
    "2. 如果初步搜索结果不明确，请尝试更具体的关键词进行搜索\n"
    "3. 对于数字计算类问题，请优先使用计算器工具\n"
    "4. 当搜索结果中包含详细信息时，请仔细阅读并提取关键信息\n\n"
    "开始！\n\n"
    "问题：{input}\n"
    "思考：{agent_scratchpad}"
)
prompt = ChatPromptTemplate.from_template(react_prompt_template)
# 合并工具描述
def render_tool_description(tools: List[Tool]) -> str:
    return "\n".join([f"{tool.name}：{tool.description}" for tool in tools])
prompt = prompt.partial(
    tools=render_tool_description(tools),
    tool_names="、".join([t.name for t in tools])
)
# —————— 5. 中文输出解析器 ——————
from langchain_core.output_parsers import BaseOutputParser
from langchain_core.agents import AgentAction, AgentFinish
import re
class ChineseReActOutputParser(BaseOutputParser):
    def parse(self, text: str) -> Any:
        if "最终答案：" in text:
            output = text.split("最终答案：", 1)[-1].strip()
            return AgentFinish(return_values={"output": output}, log=text)
        # 解析智能体要调用的函数 以及 输入的参数信息
        action_match = re.search(r"行动：\s*(.+?)\n", text, re.DOTALL)
        input_match = re.search(r"行动输入：\s*(.+)", text, re.DOTALL)
        if action_match and input_match:
            action = action_match.group(1).strip()
            action_input = input_match.group(1).strip()
            return AgentAction(tool=action, tool_input=action_input, log=text)
        # 容错：若格式混乱，当作最终回答
        return AgentFinish(return_values={"output": text}, log=text)
output_parser = ChineseReActOutputParser()
# —————— 6. 构建 Agent 链 ——————
from langchain_core.runnables import RunnableLambda
from langchain_core.agents import AgentFinish
from langchain_classic.agents.format_scratchpad import format_log_to_str
# 设置llm生成'\n观察'后就停止生成
llm_with_stop = llm.bind(stop=["\n观察"])
# 构建ReAct智能体
agent_chain = (
        {
            "input": lambda x: x["input"],
            # 将智能体执行的中间步骤格式化为字符串
            "agent_scratchpad": lambda x: format_log_to_str(x["intermediate_steps"]),
        }
        | prompt
        | llm_with_stop
        | output_parser
)
# —————— 7. 创建 AgentExecutor ——————
from langchain_classic.agents import AgentExecutor
# 创建一个AgentExecutor
agent_executor = AgentExecutor(
    agent=agent_chain,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=5,
    return_intermediate_steps=False,
)
# —————— 8. 测试运行 ——————
if __name__ == "__main__":
    # # 测试关键词抽取功能
    # print("测试关键词抽取功能:")
    # test_queries = [
    #     "2024年诺贝尔物理学奖得主是谁？",
    #     "今天北京和上海的天气怎么样？",
    #     "123乘以456等于多少？"
    # ]
    #
    # for query in test_queries:
    #     keywords = extract_keywords(query)
    #     print(f"查询: {query}")
    #     print(f"关键词: {keywords}")
    #     print("-" * 50)
    # 原有的测试用例
    test_cases = [
        "今天哈尔滨和北京的天气怎么样？",
        "123乘以456等于多少？",
        "今夕是何年",
    ]
    for i, question in enumerate(test_cases, 1):
        print(f"\n{'=' * 50}")
        print(f"测试用例 {i}: {question}")
        print('=' * 50)
        try:
            result = agent_executor.invoke({"input": question})
            print(f"\n✅ 答案：{result['output']}")
        except Exception as e:
            print(f"❌ 执行出错：{e}")
        # 添加间隔，避免请求过快
        if i < len(test_cases):
            import time
 
            time.sleep(2)

九 回调处理

方式2就是旧版本基于回调handler的方式，通过指定不同阶段的handler函数进行回调处理

所有可拦截进行回调的事件如下：

9.1 事件流回调方式

最新版langhain推荐的回调处理方式，基于事件流处理

import os
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage, ToolMessage, AIMessage
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
@tool
def square(x: int) -> int:
    """计算平方"""
    return x * x
tools = [square]
tool_map = {t.name: t for t in tools}
llm = ChatTongyi(model="qwen-max", temperature=0)
llm_with_tools = llm.bind_tools(tools)
async def run_agent_with_events(user_input: str):
    messages = [HumanMessage(content=user_input)]
    # === 第一步：让模型决定是否调用工具 ===
    print("【第1轮】模型推理中...")
    async for event in llm_with_tools.astream_events(
            messages,
            version="v2",
            include_names=["ChatTongyi"]
    ):
        event_type = event["event"]
        if event_type == "on_chat_model_stream":
            token = event["data"]["chunk"].content
            if token:
                print(f"🔤 LLM Token: {repr(token)}")
        elif event_type == "on_chat_model_end":
            ai_msg = event["data"]["output"]
            messages.append(ai_msg)
            print(f"\n📨 模型返回: {ai_msg}")
            # 检查是否有工具调用
            if hasattr(ai_msg, 'tool_calls') and ai_msg.tool_calls:
                print("\n【执行工具】")
                for tool_call in ai_msg.tool_calls:
                    tool_name = tool_call["name"]
                    tool_args = tool_call["args"]
                    tool_func = tool_map[tool_name]
                    # 执行工具
                    result = tool_func.invoke(tool_args)
                    print(f"✅ 执行 {tool_name}({tool_args}) = {result}")
                    # 构造 ToolMessage 并加入历史
                    tool_message = ToolMessage(
                        content=str(result),
                        tool_call_id=tool_call["id"]
                    )
                    messages.append(tool_message)
            else:
                # 无工具调用，直接结束
                return
    # === 第二步：将工具结果回传，获取最终回答 ===
    if any(isinstance(m, ToolMessage) for m in messages):
        print("\n【第2轮】模型生成最终回答...")
        final_response = await llm_with_tools.ainvoke(messages)
        print(f"🎯 最终回答: {final_response.content}")
    else:
        print("⚠️ 未触发工具调用")
# 运行
import asyncio
 
asyncio.run(run_agent_with_events("4 的平方是多少？"))

9.2 handler函数回调方式

旧版基于handler函数处理回调的方式

import os
from dotenv import load_dotenv
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage
 
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
 
 
class MyCallbackHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"🚀 LLM 开始推理: {prompts[-1][:50]}...")
 
    def on_llm_end(self, response: LLMResult, **kwargs):
        content = response.generations[0][0].text
        print(f"🏁 LLM 结束: {content[:50]}...")
 
    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"🔧 工具启动: {serialized['name']}({input_str})")
 
    def on_tool_end(self, output, **kwargs):
        print(f"✅ 工具结束: {output}")
 
 
# 使用
llm = ChatTongyi(model="qwen-max")
handler = MyCallbackHandler()
 
response = llm.invoke(
    [HumanMessage(content="你好！")],
    config={"callbacks": [handler]}  # ← 通过 config 注入
)