Tool

from langchain_core.output_parsers import StrOutputParser
from langchain_experimental.utilities import PythonREPL
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_ollama import ChatOllama
from loguru import logger

def debug_print(x):
    """
    调试打印函数，用于在链式调用中输出中间结果

    参数:
        x: 任意类型的输入值，将被打印并原样返回

    返回值:
        与输入值x相同的值
    """
    logger.info(f"中间结果:{x}")
    return x

# 创建Python REPL工具实例，用于执行生成的Python代码
tool = PythonREPL()

# 初始化Ollama语言模型，使用qwen3:8b模型
llm = ChatOllama(model="qwen3:8b", reasoning=False)

# 定义聊天提示模板，包含系统指令和用户问题占位符
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你只返回纯净的 Python 代码，不要解释。代码必须是单行或多行 print。"),
        ("human", "{question}")
    ]
)

# 创建调试节点，用于在链式调用中插入调试信息输出
debug_node = RunnableLambda(debug_print)

# 创建字符串输出解析器，用于解析模型输出
parser = StrOutputParser()

# 构建处理链：提示模板 -> 语言模型 -> 调试输出 -> 输出解析 -> 代码执行
chain = prompt | llm | debug_node | parser | RunnableLambda(lambda code: tool.run(code))

# 执行链式调用，计算1到100的整数总和
result = chain.invoke({"question": "计算1到100的整数总和"})
logger.info(result)

2025-11-21 10:18:18.141 | INFO     | __main__:debug_print:19 - 中间结果:content='print(sum(range(1, 101)))' additional_kwargs={} response_metadata={'model': 'qwen3:8b', 'created_at': '2025-11-21T02:18:18.139783373Z', 'done': True, 'done_reason': 'stop', 'total_duration': 1183296308, 'load_duration': 24048613, 'prompt_eval_count': 53, 'prompt_eval_duration': 394815777, 'eval_count': 12, 'eval_duration': 761370757, 'model_name': 'qwen3:8b'} id='run--870a9cdc-3684-452b-9079-3769d5ae8ceb-0' usage_metadata={'input_tokens': 53, 'output_tokens': 12, 'total_tokens': 65}
Python REPL can execute arbitrary code. Use with caution.
2025-11-21 10:18:18.155 | INFO     | __main__:<module>:48 - 5050

from langchain_core.tools import tool
from loguru import logger
from pydantic import BaseModel, Field

class FieldInfo(BaseModel):
    """
    定义加法运算所需的参数信息
    """
    a: int = Field(description="第1个参数")
    b: int = Field(description="第2个参数")

# 通过args_schema定义参数信息，也可以定义name、description、return_direct参数
@tool(args_schema=FieldInfo)
def add_number(a: int, b: int) -> int:
    """
    两个整数相加
    """
    return a + b

# 打印工具的基本信息
logger.info(f"name = {add_number.name}")
logger.info(f"args = {add_number.args}")
logger.info(f"description = {add_number.description}")
logger.info(f"return_direct = {add_number.return_direct}")

# 调用工具执行加法运算
res = add_number.invoke({"a": 1, "b": 2})
logger.info(res)

2025-11-23 21:00:48.000 | INFO     | __main__:<module>:20 - name = add_number
2025-11-23 21:00:48.001 | INFO     | __main__:<module>:21 - args = {'a': {'description': '第1个参数', 'title': 'A', 'type': 'integer'}, 'b': {'description': '第2个参数', 'title': 'B', 'type': 'integer'}}
2025-11-23 21:00:48.001 | INFO     | __main__:<module>:22 - description = 两个整数相加
2025-11-23 21:00:48.001 | INFO     | __main__:<module>:23 - return_direct = False
2025-11-23 21:00:48.010 | INFO     | __main__:<module>:27 - 3

from langchain_core.tools import StructuredTool
from loguru import logger
from pydantic import BaseModel, Field

class FieldInfo(BaseModel):
    """
    定义加法运算所需的参数信息
    """
    a: int = Field(description="第1个参数")
    b: int = Field(description="第2个参数")

def add_number(a: int, b: int) -> int:
    """
    两个整数相加
    """
    return a + b

func = StructuredTool.from_function(
    func=add_number,
    name="Add",
    description="两个整数相加",
    args_schema=FieldInfo
)
logger.info(f"name = {func.name}")
logger.info(f"description = {func.description}")
logger.info(f"args = {func.args}")

res = func.invoke({"a": 1, "b": 2})
logger.info(res)

2025-11-23 21:08:43.228 | INFO     | __main__:<module>:27 - name = Add
2025-11-23 21:08:43.228 | INFO     | __main__:<module>:28 - description = 两个整数相加
2025-11-23 21:08:43.229 | INFO     | __main__:<module>:29 - args = {'a': {'description': '第1个参数', 'title': 'A', 'type': 'integer'}, 'b': {'description': '第2个参数', 'title': 'B', 'type': 'integer'}}
2025-11-23 21:08:43.237 | INFO     | __main__:<module>:32 - 3

