LangChain学习日志: Agent

尝试使用LangChain的Agents功能。

Intro

LangChain 官方文档的 Core components 第一项就是 Agents ，我们先来看看官方是如何介绍的：

Agents combine language models with tools to create systems that can reason about tasks, decide which tools to use, and iteratively work towards solutions.

create_agent provides a production-ready agent implementation.

An LLM Agent runs tools in a loop to achieve a goal. An agent runs until a stop condition is met - i.e., when the model emits a final output or an iteration limit is reached.

这段话告诉我们三个信息：

Agent 是将语言模型（LLM）与工具（Tools）结合的系统。它的特点是具备推理能力，能自主决定该用什么工具，并通过反复迭代来解决任务。
可以使用 create_agent 函数实现
Agent 的运行是一个循环过程：它会不断运行工具，直到满足停止条件为止。

并且给出了简单的工作流程：

Core components

Model

Model 就是 Agent 的推理引擎，负责思考、决策和生成回复。
对于 LLM 模型的调用分为两种模式：

Static Model：最常用的方式。在创建 Agent 时就固定好（例如指定用 gpt-4o），整个运行过程中不会变。
Dynamic Model：一种高级用法，可以在运行时根据上下文动态选择模型。比如在简单的问题上使用便宜的模型。

Tools

Tools 就是 Agent 与外部世界交互的工具。通常是标准的 Python 函数，加上 @tool 装饰器。根据文档的示例，我们可以通过create_agent简单定义一个 Tool ：

from langchain.tools import tool
from langchain.agents import create_agent


@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

@tool
def get_weather(location: str) -> str:
    """Get weather information for a location."""
    return f"Weather in {location}: Sunny, 72°F"

agent = create_agent(model, tools=[search, get_weather])

而如果要自定义Tool错误的处理方式，就使用 @wrap_tool_call 装饰器创建中间件：

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage


@wrap_tool_call
def handle_tool_errors(request, handler):
    """Handle tool execution errors with custom messages."""
    try:
        return handler(request)
    except Exception as e:
        # Return a custom error message to the model
        return ToolMessage(
            content=f"Tool error: Please check your input and try again. ({str(e)})",
            tool_call_id=request.tool_call["id"]
        )

agent = create_agent(
    model="gpt-4o",
    tools=[search, get_weather],
    middleware=[handle_tool_errors]
)

System Prompt

顾名思义，System Prompt 就是系统层面的提示词，用于设定 Agent 的行为规范和角色。
最简单的定义方式就是在构建agent示例时传入：

agent = create_agent(
    model,
    tools,
    system_prompt="You are a helpful assistant. Be concise and accurate."
)

这是传入Str，复杂一些，可以传入LangChain的 SystemMessage.

from langchain.agents import create_agent
from langchain.messages import SystemMessage, HumanMessage

literary_agent = create_agent(
    model="anthropic:claude-sonnet-4-5",
    system_prompt=SystemMessage(
        content=[
            {
                "type": "text",
                "text": "You are an AI assistant tasked with analyzing literary works.",
            },
            {
                "type": "text",
                "text": "<the entire contents of 'Pride and Prejudice'>",
                "cache_control": {"type": "ephemeral"}
            }
        ]
    )
)

result = literary_agent.invoke(
    {"messages": [HumanMessage("Analyze the major themes in 'Pride and Prejudice'.")]}
)

Invocation

启动 Agent。

Agent 的运行是基于 State (状态) 的更新，通常只需要给它传入一个新的消息（Message），它就会开始根据图（Graph）的逻辑运行，直到得出结果。比如：

1
2
3

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]}
)

除了以上基础的必要的步骤，LangChain 还为 Agents 提供了一些高级功能。我们简单了解一下：
Structured Output (结构化输出)
可以强制返回 JSON 数据（比如提取出的姓名、日期、订单号），方便程序直接后续处理。

Memory (记忆机制)

短期记忆： 通过 State 实现记住这轮对话里你刚才说了什么。
长期记忆： 通过数据库实现历史记忆。

