Deep Agents、create_agent、LangGraph 怎么选：同一任务的实测对比

"""One support loop, three implementations: compare what each puts in front of the model."""
import json
 
from deepagents import create_deep_agent
from langchain.agents import create_agent
from langchain_core.messages import AIMessage
from langchain_core.utils.function_calling import convert_to_openai_tool
 
from shared import ScriptedChatModel, build_state_graph, lookup_order
 
PROMPT = "You are a support assistant."
 
IMPLEMENTATIONS = {
    "create_agent": lambda m: create_agent(m, tools=[lookup_order], system_prompt=PROMPT),
    "StateGraph": lambda m: build_state_graph(m, [lookup_order], PROMPT),
    "create_deep_agent": lambda m: create_deep_agent(model=m, tools=[lookup_order], system_prompt=PROMPT),
}
 
for name, build in IMPLEMENTATIONS.items():
    model = ScriptedChatModel(script=[AIMessage(content="Order A1 is shipped.")])
    agent = build(model)
    result = agent.invoke({"messages": [("user", "Where is order A1?")]})
    specs = [convert_to_openai_tool(t) for t in model.bound_tools]
    system_chars = next(size for kind, size in model.calls[0] if kind == "system")
    print(name)
    print(f"  compiled object: {type(agent).__name__}")
    print(f"  graph nodes: {', '.join(n for n in sorted(agent.nodes) if n != '__start__')}")
    print(f"  tools shown to the model: {len(specs)} ({', '.join(s['function']['name'] for s in specs)})")
    print(f"  tool schema size: {sum(len(json.dumps(s)) for s in specs)} chars")
    print(f"  system message size: {system_chars} chars")
    print(f"  state keys after one run: {', '.join(sorted(result))}")

create_agent
  compiled object: CompiledStateGraph
  graph nodes: model, tools
  tools shown to the model: 1 (lookup_order)
  tool schema size: 225 chars
  system message size: 28 chars
  state keys after one run: messages
StateGraph
  compiled object: CompiledStateGraph
  graph nodes: model, tools
  tools shown to the model: 1 (lookup_order)
  tool schema size: 225 chars
  system message size: 28 chars
  state keys after one run: messages
create_deep_agent
  compiled object: CompiledStateGraph
  graph nodes: PatchToolCallsMiddleware.before_agent, _ParentNotifierMiddleware.after_agent, model, tools
  tools shown to the model: 9 (ls, read_file, write_file, edit_file, delete, glob, grep, task, lookup_order)
  tool schema size: 10736 chars
  system message size: 28 chars
  state keys after one run: files, messages

"""A tool returns a large result: how many characters does the model see on its next call?"""
from deepagents import create_deep_agent
from langchain.agents import create_agent
from langchain_core.messages import AIMessage
 
from shared import ScriptedChatModel, build_state_graph, last_chars, make_fetch_report, tool_call
 
IMPLEMENTATIONS = {
    "create_agent": lambda m, t: create_agent(m, tools=t),
    "StateGraph": build_state_graph,
    "create_deep_agent": lambda m, t: create_deep_agent(model=m, tools=t),
}
 
print("chars of the tool message the model sees on its second call")
print(f"{'result size':>12}  " + "  ".join(f"{name:>17}" for name in IMPLEMENTATIONS))
for size in (60_000, 90_000):
    seen, offloaded = [], []
    for build in IMPLEMENTATIONS.values():
        tools = [make_fetch_report(size)]
        model = ScriptedChatModel(script=[tool_call("fetch_report", {"report_id": "R1"}, "c1"), AIMessage(content="summary")])
        result = build(model, tools).invoke({"messages": [("user", "Summarize report R1")]})
        seen.append(last_chars(model, 1, "tool"))
        offloaded.append(sorted(result.get("files", {})))
    print(f"{size:>12}  " + "  ".join(f"{n:>17}" for n in seen))
    print(f"{'files in state':>12}  " + "  ".join(f"{', '.join(f) or '-':>17}" for f in offloaded))

chars of the tool message the model sees on its second call
 result size       create_agent         StateGraph  create_deep_agent
       60000              60000              60000              60000
files in state                  -                  -                  -
       90000              90000              90000               1580
files in state                  -                  -  /large_tool_results/c1

