LangGraph 源码学习(1)

fake_chat

fake_chat 是 LangGraph 官方写好的模拟 LLM，可以让我们快速学习模型接口。

在文件开始，导入了正则表达式、迭代器、异步等功能包以及 LangChain 的核心。接下来创建了一个继承了 GenericFakeChatModel 的类，名为 FakeChatModel。成员变量有：

# 保存模型要返回的信息
messages: list[BaseMessage]

# 当前返回第几个信息
i: int = 0

bind_tools 函数表示绑定工具，Fake 模型不支持工具所以直接返回自己。

1 2	def bind_tools(self, functions: list): return self

_generate() 函数是 ChatModel 最核心的方法，当外部调用 model.invoke() 时，最终会调用 _generate()

def _generate(
    self,
    messages: list[BaseMessage],
    stop: list[str] | None = None,
    run_manager: CallbackManagerForLLMRun | None = None,
    **kwargs: Any,
) -> ChatResult:
    """Top Level call"""
    if self.i >= len(self.messages):
        self.i = 0
    message = self.messages[self.i]
    self.i += 1
    if isinstance(message, str):
        message_ = AIMessage(content=message)
    else:
        if hasattr(message, "model_copy"):
            message_ = message.model_copy()
        else:
            message_ = message.copy()
    generation = ChatGeneration(message=message_)
    return ChatResult(generations=[generation])

这里只是从消息列表中依次读取消息，然后判断是否为字符串，然后转换为 LangChain 能接收的 Message 类型。最终通过 ChatGeneration() 函数生成结果。

_stream() 用于流式输出，对应的是 model.stream() 函数。但其本质上是先调用 generate 函数拿到完整答案，然后获取 AIMessage，再获取文本。再通过正则表达式拆分文本：

content_chunks = re.split(
    r"(\s)",
    content
)

接下来创建 chunk，将拆分好的文本存入不同的 chunk。通过 run_manager.on_llm_new_token() 函数通知前端新 Token 来了，方便前端显示。整个函数完整代码如下：

def _stream(
    self,
    messages: list[BaseMessage],
    stop: list[str] | None = None,
    run_manager: CallbackManagerForLLMRun | None = None,
    **kwargs: Any,
) -> Iterator[ChatGenerationChunk]:
    """Stream the output of the model."""
    chat_result = self._generate(
        messages, stop=stop, run_manager=run_manager, **kwargs
    )
    if not isinstance(chat_result, ChatResult):
        raise ValueError(
            f"Expected generate to return a ChatResult, "
            f"but got {type(chat_result)} instead."
        )

    message = chat_result.generations[0].message

    if not isinstance(message, AIMessage):
        raise ValueError(
            f"Expected invoke to return an AIMessage, "
            f"but got {type(message)} instead."
        )

    content = message.content

    if content:
        # Use a regular expression to split on whitespace with a capture group
        # so that we can preserve the whitespace in the output.
        assert isinstance(content, str)
        content_chunks = cast(list[str], re.split(r"(\s)", content))

        for i, token in enumerate(content_chunks):
            if i == len(content_chunks) - 1:
                chunk = ChatGenerationChunk(
                    message=AIMessageChunk(
                        content=token, id=message.id, chunk_position="last"
                    )
                )
            else:
                chunk = ChatGenerationChunk(
                    message=AIMessageChunk(content=token, id=message.id)
                )
            if run_manager:
                run_manager.on_llm_new_token(token, chunk=chunk)
            yield chunk
    else:
        args = message.__dict__
        args.pop("type")
        chunk = ChatGenerationChunk(
            message=AIMessageChunk(**args, chunk_position="last")
        )
        if run_manager:
            run_manager.on_llm_new_token("", chunk=chunk)
        yield chunk

AgentState

AgentState 是整个 LangGraph 的核心，Agent 运行过程中的所有信息都存在这个字典里：

class AgentState(TypedDict):
    """The state of the agent."""

    messages: Annotated[Sequence[BaseMessage], add_messages]

    remaining_steps: NotRequired[RemainingSteps]

messages：完整对话历史。Annotated[…, add_messages] 是重点 —— add_messages 是 reducer，规定”新值如何合并进旧值”。因为没有 reducer 时，新消息会覆盖旧消息；有了它，新消息会 append 追加到历史里。
remaining_steps：剩余步数上限（防死循环）。

create_react_agent

这个函数是 LangGraph 里的核心函数之一，它的作用是根据一个大模型 + 工具集合，自动创建一个 ReAct Agent 的状态图（StateGraph），最后返回一个可以执行的 Agent。其函数入口如下：