from datetime import date
from langchain_core.tools import tool
from langchain_ollama import ChatOllama
from loguru import logger

@tool
def get_today() -> str:
    """
    获取当前系统日期

    Returns:
        str: 今天的日期字符串，格式为 yyyy-MM-dd
    """
    logger.info("执行工具：get_today")
    return date.today().isoformat()

# 设置本地模型，不使用深度思考
llm = ChatOllama(model="qwen3:14b", reasoning=False)
# 将工具绑定到语言模型
llm_with_tools = llm.bind_tools([get_today])
# 用户提问
question_list = ["你是谁？","今天是几号？"]
for question in question_list:
    logger.info(f"用户问题：{question}")
    # 调用语言模型处理用户问题
    ai_msg = llm_with_tools.invoke(question)
    logger.info(f"LLM回复：{ai_msg}")
    # 检查是否有工具调用
    if ai_msg.tool_calls:
        logger.info(ai_msg.tool_calls)
        # 获取第一个工具调用信息
        tool_call = ai_msg.tool_calls[0]
        # 执行对应的工具函数并获取结果
        tool_result = locals()[tool_call["name"]].invoke(tool_call["args"])
        logger.info(f"调用工具结果：{tool_result}")
    else:
        # 直接输出语言模型的回答
        logger.info(f"LLM 直接作答：{ai_msg.content}")

# cat .env
OPENWEATHER_API_KEY="XXXX"

from langchain_core.tools import tool
import json
import os
import httpx

@tool
def get_weather(loc):
    """
    查询即时天气函数

    :param loc: 必要参数，字符串类型，用于表示查询天气的具体城市名称。
                注意，中国的城市需要用对应城市的英文名称代替，例如如果需要查询北京市天气，
                则 loc 参数需要输入 'Beijing'。
    :return: OpenWeather API 查询即时天气的结果。具体 URL 请求地址为：
             https://api.openweathermap.org/data/2.5/weather。
             返回结果对象类型为解析之后的 JSON 格式对象，并用字符串形式进行表示，
             其中包含了全部重要的天气信息。
    """
    # Step 1. 构建请求 URL
    url = "https://api.openweathermap.org/data/2.5/weather"

    # Step 2. 设置查询参数，包括城市名、API Key、单位和语言
    params = {
        "q": loc,
        "appid": os.getenv("OPENWEATHER_API_KEY"),  # 从环境变量中读取 API Key
        "units": "metric",  # 使用摄氏度
        "lang": "zh_cn"  # 输出语言为简体中文
    }

    # Step 3. 发送 GET 请求获取天气数据
    response = httpx.get(url, params=params)

    # Step 4. 解析响应内容为 JSON 并序列化为字符串返回
    data = response.json()
    return json.dumps(data)

import dotenv
from langchain_core.output_parsers import JsonOutputKeyToolsParser, StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
from loguru import logger

from tools import get_weather

# 加载环境变量配置
dotenv.load_dotenv()

# 初始化大语言模型实例，使用 qwen3:14b 模型
llm = ChatOllama(model="qwen3:14b", reasoning=False)

# 将模型与工具绑定，使其能够调用 get_weather 工具
llm_with_tools = llm.bind_tools([get_weather])

# 创建解析器，用于提取工具调用结果中的 JSON 数据
parser = JsonOutputKeyToolsParser(key_name=get_weather.name, first_tool_only=True)

# 构建工具调用链：模型 -> 解析器 -> 调用天气工具
get_weather_chain = llm_with_tools | parser | get_weather
# print(get_weather_chain.invoke("你好， 请问北京的天气怎么样？"))
# 定义输出提示模板，将 JSON 天气数据转换为自然语言描述
output_prompt = PromptTemplate.from_template(
    """你将收到一段 JSON 格式的天气数据{weather_json}，请用简洁自然的方式将其转述给用户。
    以下是天气 JSON 数据：
    请将其转换为中文天气描述，例如：
    “北京现在天气：多云，气温 28℃，体感有点闷热(约 32℃)，湿度 75%，微风(东南风 2 米/秒)，能见度很好，大约 10 公里。建议穿短袖短裤。适合做户外运动。"
    """
)

# 创建字符串输出解析器
output_parser = StrOutputParser()

# 构建最终输出链：提示模板 -> 模型 -> 输出解析器
output_chain = output_prompt | llm | output_parser

# 构建完整的处理链：天气查询链 ->将天气数据包装为字典格式 -> 输出链
full_chain = get_weather_chain | (lambda x: {"weather_json": x}) | output_chain

# 执行完整链路，查询上海天气并打印结果
result = full_chain.invoke("请问上海今天的天气如何？")
logger.info(result)

2025-11-23 21:21:56.633 | INFO     | __main__:<module>:44 - 上海现在天气：多云，气温 15℃，体感较舒适(约 14℃)，湿度 63%，微风(西南风 2 米/秒)，能见度良好，大约 8 公里。建议穿