"""Delegating a heavy lookup to a subagent: what does the main conversation keep?"""
from deepagents import create_deep_agent
from langchain.agents import create_agent
from langchain_core.messages import AIMessage
 
from shared import ScriptedChatModel, last_chars, make_fetch_report, tool_call
 
REPORT_CHARS = 30_000  # below the offload threshold, so the full text stays in whichever context calls the tool
SUMMARY = "R1: database failover at 02:14, root cause was an expired certificate."
fetch_report = make_fetch_report(REPORT_CHARS)
 
# A. create_agent calls the tool itself: the whole report lands in the main conversation.
direct = ScriptedChatModel(script=[tool_call("fetch_report", {"report_id": "R1"}, "c1"), AIMessage(content=SUMMARY)])
create_agent(direct, tools=[fetch_report]).invoke({"messages": [("user", "What happened in R1?")]})
 
# B. create_deep_agent delegates to a subagent through the built-in `task` tool.
child = ScriptedChatModel(script=[tool_call("fetch_report", {"report_id": "R1"}, "k1"), AIMessage(content=SUMMARY)])
parent = ScriptedChatModel(
    script=[
        tool_call("task", {"description": "Read report R1 and summarise the root cause", "subagent_type": "analyst"}, "p1"),
        AIMessage(content=SUMMARY),
    ]
)
agent = create_deep_agent(
    model=parent,
    subagents=[
        {
            "name": "analyst",
            "description": "Reads incident reports and returns a short summary",
            "system_prompt": "You read incident reports.",
            "tools": [fetch_report],
            "model": child,
        }
    ],
)
result = agent.invoke({"messages": [("user", "What happened in R1?")]})
 
print(f"report size: {REPORT_CHARS} chars")
print(f"A. create_agent, tool called directly: main model sees a {last_chars(direct, 1, 'tool')}-char tool message")
print(f"B. deep agent, tool called by a subagent: main model sees a {last_chars(parent, 1, 'tool')}-char tool message")
print(f"   subagent model saw a {last_chars(child, 1, 'tool')}-char tool message in its own context")
print(f"   main conversation kept {len(result['messages'])} messages, subagent model made {len(child.calls)} calls")
print(f"   largest message kept in the main state: {max(len(m.content) for m in result['messages'])} chars")

report size: 30000 chars
A. create_agent, tool called directly: main model sees a 30000-char tool message
B. deep agent, tool called by a subagent: main model sees a 70-char tool message
   subagent model saw a 30000-char tool message in its own context
   main conversation kept 4 messages, subagent model made 2 calls
   largest message kept in the main state: 70 chars

"""Deep agent files live in graph state: how long do they survive?"""
from deepagents import create_deep_agent
from langchain_core.messages import AIMessage
from langgraph.checkpoint.memory import InMemorySaver
 
from shared import ScriptedChatModel, tool_call
 
WRITE = tool_call("write_file", {"file_path": "/notes.md", "content": "refund A1 after approval"}, "w1")
READ = "read_file"
 
 
def read_script(call_id: str):
    return [tool_call(READ, {"file_path": "/notes.md"}, call_id), AIMessage(content="done")]
 
 
def last_tool_text(result) -> str:
    return [m.content for m in result["messages"] if m.type == "tool"][-1].replace("\n", " | ")
 
 
def run(checkpointer):
    model = ScriptedChatModel(script=[WRITE, AIMessage(content="saved")] + read_script("r1") + read_script("r2"))
    agent = create_deep_agent(model=model, checkpointer=checkpointer)
 
    def turn(text: str, thread: str):
        config = {"configurable": {"thread_id": thread}} if checkpointer else {}
        return agent.invoke({"messages": [("user", text)]}, config)
 
    first = turn("Save a note", "t1")
    print(f"  turn 1 wrote: {sorted(first['files'])}")
    print(f"  turn 2, same thread:  {last_tool_text(turn('Read the note', 't1'))}")
    print(f"  turn 3, other thread: {last_tool_text(turn('Read the note', 't2'))}")
 
 
print("with a checkpointer:")
run(InMemorySaver())
print("without a checkpointer:")
run(None)

