AI Agent 从入门到落地：用 LangGraph 搭建能调用工具的任务型代理

数据狂人

664人浏览 · 2026-04-08 14:28:39

数据狂人 · 2026-04-08 14:28:39 发布

本文为原创技术分享，仅用于技术学习交流。

适用场景：想搭建能调用工具的 AI Agent，网上讲原理的多，真正给你完整可运行代码实例教程少，本文手把手带你做一个能查询天气、计算的工具调用 Agent。

一、痛点场景：Agent 开发，你是否也卡在这里？

大语言模型 Agent 现在很火，大家都想做一个能自己调用工具完成任务的智能代理。比如：

用户问"北京今天天气热吗"，Agent 知道调用天气 API 查询

用户问"399 的 85 折是多少钱"，Agent 知道调用计算器算出来

但是网上教程大多在讲 LangGraph 原理，什么状态机、循环、节点，看完原理还是不知道怎么写出一个能跑通的实例。

这篇文章就带你从 0 到 1，用 LangGraph 搭建一个支持多工具调用的任务型 Agent，代码完整可运行，看完你就能照着改出自己的工具调用 Agent。

二、最终效果展示

最终我们要实现这样的效果：

用户: 北京今天天气多少度？帮我算一下，如果开 26 度空调，一天耗电多少度？

Agent:

-> 调用 get_weather(city="北京") 获取天气

-> 得到结果: 北京今天 28 度

-> 调用 calculate(expression="(26 * 1.5 * 12)") 计算耗电

-> 得到结果: 468 度电

-> 生成最终回答: 北京今天 28°C，开 26 度空调预计耗电 468 度。

Agent 会自己判断需要调用哪些工具、按什么顺序调用，最后整合结果回答用户。本文实现的结构支持任意扩展工具，你只需要写工具函数，剩下的交给 LangGraph 调度。

三、实战代码：完整可运行

环境依赖安装

pip install langchain langgraph openai python-dotenv

完整代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
LangGraph 工具调用 Agent 完整示例
支持多工具顺序调用，自动判断需要调用什么工具
"""

import operator
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage, ToolMessage
from langchain_core.runnables import Runnable
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph
from langgraph.prebuilt import ToolNode
from openai import OpenAI
import json
import os

# ============ 配置 ============
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your-api-key")
OPENAI_API_BASE = os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1")
MODEL_NAME = "gpt-3.5-turbo"

# ============ 定义工具 ============
def get_weather(city: str) -> str:
    """查询城市今日天气
    Args:
        city: 城市名称
    """
    # 这里替换成你自己的天气 API 调用
    # 示例返回，实际项目对接真实接口
    fake_data = {
        "北京": "28°C，晴朗",
        "上海": "26°C，多云",
        "广州": "32°C，晴天",
        "深圳": "31°C，阴天",
    }
    return fake_data.get(city, f"未找到{city}天气数据")

def calculate(expression: str) -> str:
    """计算数学表达式结果
    Args:
        expression: 数学表达式，比如 "399 * 0.85"
    """
    try:
        result = eval(expression)
        return f"{result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

# 工具列表，在这里注册你所有的工具
tools = [get_weather, calculate]

# ============ 定义 Agent 状态 ============
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

# ============ 构建 Graph ============
def build_agent():
    # 初始化 LLM
    llm = ChatOpenAI(
        model=MODEL_NAME,
        openai_api_key=OPENAI_API_KEY,
        openai_api_base=OPENAI_API_BASE,
        temperature=0,
    ).bind_tools(tools)

    # 判断是否需要继续调用工具
    def should_continue(state: AgentState):
        messages = state["messages"]
        last_message = messages[-1]
        # 如果最后一条消息有工具调用，继续
        if getattr(last_message, "tool_calls", None):
            return "continue"
        # 否则结束
        return END

    # 调用 LLM 生成回复
    def call_llm(state: AgentState):
        messages = state["messages"]
        response = llm.invoke(messages)
        return {"messages": [response]}

    # 构建工作流
    workflow = StateGraph(AgentState)

    # 节点：LLM 推理
    workflow.add_node("agent", call_llm)
    # 节点：工具调用
    tool_node = ToolNode(tools)
    workflow.add_node("tools", tool_node)

    # 入口
    workflow.set_entry_point("agent")

    # 条件边：LLM 输出后，如果要调用工具，走工具节点，否则结束
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {
            "continue": "tools",
            END: END,
        },
    )

    # 工具调用完回到 LLM 整合结果
    workflow.add_edge("tools", "agent")

    # 编译得到可运行的 Agent
    app = workflow.compile()

    return app

# ============ 运行对话 ============
def run_agent(app, user_query: str):
    from langchain_core.messages import HumanMessage
    inputs = {"messages": [HumanMessage(content=user_query)]}
    output = []

    for chunk in app.stream(inputs, stream_mode="updates"):
        for node, update in chunk.items():
            output.append((node, update))
            print(f"\n--- 节点: {node} ---")
            if "messages" in update:
                last_msg = update["messages"][-1]
                if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
                    for tc in last_msg.tool_calls:
                        print(f"调用工具: {tc['name']} 参数: {tc['args']}")

    # 拿到最终回复
    final_state = chunk[-1][1]
    last_message = final_state["messages"][-1]
    return last_message.content

if __name__ == "__main__":
    app = build_agent()
    print("Agent 已启动，输入问题即可开始（输入 quit 退出）")
    while True:
        try:
            user_input = input("\n你: ")
            if user_input.lower() in ["quit", "exit", "q"]:
                break
            answer = run_agent(app, user_input)
            print(f"\nAgent: {answer}")
        except KeyboardInterrupt:
            break
    print("\n再见！")

四、原理解释：LangGraph Agent 工作流程

本文的 Agent 工作流程非常清晰，就三个节点：

用户提问 → [agent节点] → LLM 判断是否需要调用工具

↓

是 → [tools节点] → 执行工具调用 → 回到 [agent节点]

↓

否 → 结束 → 输出最终回答

节点说明

节点	作用
agent	调用大模型，根据当前对话生成下一个动作（回答或调用工具）
tools	执行工具调用，把工具结果放回对话
条件判断	决定继续调用工具还是结束