登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›[拆解LangChain执行引擎]一个实例理解LangChain的几种流 ...

[拆解LangChain执行引擎]一个实例理解LangChain的几种流模式

常士 昨天 08:55
程序园永久vip申请，500美金$，无限下载程序园所有程序/软件/数据/等
invoke/ainvoke方法看起来是采用简单的请求/回复消息交换模式，客户端需等待整个流程执行完毕后才能得到结果，其实方法背后还是会调用stream/astream方法以流的方式进行交互。如果我们直接调用调用这两个方法，并采用相应的流模式，我们就能有效解决客户端长时间无响应的问题，实时地得到对方的反馈。
  1. class Pregel(
  2.     PregelProtocol[StateT, ContextT, InputT, OutputT],
  3.     Generic[StateT, ContextT, InputT, OutputT]):
  4.    
  5.     def stream(
  6.         self,
  7.         input: InputT | Command | None,
  8.         config: RunnableConfig | None = None,
  9.         *,
  10.         context: ContextT | None = None,
  11.         stream_mode: StreamMode | Sequence[StreamMode] | None = None,
  12.         print_mode: StreamMode | Sequence[StreamMode] = (),
  13.         output_keys: str | Sequence[str] | None = None,
  14.         interrupt_before: All | Sequence[str] | None = None,
  15.         interrupt_after: All | Sequence[str] | None = None,
  16.         durability: Durability | None = None,
  17.         subgraphs: bool = False,
  18.         debug: bool | None = None,
  19.         **kwargs: Unpack[DeprecatedKwargs],
  20. ) -> Iterator[dict[str, Any] | Any]  
  21.    
  22.     async def astream(
  23.         self,
  24.         input: InputT | Command | None,
  25.         config: RunnableConfig | None = None,
  26.         *,
  27.         context: ContextT | None = None,
  28.         stream_mode: StreamMode | Sequence[StreamMode] | None = None,
  29.         print_mode: StreamMode | Sequence[StreamMode] = (),
  30.         output_keys: str | Sequence[str] | None = None,
  31.         interrupt_before: All | Sequence[str] | None = None,
  32.         interrupt_after: All | Sequence[str] | None = None,
  33.         durability: Durability | None = None,
  34.         subgraphs: bool = False,
  35.         debug: bool | None = None,
  36.         **kwargs: Unpack[DeprecatedKwargs],
  37.     ) -> AsyncIterator[dict[str, Any] | Any]
复制代码
在stream/astream方法的众多参数中，表示流模式的stream_mode参数最为重要，其对应类型StreamMode以字符串字面量的形式定义了七个选项。流是Pregel引擎向调用者提供数据的基本工作方法，它采用订阅发布的形式。Pregel对象发布的内容由对它订阅决定，因为发布客户端不敢兴趣的内容不但毫无意义，而且还会对影响造成极大的影响。
  1. StreamMode = Literal[
  2.     "values", "updates", "checkpoints", "tasks", "debug", "messages", "custom"
  3. ]
复制代码
在调用stream/astream方法时，我们可以根据需要指定一个或者多个流模式（以StreamMode序列的形式）。如果没有显式设置（None），Pregel对象自身的stream_mode字段会作为兜底，该字典的默认值为“values。如果当前Pregel对象以子图的形式被调用，会默认使用values模式，subgraphs参数用于控制是否希望得到子图的输出。下面列出了七种流模式对应的输出内容：

  • values：在每个Superstep结束后输出全部Channel的值；
  • updates：针对每个Node输出由它更新的Channel值；
  • checkpoints：在创建新的Checkpoint的时候，输出与get_state方法返回值具有相同结构的内容；
  • tasks：在Node任务开始和结束的时候输出任务ID、Node名称和其他相关信息；
  • debug：可以简单认为是tasks + checkpoints；
  • messages：输出语言模型产生的Token和相关元数据；
  • 开发者在Node处理函数中利用StreamWriter自行输出的内容；