with a checkpointer:
  turn 1 wrote: ['/notes.md']
  turn 2, same thread:  @@ lines 1-1 of 1 @@ | refund A1 after approval
  turn 3, other thread: Error: File '/notes.md' not found
without a checkpointer:
  turn 1 wrote: ['/notes.md']
  turn 2, same thread:  Error: File '/notes.md' not found
  turn 3, other thread: Error: File '/notes.md' not found

"""Deep agent file writes behind human approval (HumanInTheLoopMiddleware via interrupt_on)."""
from deepagents import create_deep_agent
from langchain_core.messages import AIMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
 
from shared import ScriptedChatModel, tool_call
 
 
def run(decision: dict) -> None:
    model = ScriptedChatModel(
        script=[tool_call("write_file", {"file_path": "/plan.md", "content": "refund A1"}, "w1"), AIMessage(content="finished")]
    )
    agent = create_deep_agent(model=model, checkpointer=InMemorySaver(), interrupt_on={"write_file": True})
    config = {"configurable": {"thread_id": "review"}}
 
    paused = agent.invoke({"messages": [("user", "Write the refund plan")]}, config)
    request = paused["__interrupt__"][0].value["action_requests"][0]
    print(f"  paused before: {request['name']} {request['args']}")
    print(f"  files while paused: {sorted(paused['files']) or 'none'}")
 
    done = agent.invoke(Command(resume={"decisions": [decision]}), config)
    print(f"  decision: {decision['type']}")
    print(f"  files after: {sorted(done['files']) or 'none'}")
    print(f"  tool message: {[m.content for m in done['messages'] if m.type == 'tool'][0].splitlines()[0]}")
 
 
print("approve:")
run({"type": "approve"})
print("reject:")
run({"type": "reject", "message": "not allowed"})

approve:
  paused before: write_file {'file_path': '/plan.md', 'content': 'refund A1'}
  files while paused: none
  decision: approve
  files after: ['/plan.md']
  tool message: Updated file /plan.md
reject:
  paused before: write_file {'file_path': '/plan.md', 'content': 'refund A1'}
  files while paused: none
  decision: reject
  files after: none
  tool message: User rejected the tool call for `write_file` with reason: not allowed

"""A model that never stops calling tools: where does each implementation cut it off?"""
from deepagents import create_deep_agent
from langchain.agents import create_agent
from langgraph._internal._config import DEFAULT_RECURSION_LIMIT  # private module: fine for a pinned lab
from langgraph.errors import GraphRecursionError
 
from shared import ScriptedChatModel, build_state_graph, lookup_order, tool_call
 
CAP = 25
TOOLS = [lookup_order]
IMPLEMENTATIONS = {
    "create_agent": lambda m: create_agent(m, tools=TOOLS),
    "StateGraph": lambda m: build_state_graph(m, TOOLS),
    "create_deep_agent": lambda m: create_deep_agent(model=m, tools=TOOLS),
}
 
 
def looping_model() -> ScriptedChatModel:
    return ScriptedChatModel(script=[tool_call("lookup_order", {"order_id": "A1"}, "c")], loop=True)
 
 
def model_calls_before_stop(build, limit: int) -> int:
    model = looping_model()
    try:
        build(model).invoke({"messages": [("user", "Check order A1 forever")]}, {"recursion_limit": limit})
    except GraphRecursionError:
        return len(model.calls)
    raise AssertionError("the loop was expected to hit the recursion limit")
 
 
print("recursion_limit each implementation starts with")
print(f"  create_agent: {create_agent(looping_model(), tools=TOOLS).config['recursion_limit']}")
print(f"  create_deep_agent: {create_deep_agent(model=looping_model(), tools=TOOLS).config['recursion_limit']}")
print(f"  StateGraph (LangGraph default): {int(DEFAULT_RECURSION_LIMIT)}")
print(f"model calls before the loop is stopped with recursion_limit={CAP} in the call config")
for name, build in IMPLEMENTATIONS.items():
    print(f"  {name}: {model_calls_before_stop(build, CAP)}")

recursion_limit each implementation starts with
  create_agent: 9999
  create_deep_agent: 9999
  StateGraph (LangGraph default): 10007
model calls before the loop is stopped with recursion_limit=25 in the call config
  create_agent: 13
  StateGraph: 13
  create_deep_agent: 12