AI入门实战|Agent骨架与工具调用|Tools工具调用

from langchain.tools import tool
from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL

# 1. 自定义工具：四则运算计算器
@tool
def calculator(expression: str) -> str:
    """
    数学计算器，用于计算四则运算表达式
    参数 expression: 数学表达式，例如 1+2、100*20、(50-10)/2
    """
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算失败：{str(e)}"

def main():
    print("===== 自定义工具 - 手动调用计算器 =====")
    expr = input("请输入数学表达式：").strip()
    if not expr:
        print("未输入内容")
        return
    
    # 手动执行工具
    res = calculator.invoke(expr)
    print(res)

if __name__ == "__main__":
    main()
    input("\n按回车关闭窗口...")

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate
from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL

# 1. 自定义工具
@tool
def calculator(expression: str) -> str:
    """
    数学计算器，专门处理加减乘除四则运算
    入参：标准数学表达式，如 99*99、(10+20)*5
    """
    try:
        return f"计算结果 = {eval(expression)}"
    except Exception:
        return "表达式格式错误，无法计算"

# 2. 初始化 DeepSeek 模型
llm = ChatOpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
    model="deepseek-chat",
    temperature=0,
    tiktoken_model_name=None
)

# 3. 工具列表（可放多个工具）
tools = [calculator]

# 4. 构建提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是智能助手，可以使用工具完成数学计算，其他问题直接回答。"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 5. 创建 Agent + 执行器
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

def main():
    print("===== Agent 自动调用自定义工具 =====")
    print("支持数学计算 + 普通问答，输入 exit 退出\n")

    while True:
        user_input = input("你：").strip()
        if user_input.lower() == "exit":
            print("助手：再见！")
            break
        if not user_input:
            continue
        
        # 执行 Agent（自动判断是否调用工具）
        result = agent_executor.invoke({"input": user_input})
        print(f"助手：{result['output']}\n")

if __name__ == "__main__":
    main()
    input("\n按回车关闭窗口...")

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
from datetime import datetime
from dotenv import load_dotenv
import os

load_dotenv()

# 工具1：计算器
@tool
def calculator(expression: str) -> str:
    """四则运算计算器，支持括号运算"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误：{e}"

# 工具2：获取当前时间
@tool
def get_current_time() -> str:
    """获取当前系统北京时间，格式：年-月-日 时:分:秒"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 工具列表
tools = [calculator, get_current_time]

# 模型初始化
llm = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    model="deepseek-chat",
    temperature=0
)
llm_with_tools = llm.bind_tools(tools)

# 通用工具调用逻辑
def run_agent(text: str):
    messages = [HumanMessage(content=text)]
    while True:
        res = llm_with_tools.invoke(messages)
        if not res.tool_calls:
            return res.content
        for call in res.tool_calls:
            tool = next(t for t in tools if t.name == call["name"])
            res_tool = tool.invoke(call["args"])
            messages.append(res)
            messages.append(ToolMessage(content=res_tool, tool_call_id=call["id"]))

if __name__ == "__main__":
    print("===== 多个自定义工具 =====")
    print("输入 exit 退出\n")
    while True:
        q = input("你：")
        if q == "exit":
            break
        print("助手：", run_agent(q), "\n")

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, ToolMessage
# 稳定内置工具：文件读取、文件写入
from langchain_community.tools import ReadFileTool, WriteFileTool
from dotenv import load_dotenv
import os

load_dotenv()

# 初始化内置工具
read_tool = ReadFileTool()
write_tool = WriteFileTool()
tools = [read_tool, write_tool]

llm = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    model="deepseek-chat",
    temperature=0
)
llm_with_tools = llm.bind_tools(tools)

def run_agent(text: str):
    messages = [HumanMessage(content=text)]
    while True:
        res = llm_with_tools.invoke(messages)
        if not res.tool_calls:
            return res.content
        for call in res.tool_calls:
            tool = next(t for t in tools if t.name == call["name"])
            tool_result = tool.invoke(call["args"])
            messages.append(res)
            messages.append(ToolMessage(content=tool_result, tool_call_id=call["id"]))

if __name__ == "__main__":
    print("===== 内置工具 - 文件读写 =====")
    print("示例指令：读取 test.txt  /  新建 note.txt 写入 学习LangChain工具\n")
    print("输入 exit 退出\n")
    while True:
        q = input("你：")
        if q == "exit":
            break
        print("助手：", run_agent(q))

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_community.tools import ReadFileTool, WriteFileTool
from datetime import datetime
from dotenv import load_dotenv
import os

load_dotenv()

# ---------- 自定义工具 ----------
@tool
def calculator(expression: str) -> str:
    """四则运算计算器，处理数学计算"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算失败：{e}"

@tool
def get_current_time() -> str:
    """获取当前系统北京时间"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# ---------- 内置工具 ----------
read_file = ReadFileTool()
write_file = WriteFileTool()

# 合并所有工具
tools = [calculator, get_current_time, read_file, write_file]

llm = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    model="deepseek-chat",
    temperature=0
)
llm_with_tools = llm.bind_tools(tools)

def run_agent(text: str):
    messages = [HumanMessage(content=text)]
    while True:
        res = llm_with_tools.invoke(messages)
        if not res.tool_calls:
            return res.content
        for call in res.tool_calls:
            tool = next(t for t in tools if t.name == call["name"])
            tool_result = tool.invoke(call["args"])
            messages.append(res)
            messages.append(ToolMessage(content=tool_result, tool_call_id=call["id"]))

if __name__ == "__main__":
    print("===== 自定义工具 + 内置工具 混合实战 =====")
    print("输入 exit 退出\n")
    while True:
        user_input = input("你：")
        if user_input == "exit":
            break
        print("助手：", run_agent(user_input))