AI大模型之LangGraph实现Agent

官方方向已经明确： 👉 未来 Agent = LangGraph，所以这里不再讲解Langchain实现Agent，直接开始LangGraph实现的LangGraph，后续会讲解LangGraph
参考文档：https://docs.langchain.org.cn/oss/python/langgraph/overview

一、什么Agent智能体？

Agent（智能体） 是一个通过动态协调 大语言模型（LLM） 和 工具（Tools） 来完成复杂任务的智能系统。它让LLM充当"决策大脑"，根据用户输入自主选择和执行工具（如搜索、计算、数据库查询等），最终生成精准的响应。

---

这些 AI Agent 可以应用于各种场景：
帮助企业解决实际问题，例如自动化运维、代码生成、智能客服等场景。 AI Agent 的核心是大语言模型（LLM）。因此，AI Agent 也常被称为 LLM 智能体。由于 LLM的知识受训练数据限制，推理能力也存在一定局限，AI Agent技术则通过工具调用来弥补这些不足，从而获取最新信息、优化工作流、拆解任务，最终实现更复杂的功能。

---

与传统 LLM 相比，AI Agent 多了几个关键能力：
（1）目标初始化与规划：在用户给定目标和工具的前提下，Agent
能够将任务拆解并制定执行计划。简单任务可以边执行边调整，复杂任务则会先做整体规划。

（2）借助工具进行推理：当遇到无法直接解决的问题时，Agent 会调用外部工具补充知识，例如联网搜索或调用 API。

（3）学习与反思：Agent 能够记住与用户的交互，并不断调整优化自己的行为。

二、AI智能体的核心能力

1）大模型(LLM)：作为大脑，提供推理、规划和知识理解能力。 比如：OpenaAI()、ChatOpenAI()
2）记忆(Memory)：具备短期记忆（上下文）和长期记忆（向量存储），支持快速知识检索。 比如：ConversationBufferMemory、ConversationSummaryMemory、ConversationBufferWindowMemory等
3）工具(Tools)：调用外部工具（如API、数据库）的执行单元比如：SearchTool、CalculatorTool
4）规划(Planning)：任务分解、反思与自省框架实现复杂任务处理
5）行动(Action)：实际执行决策的能力 比如：检索、推理、编程 6）协作：通过与其他智能体交互合作，完成更复杂的任务目标。

智能体核心能力，多个智能体还可以相互协作

三、代码示例

import os
import dotenv
import requests
from typing import Annotated, TypedDict

from langchain_openai import ChatOpenAI
from langchain_community.tools import GoogleSerperRun
from langchain_community.utilities import GoogleSerperAPIWrapper

from langchain_core.tools import BaseTool
from langchain_core.messages import (
    BaseMessage,
    HumanMessage,
)
from pydantic import BaseModel, Field

from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

# 加载环境(这里使用了谷歌和高德，所以配置文件配置了谷歌和高德的GAODE_API_KEY和SERPER_API_KEY)
dotenv.load_dotenv()

# =====================================================
# 1️⃣ 创建两个工具
# =====================================================

# ---------- 工具一、谷歌搜索 ----------
google_serper_tool = GoogleSerperRun(
    name="google_serper_tool",
    description="谷歌搜索工具",
    api_wrapper=GoogleSerperAPIWrapper(),
)


# ---------- 工具二、高德IP定位 ----------
class GaoDeIpLocationInput(BaseModel):
    ip: str = Field(description="ip地址")

def safe_str(value):
    if isinstance(value, list):
        return "".join(value)
    return str(value or "")

class GaoDeIpLocationTool(BaseTool):
    name: str = "ip_location_tool"
    description: str = "高德IP定位API"
    args_schema: type[BaseModel] = GaoDeIpLocationInput

    def _run(self, ip: str):
        url = (
            f"https://restapi.amap.com/v3/ip?"
            f"ip={ip}&output=json&key={os.getenv('GAODE_API_KEY')}"
        )

        res = requests.get(url).json()

        if res.get("status") != "1":
            return "定位失败"

        return safe_str(res.get("province")) + safe_str(res.get("city"))


ip_location_tool = GaoDeIpLocationTool()
# 工具箱
tools = [google_serper_tool, ip_location_tool]

# =====================================================
# 2️⃣ 创建大模型
# =====================================================

llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,
)
# 绑定工具
llm_with_tools = llm.bind_tools(tools)

# =====================================================
# 3️⃣ 定义状态
# =====================================================

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# =====================================================
# 4️⃣ 定义Agent 节点
# =====================================================

def agent_node(state: AgentState):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# =====================================================
# 5️⃣ Agent 节点引用工具
# =====================================================

tool_node = ToolNode(tools)

# =====================================================
# 6️⃣ 定义路由
# =====================================================

def should_continue(state: AgentState):

    last_message = state["messages"][-1]

    if getattr(last_message, "tool_calls", None):
        return "tools"

    return END

# =====================================================
# 7️⃣ 创建流程图
# =====================================================

workflow = StateGraph(AgentState)

workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)

workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
)

workflow.add_edge("tools", "agent")
# 编译
graph = workflow.compile()

# =====================================================
# 8️⃣ ⭐ 人机对话模式（核心）
# =====================================================

print("\n🤖 Agent 已启动，输入 exit 退出\n")

