1. LangGraph 进阶概览

LangGraph 的核心优势在于构建具备“循环”能力的 Agent。相比基础的 StateGraph,进阶模式通常结合 ToolNodeMessagesState 来实现自动化工具调用。

2. Agent 代码示例

以下示例演示了如何创建一个可以查询实时天气并进行智能对话的 Agent。
from typing import Literal
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph, START, MessagesState
from langgraph.prebuilt import ToolNode

# 1. 定义工具函数(模拟外部查询)
@tool
def search_weather(query: str):
    """查询指定城市的天气信息"""
    if "北京" in query.lower():
        return "阴天有雾,气温 25 度"
    return "天气晴朗,气温 35 度以上,注意防暑"

tools = [search_weather]
tool_node = ToolNode(tools)

# 2. 初始化适配 NoneLinear 的模型并绑定工具
llm = ChatOpenAI(
    model="DeepSeek-V3.2",
    api_key="YOUR_NONELINEAR_API_KEY",
    base_url="https://api.nonelinear.com/v1"
).bind_tools(tools)

# 3. 定义逻辑节点
def call_model(state: MessagesState):
    """节点:调用 LLM 获取响应"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: MessagesState) -> Literal["tools", END]:
    """路由函数:判断是否需要继续调用工具"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 4. 构建图结构
workflow = StateGraph(MessagesState)

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

workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent") # 工具执行完回到 Agent

# 5. 编译运行
app = workflow.compile()
inputs = {"messages": [("user", "帮我查一下北京的天气,然后给出穿衣建议。")]}

for chunk in app.stream(inputs, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

3. 开发规范与模式

  • 消息流管理:使用 MessagesState 可以自动处理消息历史的追加。
  • 条件边 (Conditional Edges):通过 should_continue 逻辑,Agent 可以根据模型返回的 tool_calls 自主决定是继续执行工具还是结束任务。
  • NoneLinear 兼容性:由于 NoneLinear 完美兼容 OpenAI 工具调用接口,您可以无缝使用 bind_tools

[!TIP] 进阶开发者可以结合 MemorySaver 实现具备持久化记忆的 Agent 对话系统。