1.请求数据

        处理对话入口为：

        http://{ip:port}/threads/{thread_id/runs/stream

        其中的thread_id是在创建线程中返回给前端的线程id

        请求报文如下：

{
  "input": {
    "messages": [
      {
        "id": "8c0a8d55-d590-43f9-8c16-de782fa5ccbf",#一次请求唯一标识
        "type": "human",
        "content": [
          {
            "type": "text",
            "text": "请客观评价一下张邦昌"
          }
        ]
      }
    ]
  },
  "stream_mode": [ #参见流模式
    "values",
    "messages-tuple",
    "custom"
  ],
  "stream_subgraphs": true,
  "stream_resumable": true,
  "assistant_id": "agent", #使用的agent唯一标识
  "on_disconnect": "continue"
}
 

       2.数据模型

        涉及数据库表包括assistant，run和thread表，thread，分别存放助手数据、run数据和线程数据，其中thread和run表是1:n的关系。

        assistant定义如下：

        run定义如下，其中kwargs存放请求数据内容：

        3.源码分析

        对应入口函数为langgraph-api/api/runs.py文件中的stream_run方法，具体源码如下：

"""

分析。langgraph-api中使用了redis的PUB/SUB机制，完成请求方和处理方解耦，同时保证了处理的实时性。

该方法本身很简洁，调用Runs.Stream.subscribe方法提交请求，订阅

thread:{thread_id}:runn:{run_id}:stream频道，并通过在redis的

LIST_RUN_QUEUE队列中写入数据来唤醒后台任务。

然后调用Runs.Stream.join从订阅频道读取后台任务发布的数据，并以流式方式应答。

所以重点分析这两个方法。

"""

