上下文

LangGraph 支持三种主要方式向语言模型提供“上下文”：

类型	描述	可变？	生命周期
运行时上下文	在运行开始时传入的静态数据	❌	每次运行
短期记忆（状态）	执行过程中可变化的中间状态	✅	每次运行/会话
长期记忆（存储）	可跨多次调用和会话共享的持久数据	✅	跨对话 / 全局

运行时上下文

注意点：需要将 langgraph 升级到0.6.0a2 的版本，langgraph.runtime 模块最新的版本中才加入的，之前的版本中没有

适用于：API 密钥、用户元信息等在运行过程中不会变化的数据。

使用 context 参数传入；
通过 get_runtime(ContextSchema) 获取；
推荐使用 context_schema 定义数据结构。

示例代码

在工作流中使用

# @Time：2025/9/30 15:01
# @Author：jinglv
from dataclasses import dataclass
from typing import TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime


# 定义状态
class State(TypedDict):
    """状态"""
    user_input: str
    test_cases: list


# 定义运行时的上下文参数
@dataclass
class RuntimeContext:
    """运行时上下文参数"""
    test_env: str  # 测试环境
    tester_name: str  # 测试人员名称


# ================定义运行节点函数=================

def generator_test_case(state: State):
    """生成测试用例"""
    print(f"用户输入的需求是:{state.get('user_input')},开始进行测试用例生成")
    # 这里核心的功能实现需要调用llm进行生成，暂时跳过
    print("测试用例已经生成", )
    return {"test_cases": ["测试用例1", "测试用例2"]}


def run_test_cases(state: State, runtime: Runtime[RuntimeContext]):
    """执行测试用例"""
    print("正在执行测试用例：", state['test_cases'])
    print("当前执行的测试环境", runtime.context.test_env)
    print("执行人", runtime.context.tester_name)
    # print("执行人", runtime.context.get('tester_name'))
    return {"report": "这个是一个测试报告"}


def generator_test_report(state: State, config: RunnableConfig):
    """生成测试报告"""
    print("执行generator_test_report节点")
    print("配置信息：", config)

    return {"report": "这个是一个测试报告"}


# =================工作流的创建个编排===================
graph = StateGraph(State, context_schema=RuntimeContext)
graph.add_node("生成测试用例", generator_test_case)
graph.add_node("执行测试用例", run_test_cases)
graph.add_node("生成测试报告", generator_test_report)

# 设置起点
# graph.set_entry_point("生成测试用例")
graph.add_edge(START, "生成测试用例")
graph.add_edge("生成测试用例", "执行测试用例")
graph.add_edge("执行测试用例", "生成测试报告")
# graph.set_finish_point("生成测试报告")
graph.add_edge("生成测试报告", END)

app = graph.compile()
# res = app.invoke({"user_input": "测试项目A"},
#                  config={"recursion_limit": 5},
#                  context={"test_env": "测试环境A", "tester_name": "张三"}
#                  )

res = app.invoke({"user_input": "测试项目A"},
                 config={"recursion_limit": 5},
                 context=RuntimeContext(test_env="测试环境A", tester_name="张三")
                 )

print(res)

在Agent中使用

from langgraph.runtime import get_runtime
from dataclasses import dataclass

@dataclass
class ContextSchema:
    user_name: str

# 用于构建 system prompt 的函数
def prompt(state):
    runtime = get_runtime(ContextSchema)
    user = runtime.context.user_name
    return [{"role": "system", "content": f"Hello {user}!"}, *state["messages"]]

# 创建 agent 时声明上下文 schema
agent = create_react_agent(
    model=llm,
    tools=[],
    prompt=prompt,
    context_schema=ContextSchema
)

# 传入上下文
agent.invoke(
    {"messages": [{"role": "user", "content": "Hi"}]},
    context={"user_name": "John Smith"}
)

流式输出

LangGraph对流式输出进行了改进，细化了输出模式（如Values、Updates、Debug、Messages），提升用户体验和调试效率，适用于生产环境AI应用。

LangGraph 支持从代理（Agent）或工作流（Graph）以流式方式输出执行过程与结果。

支持的流模式（stream_mode）

将以下一种或多种流模式作为列表传递给 stream（） 或 astream（） 方法：