def create_react_agent(
    model: str
    | LanguageModelLike
    | Callable[[StateSchema, Runtime[ContextT]], BaseChatModel]
    | Callable[[StateSchema, Runtime[ContextT]], Awaitable[BaseChatModel]]
    | Callable[
        [StateSchema, Runtime[ContextT]], Runnable[LanguageModelInput, BaseMessage]
    ]
    | Callable[
        [StateSchema, Runtime[ContextT]],
        Awaitable[Runnable[LanguageModelInput, BaseMessage]],
    ],
    tools: Sequence[BaseTool | Callable | dict[str, Any]] | ToolNode,
    *,
    prompt: Prompt | None = None,
    response_format: StructuredResponseSchema
    | tuple[str, StructuredResponseSchema]
    | None = None,
    pre_model_hook: RunnableLike | None = None,
    post_model_hook: RunnableLike | None = None,
    state_schema: StateSchemaType | None = None,
    context_schema: type[Any] | None = None,
    checkpointer: Checkpointer | None = None,
    store: BaseStore | None = None,
    interrupt_before: list[str] | None = None,
    interrupt_after: list[str] | None = None,
    debug: bool = False,
    version: Literal["v1", "v2"] = "v2",
    name: str | None = None,
    **deprecated_kwargs: Any,
) -> CompiledStateGraph:

其中有三个关键部分：model、tools、返回值。

model

在这个函数中，model 可以是很多种形式：
1、字符串，LangGraph 内部根据字符串帮你创建模型。
2、直接传模型对象，任何满足 LangChain Runnable 接口的模型都可以。
3、Callable 模型。例如：

Callable[
[
StateSchema,
Runtime[ContextT]
],
BaseChatModel
]

意为 model 也可以是一个函数。其具体语法如下：

# 一个函数 输入 int，int 输出 int
Callable[[int, int], int]

def add(
    a:int,
    b:int
)->int:
    return a+b

这样做可以让 Agent 动态选择模型，简单问题和复杂问题用不同的模型回答。

tools

第二个参数 tools 表示工具，它也可以是很多种类型。
1、BaseTool，LangChain 标准工具。例如：

1
2
3

@tool
def search(query:str):
    return result

2、普通函数。例如：

1 2	def add(a,b): return a+b

3、dict，工具描述，例如

{
"name":"search",
"description":"搜索网页",
"parameters":{}
}

4、ToolNode，是 LangGraph 的工具节点。

返回值 CompiledStateGraph

这里返回的并不是 Agent 对象，而是编译好的状态图，之后可以直接使用 invoke 函数。

其他参数

prompt：Agent 的系统提示词
response_format：输出格式
pre_model_hook：模型调用之前执行…
post_model_hook：模型调用之后执行…
state_schema：定义 Agent 状态
context_schema：上下文
checkpointer：检查点，用于保存 Agent 状态
store：用于长期储存
interrupt_before：执行前暂停
interrupt_after：执行后暂停

工具归一化

在 create_react_agent 函数中，会将各种形式的 tools 统一成两种：

ict 形式 → llm_builtin_tools（模型内置工具 schema，直接传给模型用）
其他 (函数 / basetool) → 包进一个 ToolNode，从它提取 tool_classes

if isinstance(tools, ToolNode):
    tool_classes = list(tools.tools_by_name.values())
    tool_node = tools
else:
    llm_builtin_tools = [t for t in tools if isinstance(t, dict)]
    tool_node = ToolNode([t for t in tools if not isinstance(t, dict)])
    tool_classes = list(tool_node.tools_by_name.values())

call_model

在 LangGraph 工作流中，这个函数负责调用 LLM，并把 AI 回复写回 State。其整体流程如下：

def call_model(state,runtime,config):

    # 1. 从state拿消息
    input = get_messages(state)


    # 2. 如果动态选择模型
    if dynamic:
        model = choose_model(state,runtime)


    # 3. 调用LLM
    response = model.invoke(input)


    # 4. 标记Agent名字
    response.name=name


    # 5. 检查是否超步骤
    if too_many_steps:
        return error_message


    # 6. 更新State
    return {
        "messages":[response]
    }

注意最后一行，节点并不直接修改 state，而是返回一个 dict 表示对 state 的更新。LangGraph 会拿这个 dict 和旧 state 合并。

should_continue

这个函数负责判断 LLM 下一步应该干什么：结束？调用工具？结构化输出？进入后处理？其核心判断代码如下：

# Define the function that determines whether to continue or not
def should_continue(state: StateSchema) -> str | list[Send]:
    messages = _get_state_value(state, "messages")
    last_message = messages[-1]
    # If there is no function call, then we finish
    if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
        if post_model_hook is not None:
            return "post_model_hook"
        elif response_format is not None:
            return "generate_structured_response"
        else:
            return END
    # Otherwise if there is, we continue
    else:
        if version == "v1":
            return "tools"
        elif version == "v2":
            if post_model_hook is not None:
                return "post_model_hook"
            return [
                Send(
                    "tools",
                    ToolCallWithContext(
                        __type="tool_call_with_context",
                        tool_call=call,
                        state=state,
                    ),
                )
                for call in last_message.tool_calls
            ]

