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()