引言

上一节实现了chatbot的连续调用工具的功能，但目前流程还是

提示词→模型→解析→调用工具（可选）→输出

线性的一条路走到底（LangChain)

只适合简单线性任务（问答、摘要、简单RAG）

但是，从chatbot转向AI agent的核心区别除了工具的调用，还包括流程区别：

agent的本质是循环：“模型决定调工具→执行→结果喂回模型→再决定……直到完成”。

举例：

读日志 → 解析 ─失败→ 重试(≤3次) ─→ 退出
          │成功
          ↓
       算统计 → 对比历史 → 生成报告 → [人工审核卡点] → 发送
          ↑__________需要补数据__________|   （回跳=循环，链做不到）

agent能分析是否需要返回之前的节点，根据新增条件重新处理，且由于存在checkpoint，支持断点重跑，设置人工审核卡点，并行分支同时运行，和排查/替换问题节点。

介绍

State、Nodes、Edges是LangGraph的三块积木

State（状态）：

一个在所有节点之间流转、不断被读写的字典，装着任务的全部上下文——对话历史、中间结果、标记位。每个节点拿到它、往上写新内容、传给下一个节点。

Python里：一个 TypedDict（或Pydantic模型）

from typing import Annotated
from typing_extensions import TypedDict

def merge_list(old: list, new: list) -> list:
    return old + new                # 合并策略：旧值+新值拼接

class AgentState(TypedDict):
    messages: Annotated[list, merge_list]   # 对话历史：追加式合并
    tool_calls: int                          # 工具调用计数：直接覆盖
    need_more: bool                          # 是否继续循环的标记位

Nodes（节点）：

流程里的一步工作——调一次模型、执行一个工具、算一段数据。每个节点就是“读工作单→干活→写回工作单”的一个环节。

Python里：普通函数（不用class不用装饰器），入参是State，返回值是“要写回State的字段”：

def call_model(state: AgentState) -> dict:
    """节点1：调模型决定下一步。入参=工作单，返回=要更新的字段"""
    response = llm.invoke(state["messages"])          # 读state
    return {"messages": [response]}                   # 写回（按合并策略追加）

def run_tool(state: AgentState) -> dict:
    """节点2：执行工具"""
    result = execute(state["latest_tool_call"])
    return {"messages": [{"role": "tool", "content": result}],
            "tool_calls": state["tool_calls"] + 1}    # 计数+1（覆盖式）

Edges（边）：

节点之间的流转规则。分两种：普通边和条件边

from langgraph.graph import END

def should_continue(state: AgentState) -> str:
    """路由函数：只返回字符串——下一站去哪"""
    if state["tool_calls"] >= 5:
        return END                  # 内置终点，防死循环
    if state["need_more"]:
        return "run_tool"           # 继续干活的环
    return "summarize"              # 收尾

graph.add_conditional_edges(
    "call_model",                   # 从哪个节点出发
    should_continue,                # 用哪个函数判断
    {                               # 路由表：函数返回值→节点名（可省略）
        "run_tool": "run_tool",
        "summarize": "summarize",
    },
)

代码实现和效果展示

backend/graph_agent.py

"""LangGraph 模型/工具循环，向网页发送节点状态更新和模型文本。

整体结构（对照"工作单/工位/传送带"的比喻）：
- AgentState : 工作单——在节点之间流转的字典（对话历史+各种标记位）
- model 节点 : 工位1——调用大模型，拿到回复或"要调工具"的指令
- tools 节点 : 工位2——真正执行工具，把结果追加进对话
- 边         : 传送带——model→tools→model→... 循环，直到模型不再要工具
- stream_graph: 对外的异步生成器入口，把图执行过程的每个事件"直播"给前端
"""
import asyncio                                   # 异步编程：并发/超时工具
from contextlib import aclosing                  # 保证异步生成器退出时被正确关闭
from typing import TypedDict                     # 用类型标注定义"工作单"结构
from uuid import uuid4                           # 生成全局唯一id，标记一次运行

from langgraph.config import get_stream_writer   # 节点内往外"直播"自定义事件的通道
from langgraph.graph import END, START, StateGraph  # 图三件套：起点/终点/建图器

from . import llm_client                         # 同包的模型客户端（发HTTP请求调LLM）
from .tool_chat import (                         # 同包的工具模块
    MAX_TOOL_CALLS, TURN_TIMEOUT_SECONDS, add_usage, followup, options, prepare,
)