模式	描述
`values`	每个图步骤后流式传输完整状态值。
`updates`	每个步骤后流式传输状态的增量更新。
`custom`	从图中节点流式传输自定义数据。
`messages`	从调用 LLM 的节点流式传输生成的 token 及元数据（元组）。
`debug`	尽可能多地流式传输调试信息。

支持将多种模式作为列表同时传递，如：

1	stream_mode=["updates", "messages", "custom"]

定义一个常见的图

from typing import TypedDict
from langgraph.graph import StateGraph, START


class State(TypedDict):
    topic: str
    joke: str


def refine_topic(state: State):
    return {"topic": state["topic"] + "和小狗"}


def generate_joke(state: State):
    return {"joke": f"这是一个关于{state['topic']}的笑话"}


graph = (
    StateGraph(State)
    .add_node(refine_topic)
    .add_node(generate_joke)
    .add_edge(START, "refine_topic")
    .add_edge("refine_topic", "generate_joke")
    .compile()
)

流式传输stream_mode=”values”

在图表的每个步骤之后传输状态的完整值

for chunk in graph.stream(
    {"topic": "冰激凌"},
    stream_mode="values",
):
    print(chunk)

执行结果

1
2
3

{'topic': '冰激凌'}
{'topic': '冰激凌和小狗'}
{'topic': '冰激凌和小狗', 'joke': '这是一个关于冰激凌和小狗的笑话'}

说明：values 则输出完整状态值。

流式传输stream_mode=”updates”

将图表每一步之后的更新流式传输到状态

for chunk in graph.stream(
    {"topic": "冰激凌"},
    stream_mode="updates",
):
    print(chunk)

执行结果

1 2	{'refine_topic': {'topic': '冰激凌和小狗'}} {'generate_joke': {'joke': '这是一个关于冰激凌和小狗的笑话'}}

说明：updates 模式输出节点的状态更新

流式传输stream_mode=”debug”

在图表的整个执行过程中传输尽可能多的信息

for chunk in graph.stream(
    {"topic": "冰激凌"},
    stream_mode="debug",
):
    print(chunk)

执行结果

{'step': 1, 'timestamp': '2025-09-18T08:51:31.478123+00:00', 'type': 'task', 'payload': {'id': '671c3949-69b8-8e90-b997-e76a1f711f5f', 'name': 'refine_topic', 'input': {'topic': '冰激凌'}, 'triggers': ('branch:to:refine_topic',)}}
{'step': 1, 'timestamp': '2025-09-18T08:51:31.478680+00:00', 'type': 'task_result', 'payload': {'id': '671c3949-69b8-8e90-b997-e76a1f711f5f', 'name': 'refine_topic', 'error': None, 'result': [('topic', '冰激凌和小狗')], 'interrupts': []}}
{'step': 2, 'timestamp': '2025-09-18T08:51:31.478828+00:00', 'type': 'task', 'payload': {'id': 'b7f1816a-7dab-4847-e8e6-78ae4d691a8e', 'name': 'generate_joke', 'input': {'topic': '冰激凌和小狗'}, 'triggers': ('branch:to:generate_joke',)}}
{'step': 2, 'timestamp': '2025-09-18T08:51:31.479006+00:00', 'type': 'task_result', 'payload': {'id': 'b7f1816a-7dab-4847-e8e6-78ae4d691a8e', 'name': 'generate_joke', 'error': None, 'result': [('joke', '这是一个关于冰激凌和小狗的笑话')], 'interrupts': []}}

从以上结果来看，可以看打执行的详细信息，每一个步骤执行

LLM token 流式传输stream_mode=”messages”

为调用LLM的图形节点传输LLM令牌和元数据

需要接入大模型

from langchain_deepseek import ChatDeepSeek
import os

llm = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0,
    api_key=os.environ.get("DEEPSEEK_API_KEY"),
    base_url=os.environ.get("DEEPSEEK_API_BASE"),
)


def generate_joke(state: State):
    llm_response = llm.invoke(
        [
            {"role": "user", "content": f"生成一个关于 {state['topic']}的笑话"}
        ]
    )
    return {"joke": llm_response.content}