如果混合使用多种流模式，stream/astream方法会返回一个字典，自带的Key表示当前输出采用的流模式。对于单一模式的调用，会直接返回输出的内容。如果采用custom模式，Node处理方法可以利用SteamWriter向客户端实时输出自定义的内容。StreamWriter和静态上下文一样，都属于当前Runtime的一部分，后者可以利用注入Node处理函数的RunnableConfig提取。下面演示程序会使用所有的流模式，我们在每个Node的处理函数中利用StreamWriter输出当前的Node名称。
  1. from langgraph.checkpoint.memory import InMemorySaver
  2. from langgraph.pregel import Pregel, NodeBuilder
  3. from langgraph.channels import LastValue, BinaryOperatorAggregate
  4. import operator
  5. from functools import partial
  6. from langchain_core.runnables import RunnableConfig
  7. from typing import Any,Sequence
  8. from langgraph.runtime import Runtime
  9. from langgraph.types import StreamWriter,StreamMode
  10. from collections import defaultdict
  12. def handle(node: str, inputs: dict[str, Any], config: RunnableConfig) -> list[str]:  
  13.     runtime:Runtime = config["configurable"].get("__pregel_runtime")
  14.     writer:StreamWriter = runtime.stream_writer
  15.     writer(f"node '{node}' is called.")
  16.     return [node]
  18. foo = (NodeBuilder()
  19.         .subscribe_to("foo",read = False)
  20.         .do(partial(handle, "foo"))
  21.         .write_to(bar="triggered by foo")
  22. )
  23. bar1 = (NodeBuilder()
  24.         .subscribe_to("bar",read = False)
  25.         .do(partial(handle, "bar1"))
  26.         .write_to("output")
  27. )
  28. bar2 = (NodeBuilder()
  29.         .subscribe_to("bar",read = False)
  30.         .do(partial(handle, "bar2"))
  31.         .write_to("output"))
  33. app = Pregel(
  34.     nodes={"foo": foo, "bar1": bar1, "bar2": bar2},
  35.     channels={
  36.         "foo": LastValue(str),
  37.         "bar": LastValue(str),
  38.         "output": BinaryOperatorAggregate(list, operator.add),
  39.     },  
  40.     input_channels=["foo"],
  41.     output_channels=["output"],
  42.     checkpointer=InMemorySaver(),
  43. )
  45. config={"configurable": {"thread_id": "123"}}
  46. stream_mode: Sequence[StreamMode] = ["values", "updates","checkpoints","tasks","debug","custom"]
  47. result: defaultdict[str, list[str]] = defaultdict(list)
  48. for (mode,chunk) in app.stream(input={"foo": None}, stream_mode= stream_mode, config=config):
  49.    result[mode].append(chunk)
  51. for mode,chunks in result.items():
  52.     index = 1
  53.     for chunk in chunks:
  54.         print(f"{index}.[{mode}] {chunk}")
  55.         index += 1
  56.     print()