class AgentState(TypedDict):
    """工作单：每个节点读它、改它、传给下一个节点。

    注意：这个项目没有用 Annotated 合并策略(reducer)，
    而是每个节点返回"完整的新列表"直接覆盖旧值——两种做法之一，
    好处是不用记合并规则，代价是每个节点都要自己拼好完整列表。
    """
    messages: list[dict]    # 对话历史：[{role, content}, ...] 含工具往返消息
    remaining: int          # 剩余可用的工具调用次数（防死循环的"弹药计数"）
    usages: list[dict]      # 每次模型调用的token用量（最后汇总给前端显示）
    step: int               # 当前走到第几步（给前端显示进度条用）
    finished: bool          # 是否已结束：True=模型给出最终回答，不用再调工具


def visible_message(message):
    """仅投影对话与工具数据，不发送系统提示词或模型推理内容。

    "投影"= 只挑允许给前端看的字段，做一份脱敏副本：
    - 系统提示词(藏着调教指令)不外发
    - 超长内容截断到12000字符，防止撑爆网页
    """
    result = {'role':message['role']}            # 角色照抄：user/assistant/tool
    content = message.get('content')
    if isinstance(content, str):                 # content是纯文本时才处理
        result['content'] = content[:12000]      # 截断：最多给前12000字符
        if len(content) > 12000:
            result['content_truncated'] = True   # 打个标记，让前端知道被截了
    for key in ('tool_calls', 'tool_call_id'):   # 工具调用的结构原样保留
        if key in message:                       # （前端要展示"调了什么工具"）
            result[key] = message[key]
    return result


def build_graph(request_once=None):
    """把图搭好并编译。request_once参数允许注入"假的模型调用"——方便单元测试。"""
    request_once = request_once or llm_client.stream_llm   # 默认用真实客户端

    async def model(state):
        """工位1：调大模型。流式收token，边收边转发给前端（打字机效果）。"""
        writer = get_stream_writer()   # 拿"直播话筒"：节点内往外发自定义事件
        writer({'type':'graph', 'phase':'node_start', 'node':'model', 'step':state['step']+1})
        parts = []                     # 收集模型吐出的每一个文本碎片(delta)
        completed = None               # 流结束时会有一个complete事件，存这里
        # aclosing：保证下面的异步生成器即使中途出错也能被正确关闭
        # request_once(...)：真正发HTTP请求调模型；final参数告诉模型"最后一轮，直接给答案"
        async with aclosing(request_once(state['messages'], **options(final=state['remaining']==0))) as source:
            async for event in source:             # 逐个消费流式事件
                if event['type'] == 'delta':       # 碎片：几个字的文本
                    parts.append(event['text'])
                    writer(event)                  # 立刻转发给前端
                elif event['type'] == 'complete':  # 收尾事件：完整消息+token用量
                    completed = event
        if completed is None:                      # 流没送complete就断了=异常
            raise llm_client.LLMError('模型响应提前结束', kind='incomplete')
        message = completed.get('message')
        # 如果模型还要调工具，complete里带的是tool_calls消息(有message)；
        # 如果模型直接给文字答案，就没有message——我们手动组装一条assistant消息
        finished = message is None
        if finished:
            text = ''.join(parts)                  # 把碎片拼成完整回答
            if not text.strip():                   # 全是空白=模型抽风
                raise llm_client.LLMError('模型返回空回复', kind='empty')
            message = {'role':'assistant', 'content':text}
        usage = completed.get('usage')             # 这一轮用了多少token
        # 返回"要写回工作单的字段"（这里是全部字段，因为是覆盖式更新）
        return {
            'messages':state['messages']+[message],   # 旧历史+新消息=完整新列表
            'remaining':state['remaining'],
            'usages':state['usages']+[usage if isinstance(usage, dict) else {}],
            'step':state['step']+1, 'finished':finished,
        }

    async def tools(state):
        """工位2：执行模型要的工具调用，把工具结果追加进对话历史。"""
        writer = get_stream_writer()
        writer({'type':'graph', 'phase':'node_start', 'node':'tools', 'step':state['step']+1})
        # followup：拿着"模型最后一条消息里的tool_calls"去执行，
        # 返回新的messages列表和本轮实际消耗的工具次数used
        messages, used = await followup(
            state['messages'][:-1], state['messages'][-1], state['remaining'],
        )
        return {
            'messages':messages, 'remaining':state['remaining']-used,  # 弹药-1
            'usages':state['usages'], 'step':state['step']+1, 'finished':False,
        }   # finished=False：工具干完活还得回模型那桌，循环继续

    # ===== 组装图：注册节点 → 连边 → 编译 =====
    graph = StateGraph(AgentState)          # 用刚定义的工作单建图
    graph.add_node('model', model)          # 工位1挂上去
    graph.add_node('tools', tools)          # 工位2挂上去
    graph.add_edge(START, 'model')          # 入口：从model节点开始
    graph.add_conditional_edges('model',    # 条件边：model干完活后二选一
        lambda state: END if state['finished'] else 'tools',  # 结束→END；要工具→tools
        {'tools':'tools', END:END})         # 路由表：返回值→节点名
    graph.add_edge('tools', 'model')        # 工具完→回到model（这个环=agent循环！）
    return graph.compile()                  # 编译成可执行对象


