LangGraph 基础用法示例

from langgraph.graph import StateGraph。END
from typing_extensions import TypedDict
# . 定义状态：所有节点共享的数据
class MyState:
message: str
说到count，int
# . 定义节点函数
def say_hello:
return {"message": f"你好！你已经来了 {state} 次"}
def add_count:
return {"count": state + 1}
# . 建立图
graph = StateGraph
graph.add_node
graph.add_node
graph.set_entry_point # 入口节点
graph.add_edge # counter → greeter
graph.add_edge # greeter → 完成
app = graph.compile
result = app.invoke
print # {'message': '你好！你已经来了 1 次','count': 1}

This tiny example already shows three core concepts:

• Status : shared dict across nodes.
• b*Node*: ordinary Python function that receives & returns partial state.
• b*Edge*: declarative routing . \end{ul

  A conditional edge can decide next node based on state:

  def should_continue:
  if state>= 5:
  return "done"
  return "again"
  graph.add_conditional_edges(
  "greeter",should_continue,{
  "again": "counter",# loop back
  从"done"来看，END,},)
  
  Step 1️⃣ – 定义统一状态结构 
  from typing import Annotated,Sequence
  from typing_extensions import TypedDict
  from langchain_core.messages import BaseMessage
  import operator
  class AgentState:
  """All nodes share this structure"""
  messages: Annotated,operator.add] # 自动追加历史消息
  intents: list # Dispatcher 填写的意图序列
  说到mode，str # sequential | parallel
  current_step: int # 正在处理第几个意图
  
  The trick is using {operator.add}。which makes every node's returned “messages” automatically appended rar than overwritten – preserving conversation history without boilerplate.
  
  Step 2️⃣ – 实现 Dispatcher 🧭️️️︎︎︎︎︎︎︎︎︎︎︎‍♀️‍♂️‍♀️‍♂️‍♀️‌​‌​‌​‌​​‌‌​​​‌​‌‌‌‌​‌‌​​‏‏‏‏‏‏‎‎‎‎ ‎ ‎ ‎ ‎ ‎ ‎ ‎ ‎ ‏ ‏ ‏ ‏ ‏‏‏ ‏‎‎‭‭‮‫‫‪‬‬‬⁡⁢​​‌‌‍⬛⬜🟦🟧🟥🟩⬜⬛🔲🔳⛔✔❎⚠⚡❓☑️💡💬📚🗂️🌐📦🔧🤖💾🚀🧩🔍⚙️💬👨‍💻🏗️👷‍♂️🚦⏱️✅
  # 简化版 Dispatcher 示例
  DISPATCHER_PROMPT = """
  You are an intent classifier.
  至于Rules，- tools : requests that need external tools
  - writer : writing tasks
  - chat : casual conversation or knowledge Q&A
  If multiple steps are required,list m in order.
  Reply format: intent1,intent2|mode
  Examples:
  上海天气 -> tools|sequential
  查天气接下来写文案 -> tools,writer|sequential
  翻译hello 并查询北京天气 -> tools,writer|parallel
  """
  def dispatcher_node:
  last_user_msg = state.content # 假设最终一条是使用者输入
  response = invoke_llm,HumanMessage])
  intents_str,mode = response.split
  intents =
  return {
  "intents": intents,"mode": mode.strip,"current_step": 0,}
  
  This node produces two pieces of information used by every downstream node:
  
  
  A **ordered list of intents** – e.g. . This drives sequential/parallel executor.
  
  An **execution mode** – “sequential” means each intent waits for previous result;“parallel” runs m concurrently.
  \end ul>
  User Pain Point #6 – Intent 分类不准怎么办？*
  
  
  If classifier is uncertain,fallback to “chat” as a safe default and let a downstream LLM clarify with follow‑up questions.
  
  You can improve accuracy by adding few‑shot examples or by fine‑tuning a small model dedicated to intent detection.
  \end div>