create_agent 还是 LangGraph？同一任务两种实现的选型决策表

"""Implementation A: the support lookup as create_agent (one call)."""
from langchain.agents import create_agent
from langchain_core.messages import AIMessage
 
from shared import ScriptedChatModel, lookup_order, summarize
 
 
def build():
    model = ScriptedChatModel(
        script=[
            AIMessage(content="", tool_calls=[{"name": "lookup_order", "args": {"order_id": "A1"}, "id": "c1"}]),
            AIMessage(content="Order A1 is shipped, total 99.0."),
        ]
    )
    return create_agent(model, tools=[lookup_order], system_prompt="You are a support assistant.")
 
 
def run():
    result = build().invoke({"messages": [("user", "Where is order A1?")]})
    return summarize(result["messages"])
 
 
if __name__ == "__main__":
    print("\n".join(run()))

"""Implementation B: the same loop written by hand as a LangGraph StateGraph."""
from langchain_core.messages import AIMessage, SystemMessage
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
 
from shared import ScriptedChatModel, lookup_order, summarize
 
 
def build():
    model = ScriptedChatModel(
        script=[
            AIMessage(content="", tool_calls=[{"name": "lookup_order", "args": {"order_id": "A1"}, "id": "c1"}]),
            AIMessage(content="Order A1 is shipped, total 99.0."),
        ]
    )
    llm = model.bind_tools([lookup_order])
 
    def call_model(state: MessagesState):
        prompt = [SystemMessage(content="You are a support assistant.")] + state["messages"]
        return {"messages": [llm.invoke(prompt)]}
 
    builder = StateGraph(MessagesState)
    builder.add_node("model", call_model)
    builder.add_node("tools", ToolNode([lookup_order]))
    builder.add_edge(START, "model")
    builder.add_conditional_edges("model", tools_condition, {"tools": "tools", END: END})
    builder.add_edge("tools", "model")
    return builder.compile()
 
 
def run():
    result = build().invoke({"messages": [("user", "Where is order A1?")]})
    return summarize(result["messages"])
 
 
if __name__ == "__main__":
    print("\n".join(run()))

"""Run both implementations and compare their message traces and code size."""
from pathlib import Path
 
import agent_create
import agent_graph
 
 
def code_lines(name: str) -> int:
    lines = Path(__file__).with_name(name).read_text(encoding="utf-8").splitlines()
    return sum(1 for line in lines if line.strip() and not line.strip().startswith("#"))
 
 
trace_a = agent_create.run()
trace_b = agent_graph.run()
print("trace A (create_agent):")
print("\n".join(f"  {line}" for line in trace_a))
print("identical to trace B (StateGraph):", trace_a == trace_b)
print(f"non-blank lines: agent_create.py={code_lines('agent_create.py')}, agent_graph.py={code_lines('agent_graph.py')}")

trace A (create_agent):
  human: Where is order A1?
  ai: tool_calls=[('lookup_order', {'order_id': 'A1'})]
  tool: A1: shipped, total 99.0
  ai: Order A1 is shipped, total 99.0.
identical to trace B (StateGraph): True
non-blank lines: agent_create.py=17, agent_graph.py=28

"""create_agent + HumanInTheLoopMiddleware: pause before a side-effecting tool, then approve or reject."""
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_core.messages import AIMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
 
from shared import REFUNDS, ScriptedChatModel, lookup_order, refund
 
 
def build():
    model = ScriptedChatModel(
        script=[
            AIMessage(content="", tool_calls=[{"name": "refund", "args": {"order_id": "A1", "amount": 99.0}, "id": "c1"}]),
            AIMessage(content="Handled."),
        ]
    )
    return create_agent(
        model,
        tools=[lookup_order, refund],
        middleware=[HumanInTheLoopMiddleware(interrupt_on={"refund": True, "lookup_order": False})],
        checkpointer=InMemorySaver(),  # interrupts need a checkpointer to persist the paused state
    )
 
 
def scenario(name: str, decision: dict) -> None:
    REFUNDS.clear()
    agent = build()
    config = {"configurable": {"thread_id": name}}
    paused = agent.invoke({"messages": [("user", "Refund order A1")]}, config)
    request = paused["__interrupt__"][0].value["action_requests"][0]
    print(f"[{name}] paused before: {request['name']}({request['args']}); refunds so far: {REFUNDS}")
    print(f"[{name}] saved next step: {agent.get_state(config).next}")
    done = agent.invoke(Command(resume={"decisions": [decision]}), config)
    print(f"[{name}] after resume: refunds={REFUNDS}; last message: {done['messages'][-1].content!r}")
 
 
