LangGraph详解：专注推理流程，让AI Agent“会思考、能落地” - AI创想

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from dotenv import load_dotenv
import os
import json
# 1. 加载环境变量，配置大模型
load_dotenv()
llm = ChatOpenAI(
model="gpt-3.5-turbo",
temperature=0.1,
api_key=os.getenv("OPENAI_API_KEY") # 替换为你的API Key
)
# 2. 定义工具（天气查询工具，模拟真实API调用）
@tool
def get_weather(city: str, date: str) -> str:
"""
功能：查询指定城市指定日期的天气情况
参数：
city: 城市名称（如“北京”“上海”）
date: 日期（格式：YYYY-MM-DD，如“2025-03-15”）
返回：天气描述字符串
"""
# 模拟天气数据，实际开发可对接真实天气API
weather_data = {
"北京-2025-03-15": "晴，-5℃~5℃，微风，空气质量良",
"上海-2025-03-15": "阴转小雨，8℃~12℃，东风3级，空气质量优"
}
key = f"{city}-{date}"
return weather_data.get(key, f"未查询到{city}{date}的天气信息")
# 3. 定义状态（智能体的记忆，包含所有推理过程中的信息）
class AgentState:
input: str # 用户原始问题
thought: str # 思考过程
action: dict # 工具调用指令（JSON格式）
observation: str # 工具调用结果
done: bool = False # 是否完成任务
# 4. 定义节点（思考、行动、观察、回答、条件节点）
# 4.1 思考节点：分析当前状态，决定下一步行动
def thought_node(state: AgentState) -> AgentState:
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个天气查询智能体，基于当前状态，思考下一步该做什么：\n"
"1. 如果已经获取到天气信息（observation有内容），则设置done=True，准备生成回答；\n"
"2. 如果没有获取到天气信息，需要调用get_weather工具，明确传入city和date参数；\n"
"3. 思考过程要清晰，明确说明下一步行动。"),
("user", "用户问题：{input}\n当前观察结果：{observation}")
])
chain = prompt | llm
response = chain.invoke({"input": state.input, "observation": state.observation})
return AgentState(
input=state.input,
thought=response.content,
action=state.action,
observation=state.observation,
done=state.done
)
# 4.2 行动节点：生成工具调用指令并执行
def action_node(state: AgentState) -> AgentState:
prompt = ChatPromptTemplate.from_messages([
("system", "基于当前思考过程，生成get_weather工具的调用指令，格式为JSON：\n"
'{"name": "get_weather", "parameters": {"city": "城市名", "date": "日期"}}'),
("user", "思考过程：{thought}\n用户问题：{input}")
])
chain = prompt | llm
response = chain.invoke({"thought": state.thought, "input": state.input})
# 解析工具调用指令（实际开发可增加异常处理）
action = json.loads(response.content.strip())
return AgentState(
input=state.input,
thought=state.thought,
action=action,
observation=state.observation,
done=state.done
)
# 4.3 观察节点：处理工具调用结果，更新状态
def observation_node(state: AgentState) -> AgentState:
action = state.action
if action["name"] == "get_weather":
# 执行工具调用
observation = get_weather(** action["parameters"])
return AgentState(
input=state.input,
thought=state.thought,
action=state.action,
observation=observation,
done=state.done
)
else:
return AgentState(
input=state.input,
thought=state.thought,
action=state.action,
observation="工具调用失败，未找到对应工具",
done=state.done
)
# 4.4 回答节点：生成最终自然语言回答
def answer_node(state: AgentState) -> AgentState:
prompt = ChatPromptTemplate.from_messages([
("system", "基于用户问题和工具调用结果，生成自然、清晰的回答，不要添加额外信息。"),
("user", "用户问题：{input}\n工具调用结果：{observation}")
])
chain = prompt | llm
response = chain.invoke({"input": state.input, "observation": state.observation})
# 将回答存入thought字段（简化处理，实际可自定义字段）
return AgentState(
input=state.input,
thought=response.content,
action=state.action,
observation=state.observation,
done=True
)
# 4.5 条件节点：判断任务是否完成，决定跳转方向
def conditional_node(state: AgentState) -> str:
if state.done:
return "answer" # 任务完成，跳转至回答节点
else:
return "action" # 任务未完成，跳转至行动节点
# 5. 构建LangGraph（状态机）
graph = StateGraph(AgentState)
# 添加节点
graph.add_node("thought", thought_node) # 思考节点
graph.add_node("action", action_node) # 行动节点
graph.add_node("observation", observation_node)# 观察节点
graph.add_node("answer", answer_node) # 回答节点
# 添加边（流转规则）
graph.add_edge("thought", conditional_node) # 思考节点后进入条件判断
graph.add_edge("action", "observation") # 行动节点后进入观察节点
graph.add_edge("observation", "thought") # 观察节点后回到思考节点，形成循环
graph.add_edge("answer", END) # 回答节点后，流程终止
# 设置起始节点，编译图
graph.set_entry_point("thought")
app = graph.compile()
# 6. 运行智能体，测试效果
user_question = "查询上海2025年03月15日的天气"
initial_state = AgentState(
input=user_question,
thought="",
action={},
observation="",
done=False
)
# 执行推理流程，获取结果
result = app.invoke(initial_state)
# 输出结果
print("用户问题：", result.input)
print("思考过程：", result.thought)
print("工具调用指令：", result.action)
print("工具调用结果：", result.observation)
print("最终回答：", result.thought)

复制代码