state = {"messages": []}

while True:

    user_input = input("👤 用户：")

    if user_input.lower() in ["exit", "quit"]:
        print("👋 对话结束")
        break

    # 添加用户消息
    state["messages"].append(
        HumanMessage(content=user_input)
    )

    # 调用 LangGraph
    result = graph.invoke(state)

    # 更新历史
    state["messages"] = result["messages"]

    # 输出AI回复
    ai_msg = result["messages"][-1]
    print(f"🤖 AI：{ai_msg.content}")

四、LangGraph调用本地工具方法案例

from langchain.tools import tool
from langchain.chat_models import init_chat_model
import dotenv

from langchain.messages import (
    SystemMessage,
    HumanMessage,
    ToolMessage,
    AnyMessage,
)

from typing_extensions import TypedDict, Annotated
import operator
from typing import Literal

from langgraph.graph import StateGraph, START, END

from IPython.display import Image, display


# =====================================================
# 1️⃣ 加载环境变量（API Key 等）
# =====================================================
dotenv.load_dotenv()


# =====================================================
# 2️⃣ 初始化大模型
# =====================================================
"""
init_chat_model 是 LangChain 1.x 推荐方式。不推荐使用ChatOpenAI，因为：①只能用OpenAI 生态②强绑定 OpenAI 参数③不可切换 provider
init_chat_model 作用：
✅ 自动识别模型厂商
✅ 自动创建对应 Provider
✅ 支持多模型切换
结论：以后都用init_chat_model 就对了
"""

llm = init_chat_model(
    model="gpt-3.5-turbo",
    temperature=0
)


# =====================================================
# 3️⃣ 定义 Tools（工具函数）
# =====================================================
"""
@tool 会把普通 Python 函数：

    函数 → AI 可调用工具

LLM 会：
1️⃣ 阅读 docstring
2️⃣ 自动决定是否调用
"""

@tool
def multiply(a: int, b: int) -> int:
    """计算 a * b"""
    return a * b


@tool
def add(a: int, b: int) -> int:
    """计算 a + b"""
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """计算 a / b"""
    return a / b


# 工具列表
tools = [add, multiply, divide]

# 方便根据名字查找工具
tools_by_name = {tool.name: tool for tool in tools}


# =====================================================
# 4️⃣ 让 LLM 拥有工具能力
# =====================================================
"""
bind_tools 本质：

LLM + Tool Schema → Tool Calling LLM

从此：
LLM 可以返回 tool_calls
"""

model_with_tools = llm.bind_tools(tools)


# =====================================================
# 5️⃣ 定义 Graph State（核心）
# =====================================================
"""
LangGraph核心：

Graph = State + Node + Edge

State = Agent 的长期记忆
"""

class MessagesState(TypedDict):
    # 对话消息历史（自动 append）
    messages: Annotated[list[AnyMessage], operator.add]

    # 统计 LLM 调用次数（观察循环次数）
    llm_calls: int


# =====================================================
# 6️⃣ LLM 节点（大脑）
# =====================================================
"""
职责：
让 LLM 决定：

✅ 是否调用工具
✅ 或直接回复用户
"""

def llm_call(state: MessagesState):
    response = model_with_tools.invoke([SystemMessage(content="你是一个负责数学计算的助手")]+ state["messages"])
    return {
        "messages": [response],      # 新AI消息加入State
        "llm_calls": state.get("llm_calls", 0) + 1
    }


# =====================================================
# 7️⃣ Tool Node（执行工具）
# =====================================================
"""
职责：

1️⃣ 读取 LLM 的 tool_calls
2️⃣ 执行真实 Python 函数
3️⃣ 返回 ToolMessage 给 LLM
"""
def tool_node(state: MessagesState):

    results = []

    # 读取AI刚刚产生的tool调用
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        # 执行工具
        observation = tool.invoke(tool_call["args"])
        # 返回给LLM的结果
        results.append(ToolMessage(content=str(observation),tool_call_id=tool_call["id"],))

    return {"messages": results}


# =====================================================
# 8️⃣ 决策节点（是否继续循环）
# =====================================================
"""
Agent核心循环：

LLM
 ↓
是否调用工具？
 ↓
Tool
 ↓
再问LLM
"""

def should_continue(state: MessagesState) -> Literal["tool_node", END]:

    last_message = state["messages"][-1]

    # 如果AI决定调用工具
    if last_message.tool_calls:
        return "tool_node"

    # 否则结束Agent
    return END


# =====================================================
# 9️⃣ 构建 LangGraph Agent
# =====================================================

agent_builder = StateGraph(MessagesState)

# 注册节点
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# START → LLM
agent_builder.add_edge(START, "llm_call")

# LLM → (条件判断)
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END],
)

# Tool执行完 → 再回LLM
agent_builder.add_edge("tool_node", "llm_call")


# 编译Graph（生成运行时）
agent = agent_builder.compile()


# =====================================================
# 🔟 可视化 Agent Workflow
# =====================================================
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))


# =====================================================
# 1️⃣1️⃣ 运行 Agent
# =====================================================

messages = [HumanMessage(content="计算:3除以4")]

result = agent.invoke({"messages": messages})

# 打印完整对话
for m in result["messages"]:
    m.pretty_print()