scenario("approve", {"type": "approve"})
scenario("reject", {"type": "reject", "message": "Refunds over 50 need a manager."})

[approve] paused before: refund({'order_id': 'A1', 'amount': 99.0}); refunds so far: []
[approve] saved next step: ('HumanInTheLoopMiddleware.after_model',)
[approve] after resume: refunds=[('A1', 99.0)]; last message: 'Handled.'
[reject] paused before: refund({'order_id': 'A1', 'amount': 99.0}); refunds so far: []
[reject] saved next step: ('HumanInTheLoopMiddleware.after_model',)
[reject] after resume: refunds=[]; last message: 'Handled.'

"""The same review loop inside create_agent: an after_model hook that jumps back to the model."""
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import after_model
from langchain_core.messages import AIMessage, HumanMessage
 
from shared import ScriptedChatModel, lookup_order
 
MAX_RETRIES = 1
 
 
@after_model(can_jump_to=["model"])
def require_order_id(state: AgentState, runtime):
    """Deterministic review: the final answer must mention the order id, retry at most once."""
    last = state["messages"][-1]
    humans = [m for m in state["messages"] if isinstance(m, HumanMessage) and m.content.startswith("REVIEW:")]
    if last.tool_calls or "A1" in last.content or len(humans) >= MAX_RETRIES:
        return None
    return {"messages": [HumanMessage(content="REVIEW: mention the order id.")], "jump_to": "model"}
 
 
def run(drafts):
    model = ScriptedChatModel(script=[AIMessage(content=d) for d in drafts])
    agent = create_agent(model, tools=[lookup_order], middleware=[require_order_id])
    result = agent.invoke({"messages": [("user", "Can I get a refund?")]})
    return model.cursor, result["messages"][-1].content
 
 
for label, drafts in [("first draft ok", ["Order A1 is shipped."]), ("retry once", ["Sure.", "Order A1 is shipped."]), ("retries exhausted", ["No id.", "Still no id."])]:
    calls, answer = run(drafts)
    print(f"[{label}] model calls: {calls}; answer: {answer!r}")

[first draft ok] model calls: 1; answer: 'Order A1 is shipped.'
[retry once] model calls: 2; answer: 'Order A1 is shipped.'
[retries exhausted] model calls: 2; answer: 'Still no id.'

"""A staged workflow that is natural as a graph: parallel fetches, one model step, a code-based review loop.
 
fetch_order and fetch_policy run in the same step and merge into `facts` through a reducer.
The review node is plain code (no model) and routes back to `draft` at most once.
"""
import operator
from typing import Annotated, TypedDict
 
from langchain_core.messages import AIMessage
from langgraph.graph import END, START, StateGraph
 
from shared import ORDERS, ScriptedChatModel
 
MAX_RETRIES = 1
 
 
class State(TypedDict):
    question: str
    facts: Annotated[list, operator.add]  # reducer: parallel branches append instead of overwrite
    trace: Annotated[list, operator.add]
    answer: str
    attempts: int
    approved: bool
 
 
def build(drafts):
    model = ScriptedChatModel(script=[AIMessage(content=d) for d in drafts])
 
    def fetch_order(state: State):
        order = ORDERS["A1"]
        return {"facts": [f"order A1 is {order['status']}"], "trace": ["fetch_order"]}
 
    def fetch_policy(state: State):
        return {"facts": ["refund window is 30 days"], "trace": ["fetch_policy"]}
 
    def draft(state: State):
        reply = model.invoke(state["question"] + " | " + "; ".join(sorted(state["facts"])))
        return {"answer": reply.content, "attempts": state.get("attempts", 0) + 1, "trace": ["draft"]}
 
    def review(state: State):
        # Deterministic rule: the answer must mention the order id.
        return {"approved": "A1" in state["answer"], "trace": ["review"]}
 
    def route(state: State):
        if state["approved"] or state["attempts"] > MAX_RETRIES:
            return END
        return "draft"
 
    builder = StateGraph(State)
    for name, fn in [("fetch_order", fetch_order), ("fetch_policy", fetch_policy), ("draft", draft), ("review", review)]:
        builder.add_node(name, fn)
    builder.add_edge(START, "fetch_order")
    builder.add_edge(START, "fetch_policy")
    builder.add_edge(["fetch_order", "fetch_policy"], "draft")  # join: wait for both branches
    builder.add_edge("draft", "review")
    builder.add_conditional_edges("review", route, {END: END, "draft": "draft"})
    return builder.compile(), model
 
 
