LangChain v1 与 LangGraph 生产故障诊断中心：13 个真实报错对照

"""Trigger the failures an LCEL/LangGraph v1 project meets most often and print what they really look like.
 
Every case uses a scripted model, so the messages are identical on every run.
"""
import operator
import re
import sys
import warnings
from pathlib import Path
from typing import Annotated, TypedDict
 
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "agent-vs-langgraph"))
 
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command
 
from shared import REFUNDS, ScriptedChatModel, lookup_order, refund
 
 
def first_line(text: str) -> str:
    line = (text.strip().splitlines() or [""])[0]
    return re.sub(r" \(/[^)]*\)", " (<path>)", line)[:230]  # drop machine-specific paths
 
 
def case(label: str, fn) -> None:
    print(label)
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always")
        try:
            outcome = fn()
            print(f"    no exception{': ' + outcome if outcome else ''}")
        except BaseException as exc:  # noqa: BLE001 - the point is to show any failure
            print(f"    {type(exc).__name__}: {first_line(str(exc))}")
        for w in caught:
            print(f"    warning {w.category.__name__}: {first_line(str(w.message))}")
 
 
def model(*messages: AIMessage) -> ScriptedChatModel:
    return ScriptedChatModel(script=list(messages))
 
 
def tool_call(name: str, args: dict, call_id: str = "c1") -> AIMessage:
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": call_id}])
 
 
# --- migration errors -------------------------------------------------------
case("A. create_agent(prompt=...)", lambda: create_agent(model(AIMessage(content="x")), tools=[lookup_order], prompt="hi"))
case("B. from langchain.agents import AgentExecutor", lambda: exec("from langchain.agents import AgentExecutor"))
case("C. from langchain.chains import LLMChain", lambda: exec("from langchain.chains import LLMChain"))
case(
    "D. from langgraph.prebuilt import create_react_agent",
    lambda: exec("from langgraph.prebuilt import create_react_agent; create_react_agent(model(AIMessage(content='x')), [lookup_order])"),
)
 
 
# --- agent loop and tools ---------------------------------------------------
def recursion_limit():
    looping = model(*[tool_call("lookup_order", {"order_id": "A1"}, f"c{i}") for i in range(10)])
    create_agent(looping, tools=[lookup_order]).invoke({"messages": [("user", "go")]}, {"recursion_limit": 3})
 
 
@tool
def db_lookup(key: str) -> str:
    """Look up a key (always fails)."""
    raise RuntimeError("database unavailable")
 
 
def tool_raises():
    scripted = model(tool_call("db_lookup", {"key": "a"}), AIMessage(content="done"))
    create_agent(scripted, tools=[db_lookup]).invoke({"messages": [("user", "go")]})
 
 
def tool_message(scripted: ScriptedChatModel) -> str:
    result = create_agent(scripted, tools=[lookup_order]).invoke({"messages": [("user", "go")]})
    tool_msg = next(m for m in result["messages"] if m.type == "tool")
    return f"ToolMessage(status={tool_msg.status}): {' '.join(tool_msg.content.split())}"
 
 
case("E. recursion_limit too small for the loop", recursion_limit)
case("F. a tool raises RuntimeError", tool_raises)
case("G. the model omits a required tool argument", lambda: tool_message(model(tool_call("lookup_order", {}), AIMessage(content="done"))))
case("H. the model calls a tool that does not exist", lambda: tool_message(model(tool_call("no_such_tool", {}), AIMessage(content="done"))))
 
 
# --- graph and checkpoint errors --------------------------------------------
class Shared(TypedDict, total=False):
    facts: list  # no reducer on purpose
 
 
def parallel_write():
    builder = StateGraph(Shared)
    builder.add_node("a", lambda state: {"facts": ["a"]})
    builder.add_node("b", lambda state: {"facts": ["b"]})
    builder.add_edge(START, "a")
    builder.add_edge(START, "b")
    builder.add_edge("a", END)
    builder.add_edge("b", END)
    builder.compile().invoke({})
 
 
def simple_graph(checkpointer=None):
    builder = StateGraph(Shared)
    builder.add_node("a", lambda state: {"facts": ["a"]})
    builder.add_edge(START, "a")
    builder.add_edge("a", END)
    return builder.compile(checkpointer=checkpointer)
 
 
def hitl_agent(checkpointer=None):
    scripted = model(tool_call("refund", {"order_id": "A1", "amount": 9.0}), AIMessage(content="done"))
    middleware = [HumanInTheLoopMiddleware(interrupt_on={"refund": True})]
    return create_agent(scripted, tools=[refund], middleware=middleware, checkpointer=checkpointer)
 
 
def resume_without_checkpointer():
    agent = hitl_agent()
    config = {"configurable": {"thread_id": "t"}}
    agent.invoke({"messages": [("user", "refund")]}, config)
    agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config)
 
 
case("I. two parallel nodes write one key without a reducer", parallel_write)
case("J. get_state on a graph compiled without a checkpointer", lambda: simple_graph().get_state({"configurable": {"thread_id": "t"}}))
case("K. checkpointer present but no thread_id", lambda: simple_graph(InMemorySaver()).invoke({}))
case("L. resume a human-in-the-loop run without a checkpointer", resume_without_checkpointer)
case("M. resume on a thread that has no saved run", lambda: simple_graph(InMemorySaver()).invoke(None, {"configurable": {"thread_id": "new"}}))

A. create_agent(prompt=...)
    TypeError: create_agent() got an unexpected keyword argument 'prompt'
B. from langchain.agents import AgentExecutor
    ImportError: cannot import name 'AgentExecutor' from 'langchain.agents' (<path>)
C. from langchain.chains import LLMChain
    ModuleNotFoundError: No module named 'langchain.chains'
D. from langgraph.prebuilt import create_react_agent
    no exception
    warning LangGraphDeprecatedSinceV10: create_react_agent has been moved to `langchain.agents`. Please update your import to `from langchain.agents import create_agent`. Deprecated in LangGraph V1.0 to be removed in V2.0.
E. recursion_limit too small for the loop
    GraphRecursionError: Recursion limit of 3 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
F. a tool raises RuntimeError
    RuntimeError: database unavailable
G. the model omits a required tool argument
    no exception: ToolMessage(status=error): Error invoking tool 'lookup_order' with kwargs {} with error: order_id: Field required Please fix the error and try again.
H. the model calls a tool that does not exist
    no exception: ToolMessage(status=error): Error: no_such_tool is not a valid tool, try one of [lookup_order].
I. two parallel nodes write one key without a reducer
    InvalidUpdateError: At key 'facts': Can receive only one value per step. Use an Annotated key to handle multiple values.
J. get_state on a graph compiled without a checkpointer
    ValueError: No checkpointer set
K. checkpointer present but no thread_id
    ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id
L. resume a human-in-the-loop run without a checkpointer
    RuntimeError: Cannot use Command(resume=...) without checkpointer
M. resume on a thread that has no saved run
    EmptyInputError: Received no input for __start__