async def stream_run(
    request: ApiRequest,
):
    ""创建一个run，对应一次请求"""
    thread_id = request.path_params["thread_id"]
    payload = await request.json(RunCreateStateful)
    on_disconnect = payload.get("on_disconnect", "continue")
    run_id = uuid7() #后端生成run_id
    async with await Runs.Stream.subscribe(run_id, thread_id) as sub:#订阅处理器
        async with connect() as conn: #conn是到postgreas的数据库连接
            run = await create_valid_run(#创建一个run并保存在数据库中
                conn,
                thread_id,
                payload,
                request.headers,
                run_id=run_id,
                request_start_time=request.scope.get("request_start_time_ms"),
            )

        return EventSourceResponse(
            Runs.Stream.join(
                run["run_id"],
                thread_id=thread_id,
                cancel_on_disconnect=on_disconnect == "cancel",
                stream_channel=sub,
                last_event_id=None,
            ),
            headers={
                "Location": f"/threads/{thread_id}/runs/{run['run_id']}/stream",
                "Content-Location": f"/threads/{thread_id}/runs/{run['run_id']}",
            },
        )

        Runs.Stream在langgraph_runtime_postgres目录ops.py中定义，其subscribe方法源码如下：

        @staticmethod
        async def subscribe(
            run_id: UUID,
            thread_id: UUID | None = None,
        ) -> StreamHandler:
            """Subscribe to the run stream, returning a stream handler.
            The stream handler must be passed to `join` to receive messages."""

            """

                    channel格式如下：

                    thread:2f0a1fd1-5281-4769-b7f0-c0fcb7382622:

                    run:01997549-519e-721f-8642-e83573178d63:stream

                    channel_old格式如下：

                    run:01997549-519e-721f-8642-e83573178d63:stream

                    pattern格式如下：

                    channel:*

                    pattern_old格式如下：

                    channel_old:*

            """
            channel = CHANNEL_RUN_STREAM.format(thread_id, run_id)
            channel_old = CHANNEL_RUN_STREAM_OLD.format(run_id)
            # Keeping the pattern for rollout so that existing other streams still get picked up
            pattern = CHANNEL_RUN_STREAM.format(thread_id, run_id) + ":*"
            pattern_old = CHANNEL_RUN_STREAM_OLD.format(run_id) + ":*"
            pubsub = await get_pubsub(#订阅channel,channel_old和通配模式频道
                channels=(
                    channel,
                    channel_old,
                ),
                patterns=(
                    pattern,
                    pattern_old,
                ),
            )
            return pubsub#返回流处理器句柄

       在langgraph-api/models/run.py文件中的create_valid_run方法做一下分析，代码如下：

async def create_valid_run(
    conn: AsyncConnectionProto,
    thread_id: str | None,
    payload: RunCreateDict,
    headers: Mapping[str, str],
    barrier: asyncio.Barrier | None = None,
    run_id: UUID | None = None,
    request_start_time: float | None = None,
    temporary: bool = False,#未传递该参数，所以为False
) -> Run:
    request_id = headers.get("x-request-id")  # 当前为空
    
    (
        assistant_id, #agent唯一标识
        thread_id_,  #前端带入
        checkpoint_id, #从payload中提取
        run_id,   #前面生成的
    ) = _get_ids(
        thread_id,
        payload,
        run_id=run_id,
    )
    if (#暂不考虑分支
        (thread_id_ is None or temporary)
        and (command := payload.get("command"))
        and command.get("resume")
    ):
        raise HTTPException(
            status_code=400,
            detail="You must provide a thread_id when resuming.",
        )
    temporary = (temporary or thread_id_ is None) and payload.get(
        "on_completion", "delete"
    ) == "delete" #暂不考虑
    stream_resumable = payload.get("stream_resumable", False)#暂不考虑真假

    #stream_mode=["values","messages-tuple","custom"]， multitask_strategy=‘queue’

    #prevent_insert_if_inflight=False
    stream_mode, multitask_strategy, prevent_insert_if_inflight = assign_defaults(
        payload
    )
    # 配置和上下文，均不需要考虑
    config = payload.get("config") or {}
    context = payload.get("context") or {}

    """

        configurable初始化为{"configurable":{}}，

        然后增加chepoint_id后为{"configurable":{}, "checkpoint_id":id}，

        然后增加"checkpoint_ns":""

    """
    configurable = config.setdefault("configurable", {})

    if configurable and context:
        raise HTTPException(
            status_code=400,
            detail="Cannot specify both configurable and context. Prefer setting context alone. Context was introduced in LangGraph 0.6.0 and is the long term planned replacement for configurable.",
        )

    # 如果payload中有context，则拷贝给configurable
    if context:
        configurable = context.copy()
        config["configurable"] = configurable
    else: #context为{‘configurable':{}}
        context = configurable.copy()

    if checkpoint_id:#如果payload中有checkpoint_id，则写入configurable
        configurable["checkpoint_id"] = str(checkpoint_id)

    #如果payload中有checkpoint ，则写入configurable
    if checkpoint := payload.get("checkpoint"):
        configurable.update(checkpoint)

    #把http header中的信息增加到configurable中
    configurable.update(get_configurable_headers(headers))

    ctx = get_auth_ctx()#暂不考虑身份认证，此时user_id为None
    if ctx:
        user = cast(BaseUser | None, ctx.user)
        user_id = get_user_id(user)
        configurable["langgraph_auth_user"] = user
        configurable["langgraph_auth_user_id"] = user_id
        configurable["langgraph_auth_permissions"] = ctx.permissions
    else:
        user_id = None
    if not configurable.get("langgraph_request_id"):
        configurable["langgraph_request_id"] = request_id
    if ls_tracing := payload.get("langsmith_tracer"):
        configurable["__langsmith_project__"] = ls_tracing.get("project_name")
        configurable["__langsmith_example_id__"] = ls_tracing.get("example_id")
    if request_start_time:
        configurable["__request_start_time_ms__"] = request_start_time
    after_seconds = cast(int, payload.get("after_seconds", 0))
    configurable["__after_seconds__"] = after_seconds
    put_time_start = time.time()
    if_not_exists = payload.get("if_not_exists", "reject")#当前仅考虑为reject

    durability = payload.get("durability")
    if durability is None:
        checkpoint_during = payload.get("checkpoint_during")
        durability = "async" if checkpoint_during in (None, True) else "exit"

    #直接看这里。Runs.put插入数据到run表，并完成thread表的更新

    run_coro = Runs.put(
        conn,
        assistant_id,
        {#这里的数据会保存到run表的kwargs字段中
            "input": payload.get("input"),
            "command": payload.get("command"),
            "config": config,
            "context": context,
            "stream_mode": stream_mode,
            "interrupt_before": payload.get("interrupt_before"),
            "interrupt_after": payload.get("interrupt_after"),
            "webhook": payload.get("webhook"),
            "feedback_keys": payload.get("feedback_keys"),
            "temporary": temporary,
            "subgraphs": payload.get("stream_subgraphs", False),
            "resumable": stream_resumable,
            "checkpoint_during": payload.get("checkpoint_during", True),
            "durability": durability,
        },
        metadata=payload.get("metadata"),
        status="pending",
        user_id=user_id,
        thread_id=thread_id_,
        run_id=run_id,
        multitask_strategy=multitask_strategy,
        prevent_insert_if_inflight=prevent_insert_if_inflight,
        after_seconds=after_seconds,
        if_not_exists=if_not_exists,
    )
    run_ = await run_coro #在Runs.put中插入的run信息迭代器

    if barrier:#暂不考虑
        await barrier.wait()

    # abort if thread, assistant, etc not found
    try:
        first = await anext(run_)
    except StopAsyncIteration:
        raise HTTPException(
            status_code=404, detail="Thread or assistant not found."
        ) from None

    # handle multitask strategy
    inflight_runs = [run async for run in run_]
    if first["run_id"] == run_id:
        logger.info(
            "Created run",
            run_id=str(run_id),
            thread_id=str(thread_id_),
            assistant_id=str(assistant_id),
            multitask_strategy=multitask_strategy,
            stream_mode=stream_mode,
            temporary=temporary,
            after_seconds=after_seconds,
            if_not_exists=if_not_exists,
            stream_resumable=stream_resumable,
            run_create_ms=(
                int(time.time() * 1_000) - request_start_time
                if request_start_time
                else None
            ),
            run_put_ms=int((time.time() - put_time_start) * 1_000),
            checkpoint_id=str(checkpoint_id),
        )
        # 以下if分支后继再分析
        if multitask_strategy in ("interrupt", "rollback") and inflight_runs:
            with contextlib.suppress(HTTPException):
                # if we can't find the inflight runs again, we can proceeed
                await Runs.cancel(
                    conn,
                    [run["run_id"] for run in inflight_runs],
                    thread_id=thread_id_,
                    action=multitask_strategy,
                )
        return first #直接返回run数据
    elif multitask_strategy == "reject":
        raise HTTPException(
            status_code=409,
            detail="Thread is already running a task. Wait for it to finish or choose a different multitask strategy.",
        )
    else:
        raise NotImplementedError
 

       Runs.put方法在langgraph_runtime_postgres目录下的ops.py文件中，源码如下：

    本方法主要逻辑如下：

    1）查询线程数据保存到临时表run_thread

    2）计算run_thread和assistant的笛卡尔积，并筛选当前会话thread_id和assistant_id相同的记录

    3）根据以上的查询结果和传入的参数组织数据插入到run表中，状态为pending，结果保存在inserted_run中

    4）对insert_run和assistant做inner join，并与传入的参数一起作为物料更新thread表的配置数据和元数据，并修改线程状态为busy

    5）执行完以上处理后，在reids的LIST_RUN_QUEUE队列中插入[1]，从而唤醒worker

    @staticmethod
    async def put(
        conn: AsyncConnection[DictRow],
        assistant_id: UUID,
        kwargs: dict,
        *,
        thread_id: UUID | None = None,
        user_id: str | None = None,
        run_id: UUID | None = None,
        status: RunStatus | None = "pending",
        metadata: MetadataInput,
        prevent_insert_if_inflight: bool,
        multitask_strategy: MultitaskStrategy = "reject",
        if_not_exists: IfNotExists = "reject", #if_not_exists="reject"
        after_seconds: int = 0,#传入参数为0，所以立刻唤醒worker
        ctx: Auth.types.BaseAuthContext | None = None,
    ) -> AsyncIterator[Run]:
        ……        

        #上面的非核心代码暂不关注
        if FF_RICH_THREADS:

            """

            下面的SQL把参数中的thread_id、"busy"和assistant表中的字段拼接成
            的metadata和config插入到thread表中，如果表中已经有线程记录，

            则执行更新操作

            """
            insert_thread_sql = """
                INSERT INTO thread (thread_id, status, metadata, config)
                SELECT
                    %(thread_id)s,
                    'busy',
                    jsonb_build_object(
                        'graph_id', assistant.graph_id,
                        'assistant_id', assistant.assistant_id
                    ) ||  coalesce(%(config)s::jsonb -> 'metadata', '{}') || %(metadata)s::jsonb,
                    assistant.config
                    || %(config)s::jsonb
                    || jsonb_build_object(
                        'configurable',
                            coalesce((assistant.config -> 'configurable'), '{}')
                        )
                FROM assistant
                WHERE assistant_id = %(assistant_id)s
                ON CONFLICT (thread_id) DO UPDATE
                    -- Return existing thread; otherwise the CTE below could return nothing
                    -- if there's a conflict here in an uncommitted transaction in another worker
                    -- but it hasn't been inserted to be available for the SELECT statement below
                    SET status = 'busy'
                RETURNING thread.*
            """
        else:
            ……//暂不关注
        thread_query_cte = (
            f"""WITH run_thread AS ({insert_thread_sql}), """
            if thread_id is None
            else (
                f"""WITH inserted_thread AS (
                {insert_thread_sql}
            ),
            run_thread AS (
                SELECT * FROM thread where thread_id = %(thread_id)s {filter_clause}
                UNION
                SELECT * FROM inserted_thread
                LIMIT 1
            ),
            {ttl_insert_query}"""
                if if_not_exists == "create"

                """

               正常情况走这里else分支。

               查询线程表，并保存到临时表run_thread表中，供inserted_run调用

                thread_query_cte = 'WITH run_thread AS (
                        SELECT * FROM thread  WHERE thread_id = %(thread_id)s
                             {filter_clause}),   {ttl_insert_query}'

                """
                else f"""WITH run_thread AS (
                        SELECT * FROM thread 
                        WHERE thread_id = %(thread_id)s
                             {filter_clause}),
                             {ttl_insert_query}"""
            )
        )

        params = {
            "multitask_strategy": multitask_strategy,
            "run_id": run_id or uuid7(),
            "thread_id": thread_id or uuid4(),
            "assistant_id": assistant_id,
            "metadata": Jsonb(metadata),
            "kwargs": Jsonb(kwargs),
            "config": Jsonb(kwargs.get("config")),
            "status": status,
            "user_id": user_id,
            "after_seconds": f"{after_seconds} second",
            "strategy": strategy,
            "env_strategy": env_strategy,
            "ttl_interval": ttl_interval_minutes,
            "env_ttl_interval": env_ttl_interval_minutes,
        }
        params.update(filter_params)

        #needs_inflight为False

        needs_inflight = thread_id is not None and (
            multitask_strategy in ("rollback", "interrupt")
            or prevent_insert_if_inflight
        )
        query = thread_query_cte

        if needs_inflight:
            query += f"""
                inflight_runs AS (
                    SELECT run.*
                    FROM run
                    {thread_join}
                    WHERE thread_id = %(thread_id)s AND run.status in ('pending', 'running') {filter_clause}
                ),
            """

        query += (
            """