graph = (
    StateGraph(State)
    .add_node(refine_topic)
    .add_node(generate_joke)
    .add_edge(START, "refine_topic")
    .add_edge("refine_topic", "generate_joke")
    .compile()
)

for message_chunk, metadata in graph.stream(
    {"topic": "冰激凌"},
    stream_mode="messages",
):
    if message_chunk.content:
        print(message_chunk.content, end="|", flush=True)

执行结果(流式输出的)

小狗|走进|一家|冰|激|凌|店|，|店员|问|：“|想要|什么|口|味的|？”
|小狗|说|：“|汪|草|味的|！”
|店员|愣|住|：“|抱歉|…|我们没有|汪|草|口味|。”
|小狗|叹气|：“|那|好吧|，|给我|一个|‘|爪子|’|饼干|筒|装|香|草|味|——|但|记住|，|这次|别|再把|我的|球|藏|进|冰|激|凌|里|了|，|上次|我|挖|了|半小时|！”|🍦|🐶|

|（|注|：|谐|音|梗|：|汪|草|=|香|草|，|爪子|=|甜|筒|品牌|“|可爱|多|”|的|经典|筒|身|设计|）|

说明：每生成一个 token 就会立即输出，并附带上下文信息。

工具中自定义数据流式输出stream_mode=”custom”

# @Time：2025/9/30 14:48
# @Author：jinglv
from dataclasses import dataclass
from typing import TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.config import get_stream_writer
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime

from app.agent.model.llms import qv_llm


# 定义状态
class State(TypedDict):
    """状态"""
    user_input: str
    test_cases: str
    result: str


# 定义运行时的上下文参数
@dataclass
class RuntimeContext:
    """运行时上下文参数"""
    test_env: str  # 测试环境
    tester_name: str  # 测试人员名称


def generator_test_case(state: State):
    """生成测试用例"""
    writer = get_stream_writer()
    writer(f"开始执行生成测试用例的节点")
    prompt = """
    请帮我生成用户5条登录的测试用例，登录账号密码的长度限制为8到16位     
    """
    cases = qv_llm.invoke(prompt)
    return {"test_cases": cases}


def run_test_cases(state: State, runtime: Runtime[RuntimeContext]):
    """执行测试用例"""
    writer = get_stream_writer()
    writer(f"开始执行【分析测试用例】的节点")
    cases = state['test_cases']
    prompt = f"""
    请分析当前的五条测试用例，是否有缺陷,
    用例数据如下：{cases}
    """
    result = qv_llm.invoke(prompt)
    return {"result": result}


def generator_test_report(state: State, config: RunnableConfig):
    """生成测试报告"""
    writer = get_stream_writer()
    writer(f"开始【生成测试报告】的节点运行")
    return {"report": "这个是一个测试报告"}


# =================工作流的创建个编排===================
graph = StateGraph(State, context_schema=RuntimeContext)
graph.add_node("生成测试用例", generator_test_case)
graph.add_node("执行测试用例", run_test_cases)
graph.add_node("生成测试报告", generator_test_report)

# 设置起点
# graph.set_entry_point("生成测试用例")
graph.add_edge(START, "生成测试用例")
graph.add_edge("生成测试用例", "执行测试用例")
graph.add_edge("执行测试用例", "生成测试报告")
# graph.set_finish_point("生成测试报告")
graph.add_edge("生成测试报告", END)

app = graph.compile()

response = app.stream({"user_input": "测试项目A"},
                      config={"recursion_limit": 5},
                      context=RuntimeContext(test_env="测试环境A", tester_name="张三"),
                      stream_mode=['messages', 'custom']
                      )

for input_type, chunk in response:
    if input_type == "messages":
        # ai的输出内容
        print(chunk[0].content, end="", flush=True)
    elif input_type == "custom":
        # 工具执行的输出内容
        print(chunk)

注意：必须在 LangGraph 执行上下文中调用 get_stream_writer()，否则无效。

Jean's Blog

LangGraph之上下文和流式输出

上下文

运行时上下文

示例代码

在工作流中使用

在Agent中使用

流式输出

支持的流模式（stream_mode）

流式传输stream_mode=”values”

流式传输stream_mode=”updates”

流式传输stream_mode=”debug”

LLM token 流式传输stream_mode=”messages”

工具中自定义数据流式输出stream_mode=”custom”