安装依赖pip install -U langchain langchian-openaiapt-get install -y graphviz graphviz-devpip install pygraphvizLangGraph Hello World示例from typing import TypedDict from langgraph.graph import StateGraph, START,END # 1、定义 State class HelloState(TypedDict): name:str greeting:str # 2、定义节点 def greet(state:HelloState)-dict: name state[name] return {greeting:f你好,{name}!} def add_emoji(state: HelloState)-dict: greeting state[greeting] return {greeting:greeting } # 3、构件图 graph StateGraph(HelloState) graph.add_node(greet,greet) graph.add_node(add_emoji,add_emoji) graph.add_edge(START,greet) graph.add_edge(greet,add_emoji) graph_add.edge(add_emoji,END) # 4、编译 app graph.compile() # 5、运行 result app.invoke({name:张三}) print(result{greeting}) from IPython.display import Image,display # 使用 Graphviz 渲染(Colab 最稳定的方案) try: display(Image(app.get_graph(xrayTrue).draw_png())) except Exception as e: print(fGraphviz 渲染失败: {e}) print(\n使用 Mermaid 文本方式显示:) print(app.get_graph(xrayTrue).draw_mermaid())条件分支示例from typing import TypedDict,Literal from langgraph.graph import StateGraph,START,END class WeatherState(TypedDict): temperature:int recommendation:str def check_temperature(state:WeatherState)-dict: # 这里可以调用真实的天气API return {temperature:25} def route_by_temperature(state:WeatherState)-Literal[cold,warm,hot]: 根据温度路由 temp state[temperature] if temp 15: return cold elif temp 25: return warm else: return hot def recommend_cold(state:WeatherState)-dict: return {recommendation:穿厚外套!} def recommend_warm(state:WeatherState)-dict: return {recommendation:穿长袖就好.} def recommend_hot(state:WeatherState)-dict: return {recommendation:穿短袖记得防晒!} # 构建图 graph StateGraph(WeatherState) graph.add_node(check:check_temperature) graph.add_node(cold:recommend_cold) graph.add_node(warm:recommend_warm) graph.add_node(hot:recommend_hot) graph.add_edge(START,check) graph.add_conditional_edges( check, route_by_temperature, { cold:cold, warm:warm, hot:hot } ) for node in [cold,warm,hot]: graph.add_edge(node,END) app graph.compile() # 测试 result app.invoke({}) print(f温度:{result[temperature]}℃) print(f建议{result[recommendation]}) from IPython.display import Image.display # 渲染图 try: display(Image(app.get_graph(xrayTrue).draw_png())) except Exception as e: print(fGraphviz 渲染失败: {e}) print(\n使用 Mermaid 文本方式显示:) print(app.get_graph(xrayTrue).draw_mermaid())第一个Langraph Agent程序# Step 1: Define tools and model from langchain.tools import tool from langchain.chat_models import init_chat_model model init_chat_model( claude-sonnet-4-6, temperature0 ) # Define tools tool def multiply(a: int, b: int) - int: Multiply a and b. Args: a: First int b: Second int return a * b tool def add(a: int, b: int) - int: Adds a and b. Args: a: First int b: Second int return a b tool def divide(a: int, b: int) - float: Divide a and b. Args: a: First int b: Second int return a / b # Augment the LLM with tools tools [add, multiply, divide] tools_by_name {tool.name: tool for tool in tools} model_with_tools model.bind_tools(tools) # Step 2: Define state from langchain.messages import AnyMessage from typing_extensions import TypedDict, Annotated import operator class MessagesState(TypedDict): messages: Annotated[list[AnyMessage], operator.add] llm_calls: int # Step 3: Define model node from langchain.messages import SystemMessage def llm_call(state: MessagesState): LLM decides whether to call a tool or not return { messages: [ model_with_tools.invoke( [ SystemMessage( contentYou are a helpful assistant tasked with performing arithmetic on a set of inputs. ) ] state[messages] ) ], llm_calls: state.get(llm_calls, 0) 1 } # Step 4: Define tool node from langchain.messages import ToolMessage def tool_node(state: MessagesState): Performs the tool call result [] for tool_call in state[messages][-1].tool_calls: tool tools_by_name[tool_call[name]] observation tool.invoke(tool_call[args]) result.append(ToolMessage(contentobservation, tool_call_idtool_call[id])) return {messages: result} # Step 5: Define logic to determine whether to end from typing import Literal from langgraph.graph import StateGraph, START, END # Conditional edge function to route to the tool node or end based upon whether the LLM made a tool call def should_continue(state: MessagesState) - Literal[tool_node, END]: Decide if we should continue the loop or stop based upon whether the LLM made a tool call messages state[messages] last_message messages[-1] # If the LLM makes a tool call, then perform an action if last_message.tool_calls: return tool_node # Otherwise, we stop (reply to the user) return END # Step 6: Build agent # Build workflow agent_builder StateGraph(MessagesState) # Add nodes agent_builder.add_node(llm_call, llm_call) agent_builder.add_node(tool_node, tool_node) # Add edges to connect nodes agent_builder.add_edge(START, llm_call) agent_builder.add_conditional_edges( llm_call, should_continue, [tool_node, END] ) agent_builder.add_edge(tool_node, llm_call) # Compile the agent agent agent_builder.compile() from IPython.display import Image, display # Show the agent display(Image(agent.get_graph(xrayTrue).draw_mermaid_png())) # Invoke from langchain.messages import HumanMessage messages [HumanMessage(contentAdd 3 and 4.)] messages agent.invoke({messages: messages}) for m in messages[messages]: m.pretty_print()脏话判别是否需要人工介入from typing import TypedDict,Literal from langgraph.graph import StateGraph,START,END from langchain.chat_models import init_char_model from langchain.messages import SystemMessage,HumanMessage # State定义 class AgentModerationState(TypedDict): content:str analysis:str decision: str #approved | rejected | needs_review reason:str confidence:float from goole.colab import userdata API_KEY uerdata.get(XXX_API_KEY) # LLM初始化 model init_chat_model( Qwen/Qwen3-8B, model_provider openai, base_urlhttps://api.siliconflow.cn/v1, api_key API_KEY, termperature 0.0 ) # 节点实现 def analyze_content(state:AgentModerationState)-dict: 使用LLM分析内容 content state[content] system_prompt 你是一个内容审核助手。分析给定的内容判断是否包含 1、不当语言脏话、侮辱 2、垃圾信息广告、刷屏 3、敏感话题政治、暴力、色情 请以 JSON 格式返回 { has_issues: true/false, issues: [issue1,issue2], severity: low/medium/high, confidence: 0.0-1.0 } messages [ SystemMessage(content system_prompt), HumanMessage(content f内容:{content}) ] response model.invoke(messages) return {analysis:response.content} def make_agent_decision(state:AgentModerationState)-dict: 基于 LLM 分析做决策 analysis state[analysis] content state[content] system_prompt 基于之前的分析结果做出审核决策 - approved:内容正常通过 - rejected:明显违规直接拒绝 - need_review:需要人工审核 返回 JSON 格式 { decision:approved/rejected/need_review, reason:简短说明, confidence:0.0-1.0 } messages [ SystemMessage(content system_prompt), HumanMessage(content f原内容:{content}\n\n分析结果{analysis}) ] response model.invoke(messages) # 简化处理直接解析响应实际应用中应该用JSON mode) import json try: result json.load(response.content) return { decision:result[decision], reason:result[reason], confidence:result[confidence] } expect: #解析失败 return { decision : needs_review, reason : 分析结果解析失败, confidence:0.0 } # 路由函数 def should_auto_decide(state:AgentModerationState)-Literal[decide,review]: 根据分析结果决定是否需要人工 # 这里可以添加复杂逻辑 # 例如如果分析中提到搞风险直接转人工 if high in state.get(analysis,).lower(): return review return decide def human_review_placeholder(state:AgentModerationState)-dict: 人工审核占位符实际应用中会中断等待人工 return { decision:need_review, reason:已转人工审核, confidence:1.0 } # 构建Agent Graph agent_graph StateGraph(AgentModerationState) agent_graph.add_node(analyze,analyze_content) agent_graph.add_node(decide,make_agent_decision) agent_graph.add_node(human_review,human_review_placeholder) agent_graph.add_edge(START,analyze) agent_graph.add_conditional_edges( analyze, should_auto_decide, { decide:decide, review:human_review } ) agent_graph.add_edge(decide,END) agent_graph.add_edge(review,END) agent_app agent_graph.compile() from IPython.display import Image,display try: display(Image(agent_app.get_graph(xrayTrue).draw_png())) except Exception as e: print(fGraphviz 渲染失败 {e}) print(\n使用 Mermaid 文本方式显示:) print(agent_app.get_graph(xrayTrue).draw_mermaid())
LangGraph【入门】
发布时间：2026/7/30 19:48:03
安装依赖pip install -U langchain langchian-openaiapt-get install -y graphviz graphviz-devpip install pygraphvizLangGraph Hello World示例from typing import TypedDict from langgraph.graph import StateGraph, START,END # 1、定义 State class HelloState(TypedDict): name:str greeting:str # 2、定义节点 def greet(state:HelloState)-dict: name state[name] return {greeting:f你好,{name}!} def add_emoji(state: HelloState)-dict: greeting state[greeting] return {greeting:greeting } # 3、构件图 graph StateGraph(HelloState) graph.add_node(greet,greet) graph.add_node(add_emoji,add_emoji) graph.add_edge(START,greet) graph.add_edge(greet,add_emoji) graph_add.edge(add_emoji,END) # 4、编译 app graph.compile() # 5、运行 result app.invoke({name:张三}) print(result{greeting}) from IPython.display import Image,display # 使用 Graphviz 渲染(Colab 最稳定的方案) try: display(Image(app.get_graph(xrayTrue).draw_png())) except Exception as e: print(fGraphviz 渲染失败: {e}) print(\n使用 Mermaid 文本方式显示:) print(app.get_graph(xrayTrue).draw_mermaid())条件分支示例from typing import TypedDict,Literal from langgraph.graph import StateGraph,START,END class WeatherState(TypedDict): temperature:int recommendation:str def check_temperature(state:WeatherState)-dict: # 这里可以调用真实的天气API return {temperature:25} def route_by_temperature(state:WeatherState)-Literal[cold,warm,hot]: 根据温度路由 temp state[temperature] if temp 15: return cold elif temp 25: return warm else: return hot def recommend_cold(state:WeatherState)-dict: return {recommendation:穿厚外套!} def recommend_warm(state:WeatherState)-dict: return {recommendation:穿长袖就好.} def recommend_hot(state:WeatherState)-dict: return {recommendation:穿短袖记得防晒!} # 构建图 graph StateGraph(WeatherState) graph.add_node(check:check_temperature) graph.add_node(cold:recommend_cold) graph.add_node(warm:recommend_warm) graph.add_node(hot:recommend_hot) graph.add_edge(START,check) graph.add_conditional_edges( check, route_by_temperature, { cold:cold, warm:warm, hot:hot } ) for node in [cold,warm,hot]: graph.add_edge(node,END) app graph.compile() # 测试 result app.invoke({}) print(f温度:{result[temperature]}℃) print(f建议{result[recommendation]}) from IPython.display import Image.display # 渲染图 try: display(Image(app.get_graph(xrayTrue).draw_png())) except Exception as e: print(fGraphviz 渲染失败: {e}) print(\n使用 Mermaid 文本方式显示:) print(app.get_graph(xrayTrue).draw_mermaid())第一个Langraph Agent程序# Step 1: Define tools and model from langchain.tools import tool from langchain.chat_models import init_chat_model model init_chat_model( claude-sonnet-4-6, temperature0 ) # Define tools tool def multiply(a: int, b: int) - int: Multiply a and b. Args: a: First int b: Second int return a * b tool def add(a: int, b: int) - int: Adds a and b. Args: a: First int b: Second int return a b tool def divide(a: int, b: int) - float: Divide a and b. Args: a: First int b: Second int return a / b # Augment the LLM with tools tools [add, multiply, divide] tools_by_name {tool.name: tool for tool in tools} model_with_tools model.bind_tools(tools) # Step 2: Define state from langchain.messages import AnyMessage from typing_extensions import TypedDict, Annotated import operator class MessagesState(TypedDict): messages: Annotated[list[AnyMessage], operator.add] llm_calls: int # Step 3: Define model node from langchain.messages import SystemMessage def llm_call(state: MessagesState): LLM decides whether to call a tool or not return { messages: [ model_with_tools.invoke( [ SystemMessage( contentYou are a helpful assistant tasked with performing arithmetic on a set of inputs. ) ] state[messages] ) ], llm_calls: state.get(llm_calls, 0) 1 } # Step 4: Define tool node from langchain.messages import ToolMessage def tool_node(state: MessagesState): Performs the tool call result [] for tool_call in state[messages][-1].tool_calls: tool tools_by_name[tool_call[name]] observation tool.invoke(tool_call[args]) result.append(ToolMessage(contentobservation, tool_call_idtool_call[id])) return {messages: result} # Step 5: Define logic to determine whether to end from typing import Literal from langgraph.graph import StateGraph, START, END # Conditional edge function to route to the tool node or end based upon whether the LLM made a tool call def should_continue(state: MessagesState) - Literal[tool_node, END]: Decide if we should continue the loop or stop based upon whether the LLM made a tool call messages state[messages] last_message messages[-1] # If the LLM makes a tool call, then perform an action if last_message.tool_calls: return tool_node # Otherwise, we stop (reply to the user) return END # Step 6: Build agent # Build workflow agent_builder StateGraph(MessagesState) # Add nodes agent_builder.add_node(llm_call, llm_call) agent_builder.add_node(tool_node, tool_node) # Add edges to connect nodes agent_builder.add_edge(START, llm_call) agent_builder.add_conditional_edges( llm_call, should_continue, [tool_node, END] ) agent_builder.add_edge(tool_node, llm_call) # Compile the agent agent agent_builder.compile() from IPython.display import Image, display # Show the agent display(Image(agent.get_graph(xrayTrue).draw_mermaid_png())) # Invoke from langchain.messages import HumanMessage messages [HumanMessage(contentAdd 3 and 4.)] messages agent.invoke({messages: messages}) for m in messages[messages]: m.pretty_print()脏话判别是否需要人工介入from typing import TypedDict,Literal from langgraph.graph import StateGraph,START,END from langchain.chat_models import init_char_model from langchain.messages import SystemMessage,HumanMessage # State定义 class AgentModerationState(TypedDict): content:str analysis:str decision: str #approved | rejected | needs_review reason:str confidence:float from goole.colab import userdata API_KEY uerdata.get(XXX_API_KEY) # LLM初始化 model init_chat_model( Qwen/Qwen3-8B, model_provider openai, base_urlhttps://api.siliconflow.cn/v1, api_key API_KEY, termperature 0.0 ) # 节点实现 def analyze_content(state:AgentModerationState)-dict: 使用LLM分析内容 content state[content] system_prompt 你是一个内容审核助手。分析给定的内容判断是否包含 1、不当语言脏话、侮辱 2、垃圾信息广告、刷屏 3、敏感话题政治、暴力、色情 请以 JSON 格式返回 { has_issues: true/false, issues: [issue1,issue2], severity: low/medium/high, confidence: 0.0-1.0 } messages [ SystemMessage(content system_prompt), HumanMessage(content f内容:{content}) ] response model.invoke(messages) return {analysis:response.content} def make_agent_decision(state:AgentModerationState)-dict: 基于 LLM 分析做决策 analysis state[analysis] content state[content] system_prompt 基于之前的分析结果做出审核决策 - approved:内容正常通过 - rejected:明显违规直接拒绝 - need_review:需要人工审核 返回 JSON 格式 { decision:approved/rejected/need_review, reason:简短说明, confidence:0.0-1.0 } messages [ SystemMessage(content system_prompt), HumanMessage(content f原内容:{content}\n\n分析结果{analysis}) ] response model.invoke(messages) # 简化处理直接解析响应实际应用中应该用JSON mode) import json try: result json.load(response.content) return { decision:result[decision], reason:result[reason], confidence:result[confidence] } expect: #解析失败 return { decision : needs_review, reason : 分析结果解析失败, confidence:0.0 } # 路由函数 def should_auto_decide(state:AgentModerationState)-Literal[decide,review]: 根据分析结果决定是否需要人工 # 这里可以添加复杂逻辑 # 例如如果分析中提到搞风险直接转人工 if high in state.get(analysis,).lower(): return review return decide def human_review_placeholder(state:AgentModerationState)-dict: 人工审核占位符实际应用中会中断等待人工 return { decision:need_review, reason:已转人工审核, confidence:1.0 } # 构建Agent Graph agent_graph StateGraph(AgentModerationState) agent_graph.add_node(analyze,analyze_content) agent_graph.add_node(decide,make_agent_decision) agent_graph.add_node(human_review,human_review_placeholder) agent_graph.add_edge(START,analyze) agent_graph.add_conditional_edges( analyze, should_auto_decide, { decide:decide, review:human_review } ) agent_graph.add_edge(decide,END) agent_graph.add_edge(review,END) agent_app agent_graph.compile() from IPython.display import Image,display try: display(Image(agent_app.get_graph(xrayTrue).draw_png())) except Exception as e: print(fGraphviz 渲染失败 {e}) print(\n使用 Mermaid 文本方式显示:) print(agent_app.get_graph(xrayTrue).draw_mermaid())