inserted_run AS (#insert_run供updated_thread调用。数据插入run表
    INSERT INTO run (run_id, thread_id, assistant_id, metadata, status, kwargs, multitask_strategy, created_at)
    SELECT #从函数参数及后面的笛卡尔积中选择插入到run表的字段
        %(run_id)s,
        thread_id,
        assistant_id,
        %(metadata)s,
        %(status)s,
        %(kwargs)s::jsonb || jsonb_build_object(#记住kwargs包括了对话请求数据
            'config', assistant.config || run_thread.config || %(config)s::jsonb || jsonb_build_object(
                'configurable',
                    coalesce((assistant.config -> 'configurable'), '{}') ||
                    coalesce((run_thread.config -> 'configurable'), '{}') ||
                    coalesce(%(config)s::jsonb -> 'configurable', '{}') ||
                    jsonb_build_object(
                        'run_id', %(run_id)s::text,
                        'thread_id', thread_id,
                        'graph_id', graph_id,
                        'assistant_id', assistant_id,
                        'user_id', coalesce(
                            %(config)s::jsonb -> 'configurable' ->> 'user_id',
                            run_thread.config -> 'configurable' ->> 'user_id',
                            assistant.config -> 'configurable' ->> 'user_id',
                            %(user_id)s::text
                        )
                    ),
                'metadata',
                    assistant.metadata || run_thread.metadata || coalesce(%(config)s::jsonb -> 'metadata', '{}') || %(metadata)s
            ),
            'context', coalesce(assistant.context, '{}') || coalesce(%(kwargs)s::jsonb -> 'context', '{}')
        ),
        %(multitask_strategy)s,
        now() + %(after_seconds)s::interval

    """

        run_thread临时表保存的是当前线程信息，assistant保存的是助手信息。

       计算两个表的笛卡尔积，并从中选择与参数中的thread_id和assistant_id相同的记录,

        这个结果作为SELECT的源数据

    """
    FROM run_thread
    CROSS JOIN assistant 
    WHERE thread_id = %(thread_id)s
        AND assistant_id = %(assistant_id)s"""
            + (
                " AND NOT EXISTS (SELECT 1 FROM inflight_runs)"
                if prevent_insert_if_inflight and thread_id is not None
                else ""
            )
            + """ RETURNING run.*
)"""
        )
        if FF_RICH_THREADS:
            query += """

               ,