这段代码检查最后一条 AIMessage 有没有 tool_calls，没有就结束，有的话就将每个工具调用单独发给一个 tools 节点（并行）。在判断完成后就会创建新的图，其核心代码如下：

workflow = StateGraph(state_schema=state_schema, context_schema=context_schema)
workflow.add_node("agent", RunnableCallable(call_model, acall_model), input_schema=input_schema)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")                    # 入口
workflow.add_conditional_edges("agent", should_continue, path_map=agent_paths)  # ★条件边
workflow.add_edge("tools", entrypoint)               # 工具执行完回到 agent
return workflow.compile(checkpointer=..., store=...)

其整体逻辑如下图。

┌──────────► agent (调 LLM) ──────────┐
│              │                      │
│              │ should_continue      │ (无 tool_calls → END)
│              ▼                      ▼
│         "tools" (执行工具)         END
└──────────────┘

这就是 ReAct 循环：agent → 有工具调用吗？→ 有则执行 tools → 回到 agent → 直到没有工具调用为止。

StateGraph

打开 state.py，可以看到 StateGraph 类下的注释：

节点签名是 State -> Partial：输入整个状态，输出部分状态更新。
每个 key 可以用 reducer 注解，签名 (Value, Value) -> Value。
StateGraph 是 builder，不能直接执行，必须 .compile()。
.compile() 之后支持 invoke() / stream() / ainvoke() / astream()。

其核心属性如下：

edges: set[tuple[str, str]] # 静态边: {(来源, 目标)}
nodes: dict[str, StateNodeSpec] # 节点: {名字: 节点规范}
branches: defaultdict[str, dict[str, BranchSpec]] # 条件边: {节点: {分支名: 分支}}
channels: dict[str, BaseChannel] # 状态通道
schemas: dict[type, dict[str, …]] # 每个 schema 的通道

这就是”图纸”的完整数据模型：节点 + 边 + 条件边，全部存起来，compile 时才变成可执行结构。

ToolNode 与 tools_condition

tools_condition 是一个路由函数，用于判断 LLM 是否产生了模型调用，其核心判断逻辑如下：

def tools_condition(state, messages_key="messages") -> Literal["tools", "__end__"]:
    if isinstance(state, list):
        ai_message = state[-1]
    elif (isinstance(state, dict) and (messages := state.get(messages_key, []))) or \
         (messages := getattr(state, messages_key, [])):
        ai_message = messages[-1]
    else:
        raise ValueError(f"No messages found in input state to tool_edge: {state}")
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"       # 有工具调用 → 去执行工具
    return "__end__"         # 没有 → 结束

它兼容三种格式状态：list / dict / BaseModel，检查最后一句话是否有 tool_calls。其实这个函数与 should_continue 是同一模式，只是这个函数更加精简。

ToolNode 是一个大类，其输入可以有三种格式：①图状态（有 messages 键）；②消息列表；③直接的工具调用列表。输出是 {“messages”: [ToolMessage(…)]} 或 [ToolMessage(…)]。

不同工具的结果通过 _combine_tool_outputs 函数汇总，多个工具并行执行完，把结果收集成一个 list。普通 ToolMessage 直接包成 {messages: […]} 返回（配合 add_messages reducer 追加进历史）。有 Command 时走复杂路径：合并 Command(goto=…)

小结：ToolNode 负责”执行工具 + 把结果转成 ToolMessage 塞回状态”。tools_condition 负责”要不要执行工具”。两者配合就是完整的 ReAct 闭环。

练习

目标

使用今天看过的东西复现假 ReAct 模型。理解 ReAct 的循环。

代码

完整代码如下：

from langgraph.prebuilt.chat_agent_executor import create_react_agent
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, BaseMessage
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.tools import tool

class MyFakeChatModel(GenericFakeChatModel):
    def bind_tools(self, functions: list):
        return self

model = MyFakeChatModel(messages=iter([
    AIMessage(content="", tool_calls=[{"id": "1", "name": "search_weather", "args": {"city": "sf"}}]),
    AIMessage(content="天气很好"),
]))

@tool
def search_weather(city: str):
    """查询城市天气"""
    return f"result for {city}"

client = create_react_agent(model, tools=[search_weather])
result = client.invoke({"messages":[HumanMessage(content="你好")]})

print("=== 最终对话历史 ===")
for m in result["messages"]:
    print(f"  {type(m).__name__}: content={m.content!r} tool_calls={getattr(m, 'tool_calls', None)!r}")
print(f"=== 消息条数: {len(result['messages'])} ===")

最终输出如下：

=== 最终对话历史 ===
  HumanMessage: content='你好' tool_calls=None
  AIMessage: content='' tool_calls=[{'name': 'search_weather', 'args': {'city': 'sf'}, 'id': '1', 'type': 'tool_call'}]
  ToolMessage: content='result for sf' tool_calls=None
  AIMessage: content='天气很好' tool_calls=[]
=== 消息条数: 4 ===