def run(label, drafts):
    graph, model = build(drafts)
    out = graph.invoke({"question": "Can I get a refund?", "facts": [], "trace": [], "attempts": 0})
    print(f"[{label}] trace: {' -> '.join(out['trace'])}")
    print(f"[{label}] model calls: {model.cursor}; approved: {out['approved']}; answer: {out['answer']!r}")
 
 
def show_supersteps(drafts):
    graph, _ = build(drafts)
    steps: dict = {}
    for event in graph.stream({"question": "Can I get a refund?", "facts": [], "trace": [], "attempts": 0}, stream_mode="debug"):
        if event["type"] == "task":
            steps.setdefault(event["step"], []).append(event["payload"]["name"])
    print("supersteps:", {step: sorted(names) for step, names in steps.items()})
 
 
if __name__ == "__main__":
    run("first draft ok", ["Order A1 is shipped; refunds are allowed within 30 days."])
    run("retry once", ["Sure, refunds are allowed.", "Order A1 is shipped; refunds are allowed within 30 days."])
    run("retries exhausted", ["No id here.", "Still no id."])
    show_supersteps(["Order A1 is shipped; refunds are allowed within 30 days."])

[first draft ok] trace: fetch_order -> fetch_policy -> draft -> review
[first draft ok] model calls: 1; approved: True; answer: 'Order A1 is shipped; refunds are allowed within 30 days.'
[retry once] trace: fetch_order -> fetch_policy -> draft -> review -> draft -> review
[retry once] model calls: 2; approved: True; answer: 'Order A1 is shipped; refunds are allowed within 30 days.'
[retries exhausted] trace: fetch_order -> fetch_policy -> draft -> review -> draft -> review
[retries exhausted] model calls: 2; approved: False; answer: 'Still no id.'
supersteps: {1: ['fetch_order', 'fetch_policy'], 2: ['draft'], 3: ['review']}

"""Two common HITL misconfigurations: no checkpointer, and an interrupt_on key that matches no tool."""
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_core.messages import AIMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
 
from shared import REFUNDS, ScriptedChatModel, refund
 
 
def model():
    return ScriptedChatModel(
        script=[
            AIMessage(content="", tool_calls=[{"name": "refund", "args": {"order_id": "A1", "amount": 99.0}, "id": "c1"}]),
            AIMessage(content="Handled."),
        ]
    )
 
 
print("1) interrupt_on={'refund': True} but no checkpointer")
REFUNDS.clear()
agent = create_agent(model(), tools=[refund], middleware=[HumanInTheLoopMiddleware(interrupt_on={"refund": True})])
config = {"configurable": {"thread_id": "t"}}
paused = agent.invoke({"messages": [("user", "Refund A1")]}, config)
print(f"   interrupted: {'__interrupt__' in paused}; refunds executed: {REFUNDS}")
try:
    agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config)
except Exception as exc:
    print(f"   resume fails: {type(exc).__name__}: {exc}")
 
print("2) key 'refunds' (typo) does not match the tool name 'refund'")
REFUNDS.clear()
agent = create_agent(
    model(),
    tools=[refund],
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"refunds": True})],
    checkpointer=InMemorySaver(),
)
result = agent.invoke({"messages": [("user", "Refund A1")]}, {"configurable": {"thread_id": "t"}})
print(f"   interrupted: {'__interrupt__' in result}")
print(f"   refunds executed: {REFUNDS}")

1) interrupt_on={'refund': True} but no checkpointer
   interrupted: True; refunds executed: []
   resume fails: RuntimeError: Cannot use Command(resume=...) without checkpointer
2) key 'refunds' (typo) does not match the tool name 'refund'
   interrupted: False
   refunds executed: [('A1', 99.0)]

"""Two parallel nodes write the same state key. Without a reducer LangGraph refuses the update."""
from typing import TypedDict
 
from langgraph.graph import END, START, StateGraph
 
 
class State(TypedDict):
    facts: list  # no reducer on purpose
 
 
def fetch_order(state: State):
    return {"facts": ["order A1 is shipped"]}
 
 
def fetch_policy(state: State):
    return {"facts": ["refund window is 30 days"]}
 
 
builder = StateGraph(State)
builder.add_node("fetch_order", fetch_order)
builder.add_node("fetch_policy", fetch_policy)
builder.add_edge(START, "fetch_order")
builder.add_edge(START, "fetch_policy")
builder.add_edge("fetch_order", END)
builder.add_edge("fetch_policy", END)
 
try:
    builder.compile().invoke({"facts": []})
except Exception as exc:
    print(type(exc).__name__)
    print(str(exc).splitlines()[0])

InvalidUpdateError
At key 'facts': Can receive only one value per step. Use an Annotated key to handle multiple values.