async def stream_graph(messages, request_once=None):
    """对外主入口：async生成器。前端(SSE)逐条消费这里yield的事件。

    事件流剧本：start → (node_start/delta/node_end)×N轮 → end → complete
    """
    run_id = uuid4().hex               # 本次运行的唯一id，前端用它区分"哪一次回答"
    # 拼初始工作单：prepare()会注入系统提示词等准备工作
    initial = AgentState(
        messages=prepare(messages), remaining=MAX_TOOL_CALLS, usages=[], step=0, finished=False,
    )
    count = len(initial['messages'])   # 记住"已推送给前端的消息数"，用于算增量
    # 第一枪：告诉前端"开始了"，附上最近6条可见对话（脱敏后）和用户本轮提问
    yield {
        'type':'graph', 'phase':'start', 'run_id':run_id,
        'message_count':count, 'remaining':MAX_TOOL_CALLS,
        'messages':[visible_message(m) for m in messages[-6:] if m['role']!='system'],
        'query':messages[-1]['content'],
    }
    # 整轮截止时刻=现在+超时秒数（防模型/工具卡死把连接拖一辈子）
    deadline = asyncio.get_running_loop().time()+TURN_TIMEOUT_SECONDS
    graph = build_graph(request_once)  # 建图（每次调用现建，方便测试注入）
    last_state = initial               # 追踪最新工作单（结尾要汇总token用量）
    active = None                      # 记录"崩在哪个节点"，报错时告诉前端
    try:
        # astream：异步逐块消费图的执行过程。stream_mode两种通道：
        #   'custom' = 节点里writer()发的自定义直播(delta打字机/node_start)
        #   'updates'= 每个节点跑完后的状态更新(拿到节点返回的state)
        # recursion_limit：最多走多少步，双保险防死循环
        async with aclosing(graph.astream(
            initial, stream_mode=['custom', 'updates'], version='v2',
            config={'recursion_limit':MAX_TOOL_CALLS*2+4},
        )) as source:
            while True:
                # 超时作用域只包围读取，不跨过发往 SSE 的 yield。
                # （yield如果卡住是前端的事，不该计入模型的超时）
                async with asyncio.timeout_at(deadline):
                    chunk = await anext(source, None)   # 取下一块；None=流结束
                if chunk is None:
                    break
                if chunk['type'] == 'custom':           # ——节点内的直播事件——
                    event = dict(chunk['data'])         # 复制一份再改，别污染原数据
                    if event['type'] == 'graph':
                        event['run_id'] = run_id        # 补上运行id
                        active = event['node']          # 记住当前在哪个节点
                    yield event                         # 原样转发给前端
                elif chunk['type'] == 'updates':        # ——节点跑完的状态更新——
                    for node, state in chunk['data'].items():
                        # 推算下一站：已结束→end；刚跑完model且没结束→tools；否则→model
                        next_node = 'end' if state['finished'] else ('tools' if node=='model' else 'model')
                        new_count = len(state['messages'])
                        # 告诉前端"这个节点完工了"，只发新增的消息（增量推送）
                        yield {
                            'type':'graph', 'phase':'node_end', 'run_id':run_id,
                            'node':node, 'step':state['step'], 'next_node':next_node,
                            'before_count':count, 'message_count':new_count,
                            'remaining':state['remaining'],
                            'messages':[visible_message(m) for m in state['messages'][count:]],
                        }
                        count = new_count               # 更新"已推送"游标
                        last_state = state              # 更新最新工作单
    # ---- 收尾：三种结局 ----
        if not last_state['finished']:   # ① 图停了但没走到"完成"=中途夭折
            raise llm_client.LLMError('图执行提前结束', kind='incomplete')
        yield {'type':'graph', 'phase':'end', 'run_id':run_id}      # ② 正常收官
        yield {'type':'complete', 'usage':add_usage(last_state['usages'])}  # 汇总token账单
    except Exception as error:           # ③ 出错：先给前端发个错误事件再抛
        yield {'type':'graph', 'phase':'error', 'node':active, 'run_id':run_id}
        if isinstance(error, TimeoutError):   # 超时翻译成人话
            raise llm_client.LLMError('整轮任务执行超时', kind='timeout') from error
        raise                                 # 其他错误原样上抛给上层(SSE层)处理