调用图的两种方式
invoke
- 阻塞型端到端调用，输入初始状态、输出最终状态
# 代码仅为逻辑示意，具体视 LangGraph 最新版本而定user_input = "请查一下今天新加坡的天气"# invoke 直接返回图执行完毕后的最终状态 (通常是一个字典 dict)final_state = graph.invoke({"messages": [{"role": "user", "content": user_input}]})# 直接从最终状态中提取最后一条消息（通常是 Agent 的最终回答）print("最终回答:", final_state["messages"][-1].content)
stream
- 生成器模式，流式调用，获取每一个步骤的输出
def stream_graph_updates(user_input: str):# stream 返回的是一个迭代器。# 默认情况下，每次某个节点（Node）执行完毕，就会 yield 一个 event。for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):# event 通常是一个字典，键是刚刚执行完的 Node 的名字（如 'agent', 'tools'）# 值是该 Node 对总状态（State）做出的增量更新（Updates）# 例如: {'agent': {'messages': [AIMessage(content="我要调用天气工具")]}}for value in event.values():# 打印该节点输出的最后一条消息print("Assistant/Node Output:", value["messages"][-1].content)
对状态的处理
在langgraph中不同节点间流转的是状态，通常定义为AgentState，状态经过节点时会发生改变，langgraph底层实现了增量改变，节点输入当前状态、输出对状态的增量改变即可:
def _execute_sql_node(state: AgentState) -> dict:"""执行优化后的 SQL。"""sql = state.get("optimized_sql", state.get("sql", ""))result = execute_sql(sql) # 假设这是你的 DuckDB 执行逻辑# 【推荐写法】只返回你要更新/增加的字段！# LangGraph 底层会自动保留 state 里的 question, route 等其他字段return {"query_result": result}