复制代码
创建的Pregel由节点foo、bar1和bar2构成。节点foo率先执行，bar1和bar2随后并行执行。我们将调用stream方法收集到的内容根据流模式分组进行输出。我们来分析一下如下的输出结果：

  • 由于三个Node都会涉及到针对Channel的更新，所以会有三个updates模式的输出。
  • 整个流程涉及三个Supperstep，两个values模式的输出的全量的状态对应于后两个Superstep。
  • 三个Node对应三个任务，所以具有六个tasks模式的输出反映这三个任务的开始和结束。
  • 经历的三个Superstep对应三次Checkpoint的创建，所以我们能看到三个checkpoints模式的输出。
  • 六个tasks加三个checkpoints，所以有九个debug模式的输出。
  • 三个Node中针对StreamWriter的调用对应三个custom模式的输出。
  1. 1.[checkpoints] {'config': {'configurable': {'checkpoint_ns': '', 'thread_id': '123', 'checkpoint_id': '1f0fb144-d9ea-6dad-bfff-aaf158b4ced6'}}, 'parent_config': None, 'values': {'foo': None, 'output': []}, 'metadata': {'source': 'input', 'step': -1, 'parents': {}}, 'next': ['foo'], 'tasks': [{'id': 'd40d3fbc-0f70-33fe-1b31-51a3e12846fe', 'name': 'foo', 'interrupts': (), 'state': None}]}
  2. 2.[checkpoints] {'config': {'configurable': {'checkpoint_ns': '', 'thread_id': '123', 'checkpoint_id': '1f0fb144-d9ef-658f-8000-fe11b210fb05'}}, 'parent_config': {'configurable': {'checkpoint_ns': '', 'thread_id': '123', 'checkpoint_id': '1f0fb144-d9ea-6dad-bfff-aaf158b4ced6'}}, 'values': {'foo': None, 'bar': 'triggered by foo', 'output': []}, 'metadata': {'source': 'loop', 'step': 0, 'parents': {}}, 'next': ['bar1', 'bar2'], 'tasks': [{'id': 'ca9f30fb-ebea-87bf-074c-9a397616125a', 'name': 'bar1', 'interrupts': (), 'state': None}, {'id': 'c4540722-5a7a-4df3-14e1-abe0fddc9d73', 'name': 'bar2', 'interrupts': (), 'state': None}]}  
  3. 3.[checkpoints] {'config': {'configurable': {'checkpoint_ns': '', 'thread_id': '123', 'checkpoint_id': '1f0fb144-d9f3-6cbf-8001-e4219a2f4b58'}}, 'parent_config': {'configurable': {'checkpoint_ns': '', 'thread_id': '123', 'checkpoint_id': '1f0fb144-d9ef-658f-8000-fe11b210fb05'}}, 'values': {'foo': None, 'bar': 'triggered by foo', 'output': ['bar1', 'bar2']}, 'metadata': {'source': 'loop', 'step': 1, 'parents': {}}, 'next': [], 'tasks': []}
  5. 1.[debug] {'step': -1, 'timestamp': '2026-01-27T00:08:26.865918+00:00', 'type': 'checkpoint', 'payload': {'config': {'configurable': {'checkpoint_ns': '', 'thread_id': '123', 'checkpoint_id': '1f0fb144-d9ea-6dad-bfff-aaf158b4ced6'}}, 'parent_config': None, 'values': {'foo': None, 'output': []}, 'metadata': {'source': 'input', 'step': -1, 'parents': {}}, 'next': ['foo'], 'tasks': [{'id': 'd40d3fbc-0f70-33fe-1b31-51a3e12846fe', 'name': 'foo', 'interrupts': (), 'state': None}]}}
  6. 2.[debug] {'step': 0, 'timestamp': '2026-01-27T00:08:26.865933+00:00', 'type': 'task', 'payload': {'id': 'd40d3fbc-0f70-33fe-1b31-51a3e12846fe', 'name': 'foo', 'input': {}, 'triggers': ('foo',)}}
  7. 3.[debug] {'step': 0, 'timestamp': '2026-01-27T00:08:26.866516+00:00', 'type': 'task_result', 'payload': {'id': 'd40d3fbc-0f70-33fe-1b31-51a3e12846fe', 'name': 'foo', 'error': None, 'result': {'bar': 'triggered by foo'}, 'interrupts': []}}
  8. 4.[debug] {'step': 0, 'timestamp': '2026-01-27T00:08:26.867326+00:00', 'type': 'checkpoint', 'payload': {'config': {'configurable': {'checkpoint_ns': '', 'thread_id': '123', 'checkpoint_id': '1f0fb144-d9ef-658f-8000-fe11b210fb05'}}, 'parent_config': {'configurable': {'checkpoint_ns': '', 'thread_id': '123', 'checkpoint_id': '1f0fb144-d9ea-6dad-bfff-aaf158b4ced6'}}, 'values': {'foo': None, 'bar': 'triggered by foo', 'output': []}, 'metadata': {'source': 'loop', 'step': 0, 'parents': {}}, 'next': ['bar1', 'bar2'], 'tasks': [{'id': 'ca9f30fb-ebea-87bf-074c-9a397616125a', 'name': 'bar1', 'interrupts': (), 'state': None}, {'id': 'c4540722-5a7a-4df3-14e1-abe0fddc9d73', 'name': 'bar2', 'interrupts': (), 'state': None}]}}
  9. 5.[debug] {'step': 1, 'timestamp': '2026-01-27T00:08:26.867338+00:00', 'type': 'task', 'payload': {'id': 'ca9f30fb-ebea-87bf-074c-9a397616125a', 'name': 'bar1', 'input': {}, 'triggers': ('bar',)}}
  10. 6.[debug] {'step': 1, 'timestamp': '2026-01-27T00:08:26.867343+00:00', 'type': 'task', 'payload': {'id': 'c4540722-5a7a-4df3-14e1-abe0fddc9d73', 'name': 'bar2', 'input': {}, 'triggers': ('bar',)}}
  11. 7.[debug] {'step': 1, 'timestamp': '2026-01-27T00:08:26.868043+00:00', 'type': 'task_result', 'payload': {'id': 'c4540722-5a7a-4df3-14e1-abe0fddc9d73', 'name': 'bar2', 'error': None, 'result': {'output': ['bar2']}, 'interrupts': []}}
  12. 8.[debug] {'step': 1, 'timestamp': '2026-01-27T00:08:26.868117+00:00', 'type': 'task_result', 'payload': {'id': 'ca9f30fb-ebea-87bf-074c-9a397616125a', 'name': 'bar1', 'error': None, 'result': {'output': ['bar1']}, 'interrupts': []}}
  13. 9.[debug] {'step': 1, 'timestamp': '2026-01-27T00:08:26.868533+00:00', 'type': 'checkpoint', 'payload': {'config': {'configurable': {'checkpoint_ns': '', 'thread_id': '123', 'checkpoint_id': '1f0fb144-d9f3-6cbf-8001-e4219a2f4b58'}}, 'parent_config': {'configurable': {'checkpoint_ns': '', 'thread_id': '123', 'checkpoint_id': '1f0fb144-d9ef-658f-8000-fe11b210fb05'}}, 'values': {'foo': None, 'bar': 'triggered by foo', 'output': ['bar1', 'bar2']}, 'metadata': {'source': 'loop', 'step': 1, 'parents': {}}, 'next': [], 'tasks': []}}
  15. 1.[tasks] {'id': 'd40d3fbc-0f70-33fe-1b31-51a3e12846fe', 'name': 'foo', 'input': {}, 'triggers': ('foo',)}
  16. 2.[tasks] {'id': 'd40d3fbc-0f70-33fe-1b31-51a3e12846fe', 'name': 'foo', 'error': None, 'result': {'bar': 'triggered by foo'}, 'interrupts': []}
  17. 3.[tasks] {'id': 'ca9f30fb-ebea-87bf-074c-9a397616125a', 'name': 'bar1', 'input': {}, 'triggers': ('bar',)}
  18. 4.[tasks] {'id': 'c4540722-5a7a-4df3-14e1-abe0fddc9d73', 'name': 'bar2', 'input': {}, 'triggers': ('bar',)}
  19. 5.[tasks] {'id': 'c4540722-5a7a-4df3-14e1-abe0fddc9d73', 'name': 'bar2', 'error': None, 'result': {'output': ['bar2']}, 'interrupts': []}
  20. 6.[tasks] {'id': 'ca9f30fb-ebea-87bf-074c-9a397616125a', 'name': 'bar1', 'error': None, 'result': {'output': ['bar1']}, 'interrupts': []}
  22. 1.[custom] node 'foo' is called.
  23. 2.[custom] node 'bar2' is called.
  24. 3.[custom] node 'bar1' is called.
  26. 1.[updates] {'foo': {'bar': 'triggered by foo'}}
  27. 2.[updates] {'bar2': {'output': ['bar2']}}
  28. 3.[updates] {'bar1': {'output': ['bar1']}}
  30. 1.[values] {'foo': None, 'bar': 'triggered by foo', 'output': []}
  31. 2.[values] {'foo': None, 'bar': 'triggered by foo', 'output': ['bar1', 'bar2']}
复制代码
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
LangChain拆解执行引擎一个

使用道具 举报

相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
常士
昨天 08:55

0

粉丝关注

23

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