LangChain Tools — 工具集成

内置工具

from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

search = DuckDuckGoSearchRun()
wiki = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(lang="zh"))

print(search.run("2024年中国GDP增速"))
print(wiki.run("巴塞尔协议"))

自定义工具

from langchain_core.tools import tool
from typing import Optional

@tool
def query_loan_status(loan_id: str) -> dict:
    """查询贷款申请状态。输入贷款申请编号，返回当前审批状态和进度。"""
    # 实际业务逻辑
    mock_data = {
        "L001": {"status": "审批中", "progress": "风控审核", "days": 2},
        "L002": {"status": "已批准", "progress": "放款中", "amount": 500000},
    }
    return mock_data.get(loan_id, {"status": "未找到", "loan_id": loan_id})

@tool
def calculate_loan_rate(
    amount: float,
    term_months: int,
    credit_score: int,
    loan_type: str = "企业贷款"
) -> dict:
    """
    计算贷款利率和月供。
    
    Args:
        amount: 贷款金额（元）
        term_months: 贷款期限（月）
        credit_score: 信用评分 300-850
        loan_type: 贷款类型
    
    Returns:
        包含年利率、月供、总利息的字典
    """
    base_rate = 0.045  # 基准利率 4.5%
    
    # 根据信用评分调整
    if credit_score >= 750:
        rate = base_rate - 0.005
    elif credit_score >= 650:
        rate = base_rate
    else:
        rate = base_rate + 0.01
    
    monthly_rate = rate / 12
    monthly_payment = amount * monthly_rate / (1 - (1 + monthly_rate) ** (-term_months))
    total_interest = monthly_payment * term_months - amount
    
    return {
        "annual_rate": f"{rate*100:.2f}%",
        "monthly_payment": round(monthly_payment, 2),
        "total_interest": round(total_interest, 2),
        "total_payment": round(monthly_payment * term_months, 2)
    }

# 测试工具
print(query_loan_status.invoke({"loan_id": "L001"}))
print(calculate_loan_rate.invoke({
    "amount": 500000, "term_months": 36, "credit_score": 720
}))

ReAct Agent

from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub

tools = [query_loan_status, calculate_loan_rate, DuckDuckGoSearchRun()]

# 使用 hub 上的 ReAct prompt
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,
    handle_parsing_errors=True
)

result = agent_executor.invoke({
    "input": "帮我查询贷款 L001 的状态，并计算如果批准50万、36期的月供是多少（信用分720）"
})
print(result["output"])

Tool Calling Agent（推荐）

from langchain.agents import create_tool_calling_agent

# 更稳定的工具调用 agent（基于 Function Calling）
agent = create_tool_calling_agent(llm, tools, prompt=ChatPromptTemplate.from_messages([
    ("system", "你是专业的贷款助手，使用工具回答用户问题"),
    MessagesPlaceholder("chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
]))

executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "查询 L002 的贷款状态"})

工具错误处理

from langchain_core.tools import ToolException

@tool
def get_customer_info(customer_id: str) -> dict:
    """获取客户信息"""
    if not customer_id.startswith("C"):
        raise ToolException(f"无效的客户ID格式: {customer_id}，应以 'C' 开头")
    
    # 模拟数据库查询
    return {"id": customer_id, "name": "张三", "credit_score": 720}

# 在 AgentExecutor 中启用错误处理
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors=True,  # 处理解析错误
    verbose=True
)