updated_thread AS (#以上面插入的run数据作为物料更新线程元数据和配置数据
    UPDATE thread SET
        metadata = jsonb_set(
            jsonb_set(thread.metadata, '{graph_id}', to_jsonb(assistant.graph_id)),
            '{assistant_id}',
            to_jsonb(assistant.assistant_id)
        ),
        config = assistant.config
            || thread.config
            || %(config)s::jsonb
            || jsonb_build_object(
                'configurable',
                    coalesce((assistant.config -> 'configurable'), '{}') ||
                    coalesce(thread.config -> 'configurable', '{}')
            ),
        status = 'busy'

    """

        以下SQL内部连接刚刚插入的run记录和assistant，并且当前线程状态不是'busy'

    """
    FROM inserted_run
    INNER JOIN assistant
        ON assistant.assistant_id = inserted_run.assistant_id
    WHERE
        thread.thread_id = inserted_run.thread_id  AND thread.status != 'busy'
)"""
        if needs_inflight:
            query += """
                SELECT * FROM inserted_run
                UNION ALL
                SELECT * FROM inflight_runs
            """
        else:
            query += """#返回对应的run记录
                SELECT * FROM inserted_run 
            """
        cur = await conn.execute(query, params, binary=True)#异步执行以上的SQL操作

        async def consume() -> AsyncIterator[Run]:#处理执行批量SQL后返回的结果
            async for row in cur:
                yield row
                if row["run_id"] == run_id: #确保是当前run的数据
                    # inserted run, notify queue
                    if not after_seconds: #当前走本分支，马上唤醒worker
                        await wake_up_worker() #实际是在redis的LIST_RUN_QUEUE队列中插入[1]
                    else:
                        create_task(wake_up_worker(after_seconds))

        return consume()

        

        前面已经完成stream_run方法前半部分的分析，下面分析Runs.Stream.join方法，源码如下：

       #直接看最核心的代码，把流程贯穿起来。

       #直接从订阅频道获取数据，并以流式返回。

        @staticmethod
        async def join(
            run_id: UUID,
            *,
            stream_channel: StreamHandler,
            thread_id: UUID,
            ignore_404: bool = False,
            cancel_on_disconnect: bool = False,
            stream_mode: StreamMode | list[StreamMode] | None = None,
            last_event_id: str | None = None,
            ctx: Auth.types.BaseAuthContext | None = None,
        ) -> AsyncIterator[tuple[bytes, bytes, bytes | None]]:

            """Stream the run output, either from a stream handler or a stream mode."""
            start_time = datetime.now(UTC)
            await Runs.Stream.check_run_stream_auth(run_id, thread_id, ctx)

            pubsub: StreamHandler = stream_channel
            try:
                logger.info(
                    "Joined run stream",
                    run_id=str(run_id),
                    thread_id=str(thread_id),
                    cancel_on_disconnect=cancel_on_disconnect,
                )

                #状态检查。如果流已经结束，则清空流，否则设置下一次超时时间为0.1秒

                run_status_string = RUN_STATUS_STRING.format(thread_id, run_id)
                if value := await get_redis().get(run_status_string):
                    if value == b"done":
                        # if already done, we can drain the stream
                        timeout: int | float = DRAIN_TIMEOUT
                    else:
                        timeout = WAIT_LESS_TIMEOUT
                else:#如果为从redis获取到流状态，则设置下一次超时时间为0.1秒
                    # Start with shorter timeout for cases after the control channel has expired or the run doesn't exist
                    timeout = WAIT_LESS_TIMEOUT
                highest_stream_id: str | None = None

                ……#中间代码省略
                # stream events
                while True:

                    #直接从订阅的频道获取数据，保存到store中
                    if store := await pubsub.get_message(timeout=timeout):
                        if LOG_LEVEL_DEBUG:
                            await logger.adebug(
                                "Received redis stream event",
                                run_id=str(run_id),
                                type=store["type"],
                                channel=store["channel"],
                                data=store["data"],
                            )
                        if store["type"].encode() in pubsub.SUBUNSUB_MESSAGE_TYPES:
                            # This is a subscription message, not a data message
                            pass
                        else:
                            decoded = decode_stream_message(
                                store["data"], channel=store["channel"]
                            )
                            event = decoded.event_bytes
                            event_name = event.decode("utf-8")
                            message = decoded.message_bytes
                            len_str = decoded.stream_id_bytes
                            if event_name == "control":
                                if message == b"done":
                                    timeout = DRAIN_TIMEOUT
                                else:
                                    timeout = WAIT_LESS_TIMEOUT

                            #因为未传入stream_mode，所以走这个分支
                            elif (
                                not stream_mode
                                or event_name in stream_mode
                                or (
                                    (
                                        "messages" in stream_mode
                                        or "messages-tuple" in stream_mode
                                    )
                                    and event_name.startswith("messages")
                                )
                            ):
                                yield (#这里就是返回前端的流式数据
                                    event,
                                    message,
                                    len_str,
                                )
                    elif timeout == DRAIN_TIMEOUT:#如果流已传输完成，则跳出循环
                        break
                    else:#重新从订阅频道中获取数据
                        async with connect() as conn:
                            run_iter = await Runs.get(
                                conn, run_id, thread_id=thread_id, ctx=ctx
                            )
                            run = await anext(run_iter, None)

                         #如果未获取到run数据或者状态不是'pending'或'running'，则设置超时为

                         #DRAIN_TIMEOUT
                        if run is None or run["status"] not in (
                            "pending",
                            "running",
                        ):
                            timeout = DRAIN_TIMEOUT

                       #如果后台任务正在运行，但从订阅频道中未获取到事件信息，

                       #则设置超时时间为5秒
                        else:
                            timeout = WAIT_TIMEOUT

                        if run is None and not ignore_404:
                            yield (
                                b"error",
                                json_dumpb(
                                    HTTPException(
                                        status_code=404,
                                        detail=f"Run with ID '{run_id}' not found. Please verify the ID is correct and the run hasn't been deleted or expired.",
                                    )
                                ),
                                None,
                            )
            except asyncio.CancelledError:
                if cancel_on_disconnect:
                    create_task(cancel_run(thread_id, run_id))

                # Don't do anything before cancelling the run to minimize race conditions
                elapsed_time = (datetime.now(UTC) - start_time).total_seconds()
                await logger.awarning(
                    f"Client disconnected after {elapsed_time} seconds. Consider adjusting the client or network timeouts if this is unexpected.",
                    run_id=str(run_id),
                    elapsed_time=elapsed_time,
                )
                raise

       3.处理流程

       在一次对话处理中主要流程如下图所示：

        到这里已经完成了一次对话请求和应答的分析，先订阅该线程本次会话的频道，然后从频道中获取发布的数据，并以流式返回。但agent的应答数据是如何发布到对应频道的呢？这个问题在下一篇博文中讲解。