Streaming (流式输出)
Agent 可以一边调用工具，一边就在屏幕上即时打印出它的思考过程。

Middleware (中间件)
它可以在 Agent 说话前、说话后，或者工具报错时工作。比如在工具报错时，自动让 Agent 重试，而不是直接让程序崩溃。

这些Agents的功能似乎封装的还不错，那我们来实际尝试一下。

实战

在实战中，我们需要实现一个日常小助手，他有以下两个小功能：

🤖Daily Assistant

WeatherTool: 查询某地天气

Loan Calculator: 计算等额本息

WeatherTool

这是一个天气查询工具。为了查询真实的天气，我们去 OpenWeatherMap 官网申请一个免费的 API Key，然后使用其api查询指定城市的天气。

import os
import requests
from typing import Type, Optional
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field

# 1. 定义输入参数的结构
class WeatherInput(BaseModel):
    city: str = Field(
        description="需要查询天气的城市英文名，例如 'Beijing', 'London', 'New York'"
    )

# 2. 定义工具类
class RealWeatherTool(BaseTool):
    name: str = "get_real_weather"
    description: str = (
        "获取指定城市的实时天气详情。包含温度、天气状况、湿度等信息。"
        "当用户询问'今天天气如何'或'下雨吗'时使用此工具。"
    )
    args_schema: Type[BaseModel] = WeatherInput
    api_key: str = os.getenv("OPENWEATHER_API_KEY", "******")

    def _run(self, city: str) -> str:
        base_url = "http://api.openweathermap.org/data/2.5/weather"
        params = {
            "q": city,
            "appid": self.api_key,
            "units": "metric",
            "lang": "zh_cn"
        }

        try:
            response = requests.get(base_url, params=params)
            response.raise_for_status()
            
            data = response.json()
            
            weather_desc = data["weather"][0]["description"]
            temp = data["main"]["temp"]
            humidity = data["main"]["humidity"]
            wind_speed = data["wind"]["speed"]
            
            return (
                f"【{city} 实时天气】\n"
                f"- 状况: {weather_desc}\n"
                f"- 温度: {temp}°C\n"
                f"- 湿度: {humidity}%\n"
                f"- 风速: {wind_speed} m/s"
            )
            
        except requests.exceptions.HTTPError as e:
            if response.status_code == 404:
                return f"Error: 找不到城市 '{city}'，请检查拼写。"
            elif response.status_code == 401:
                return "Error: API Key 无效。"
            return f"Error: 请求失败 ({str(e)})"
        except Exception as e:
            return f"Error: 查询过程中发生未知错误: {str(e)}"

在上述代码中，我们先定义了输入参数的结构WeatherInput，其实就是输入一个城市名字符串。随后，继承BaseTool类定义一个RealWeatherTool类，定义内部字段并重写_run方法。

和上一篇笔记一样，我们使用阿里的 qwen 模型，创建一个agent，并把自定义的tool交给他测试一下：

import os
from langchain_community.chat_models import ChatTongyi
from langchain.agents import create_agent
from tools.weathertool import RealWeatherTool
from langchain_core.messages import HumanMessage

os.environ["DASHSCOPE_API_KEY"] = "******"
llm = ChatTongyi(model_name="qwen-max")

weather_tool = RealWeatherTool()
tools = [weather_tool]

agent = create_agent(llm, tools=tools)

query = "帮我查查杭州的天气。"
result = agent.invoke({
    "messages": [HumanMessage(content=query)]
})

for message in result["messages"]:
    message.pretty_print()

我们打印完整的输出，记录了大模型调用工具的过程：

=============== Human Message ===============
帮我查查杭州的天气。
=============== Ai Message ================
Tool Calls:
  get_real_weather (call_9a7740f492704862b7ae77)
 Call ID: call_9a7740f492704862b7ae77
  Args:
    city: Hangzhou
================ Tool Message ===============
Name: get_real_weather
【Hangzhou 实时天气】
- 状况: 晴
- 温度: 8.95°C
- 湿度: 28%
- 风速: 3.94 m/s
================= Ai Message ===============
杭州现在的天气情况如下：
- 状况: 晴
- 温度: 8.95°C
- 湿度: 28%
- 风速: 3.94 m/s
天气晴朗，外出注意防晒哦！

Loan Calculator

这一部分我们做一个简单的贷款计算器，以实现精准的计算，这是 LLM 不具有的功能。

from typing import Type
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
import math

# 1. 定义输入模型
class LoanInput(BaseModel):
    principal: float = Field(description="贷款本金总额，单位为元。例如 1000000")
    annual_rate_percent: float = Field(description="年利率百分比。例如 3.5 代表 3.5%")
    years: int = Field(description="贷款期限（年）。例如 30")

# 2. 定义贷款计算工具
class LoanCalculatorTool(BaseTool):
    name: str = "calculate_loan_payment"
    description: str = (
        "专业的贷款计算器（等额本息方式）。"
        "当用户询问'房贷月供'、'车贷计算'、'分期付款利息'时使用此工具。"
        "必须使用此工具进行计算，禁止模型自行估算。"
    )
    args_schema: Type[BaseModel] = LoanInput

    def _run(self, principal: float, annual_rate_percent: float, years: int) -> str:
        try:
            # 数据预处理
            monthly_rate = annual_rate_percent / 100 / 12
            total_months = years * 12
            
            # 核心计算逻辑
            if monthly_rate == 0:
                monthly_payment = principal / total_months
            else:
                numerator = principal * monthly_rate * math.pow(1 + monthly_rate, total_months)
                denominator = math.pow(1 + monthly_rate, total_months) - 1
                monthly_payment = numerator / denominator

            # 计算总额
            total_payment = monthly_payment * total_months
            total_interest = total_payment - principal

            return (
                f" 贷款本金: {principal:,.2f} 元\n"
                f" 年利率: {annual_rate_percent}%\n"
                f" 贷款期限: {years} 年 ({total_months} 期)\n"
                f" 每月月供: {monthly_payment:,.2f} 元\n"
                f" 总利息: {total_interest:,.2f} 元\n"
                f" 本息合计: {total_payment:,.2f} 元"
            )
        except Exception as e:
            return f"计算出错: {str(e)}"

只要在调用时示例化并加入 Tool 列表即可：

1 2	calculator_tool = LoanCalculatorTool() tools = [weather_tool, calculator_tool]

Test

我们对上述 Agent 进行一个简单的测试：

================ Human Message ===============

贷 100 万，30 年，利率 3.25%，每月月供多少？另外，北京今天天气如何？
================ Ai Message ===============
Tool Calls:
  calculate_loan_payment (call_2f9e1a62ae0f42cab0eabd)
 Call ID: call_2f9e1a62ae0f42cab0eabd
  Args:
    principal: 1000000
    annual_rate_percent: 3.25
    years: 30
  get_real_weather (call_05f08c54e13747b4b6513d)
 Call ID: call_05f08c54e13747b4b6513d
  Args:
    city: Beijing
================ Tool Message ================
Name: calculate_loan_payment

 贷款本金: 1,000,000.00 元
 年利率: 3.25%
 贷款期限: 30 年 (360 期)
 每月月供: 4,352.06 元
 总利息: 566,742.75 元
 本息合计: 1,566,742.75 元
================ Tool Message ================
Name: get_real_weather

【Beijing 实时天气】
- 状况: 晴
- 温度: -4.06°C
- 湿度: 17%
- 风速: 7.74 m/s
================= Ai Message ================

您的贷款每月月供为 4,352.06 元。下面是详细信息：
- 贷款本金: 1,000,000.00 元
- 年利率: 3.25%
- 贷款期限: 30 年 (360 期)
- 总利息: 566,742.75 元
- 本息合计: 1,566,742.75 元

北京今天的实时天气情况如下：
- 状况: 晴
- 温度: -4.06°C
- 湿度: 17%
- 风速: 7.74 m/s

请注意保暖，外出时可能需要携带外套。