实用提示：2026年最佳智能体框架：LangGraph与OpenAI Agents对比
《实用笔记》操作指南：2026年最佳智能体框架——LangGraph与OpenAI Agents：适用于采用该架构的团队的合约、校验机制及可直接插入的代码模块。
本指南将逐步展示如何从原始材料构建出可运行的系统，主题为“2026年最佳智能体框架：LangGraph vs OpenAI Agents SDK vs Claude Agent SDK”。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库的代码，无需猜测其用途。 在修改代码之前，应先明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需推测隐藏状态。 除了功能结果外，还需记录执行时间以及token或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外费用。
三种基本组件
在处理《三种基本元素》时，首先写下契约：所需的输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放，以便操作员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型。
2025年应摒弃的观念
在制定“2025年需摒弃的信念”计划时，首先要写明相关约定：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改始终符合初始规划。
# Install: pip install "openai-agents[litellm]"
# Env: export GEMINI_API_KEY=...
import os
from agents import Agent, Runner, function_tool
from agents.extensions.models.litellm_model import LitellmModel
@function_tool
def current_time_utc() -> str:
"""Return the current UTC time as an ISO-8601 string."""
from datetime import datetime, timezone
return datetime.now(timezone.utc).isoformat(timespec="seconds")
# OpenAI Agents SDK using Gemini via LiteLLM. No OpenAI key required.
gemini_model = LitellmModel(
model="gemini/gemini-2.5-pro",
api_key=os.environ["GEMINI_API_KEY"],
)
agent = Agent(
name="time-agent",
instructions="Answer time questions using the current_time_utc tool.",
model=gemini_model,
tools=[current_time_utc],
)
result = Runner.run_sync(agent, "What is the current UTC time?")
print(result.final_output)
# -> "The current UTC time is 2026-07-06T14:32:11+00:00."
介绍 Meridian
在学习 Introducing Meridian 时，首先需明确接口规范：所需的输入参数、成功标识以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 建议采用小型、可测试的单元而非庞大的脚本。当某个步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大语言模型接口。 在学习 Introducing Meridian 时，首先需明确接口规范：所需的输入参数、成功标识以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及token或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。
工作负载1：语音与实时流处理
工作负载1：将语音与实时流处理视为可度量的对象时，其性能最佳。在扩大范围之前，需记录一份理想的转录文本、一个故障案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放，以便操作人员无需查看整个架构即可进行审计。 保持架构状态的扁平化与类型化。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，且在中断后会导致无法继续处理。
# Install: pip install openai-agents
# Env: export OPENAI_API_KEY=...
import asyncio
from agents import function_tool
from agents.realtime import RealtimeAgent, RealtimeRunner
@function_tool
def lookup_billing_balance(account_id: str) -> str:
"""Return the current outstanding balance for an account."""
# In production, this hits the billing service. Here it is a stub.
return "42.17 USD outstanding as of 2026-07-06."
voice_agent = RealtimeAgent(
name="meridian-billing-voice",
instructions=(
"You are Meridian's billing voice assistant. Answer politely, briefly. "
"Confirm the account_id before disclosing any balance."
),
tools=[lookup_billing_balance],
)
async def main():
runner = RealtimeRunner(
starting_agent=voice_agent,
config={"model_settings": {"model_name": "gpt-realtime-2.1"}},
)
# session handles the audio stream and tool calls
session = await runner.run()
async with session:
# Wire the audio input source here via sounddevice or pyaudio
async for event in session:
if event.type == "history_updated":
# The item contains the finalized transcript once the turn ends
print(f"History updated with item: {event.item}")
elif event.type == "error":
print(f"Error: {event.error}")
break
asyncio.run(main())
工作负载2：基于HITL的持久多智能体编排
工作负载2：基于HITL的持久多智能体协调机制，若将其视为可度量的模型则效果最佳。在扩大范围之前，需记录一份理想的执行流程、一个故障案例以及回滚说明。 同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。 保持图结构的状态简洁且类型明确。嵌套的数据块会掩盖某个节点编写了哪个字段的信息，且在中断后会导致流程无法继续。
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain_google_genai import ChatGoogleGenerativeAI
# LangGraph is provider-agnostic. Here it uses Gemini.
llm = ChatGoogleGenerativeAI(model="gemini-2.5-pro", temperature=0)
class RefundState(TypedDict):
order_id: str
amount_usd: float
customer_reason: str
fraud_risk: Literal["low", "medium", "high"] | None
finance_decision: Literal["approved", "denied"] | None
human_review_needed: bool
def fraud_check(state: RefundState) -> RefundState:
"""Run the LLM-backed fraud check against the customer's stated reason."""
prompt = (
f"Assess fraud risk for refund of ${state['amount_usd']:.2f}. "
f"Customer reason: {state['customer_reason']!r}. "
"Respond with one word: low, medium, or high."
)
verdict = llm.invoke(prompt).content.strip().lower()
if verdict not in {"low", "medium", "high"}:
verdict = "high" # fail-closed on ambiguous LLM output
return {**state, "fraud_risk": verdict}
def finance_approval(state: RefundState) -> RefundState:
"""Above $500 or medium risk, pause for a human. Otherwise auto-approve."""
needs_human = state["amount_usd"] > 500 or state["fraud_risk"] in {"medium", "high"}
if needs_human:
# Pause the graph. On resume, interrupt returns the human's decision.
human_decision = interrupt({
"order_id": state["order_id"],
"amount_usd": state["amount_usd"],
"fraud_risk": state["fraud_risk"],
"prompt": "Approve (yes/no)?",
})
return {**state, "human_review_needed": True, "finance_decision": human_decision}
return {**state, "human_review_needed": False, "finance_decision": "approved"}
# Build the graph
graph = StateGraph(RefundState)
graph.add_node("fraud_check", fraud_check)
graph.add_node("finance_approval", finance_approval)
graph.add_edge(START, "fraud_check")
graph.add_conditional_edges(
"fraud_check",
lambda s: "finance_approval" if s["fraud_risk"] != "high" else END,
)
graph.add_edge("finance_approval", END)
# Checkpointer. For production, swap MemorySaver for PostgresSaver.
compiled = graph.compile(checkpointer=MemorySaver())
# Run it. Interrupt fires on the $850 refund and the graph pauses.
config = {"configurable": {"thread_id": "order-4291"}}
result = compiled.invoke(
{
"order_id": "4291",
"amount_usd": 850.00,
"customer_reason": "arrived damaged, no photo",
"fraud_risk": None,
"finance_decision": None,
"human_review_needed": False,
},
config=config,
)
# Later, a human reviewer says yes. Resume with Command.
final = compiled.invoke(Command(resume="approved"), config=config)
工作负载3：与编码相关且以文件和shell操作为中心
工作负载3：与编码相关以及以文件和Shell操作为中心的任务，若将其视为可度量的对象来处理会更为高效。在扩大范围之前，需记录一份理想的操作日志、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某一步骤出现故障时，故障应能指向单一责任点，而非复杂的流程链。 保持图表状态简洁且具有类型定义。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会在进程中断后导致无法继续运行。 工作负载3：与编码相关以及以文件和Shell操作为中心的任务，若将其视为可度量的对象来处理会更为高效。在扩大范围之前，需记录一份理想的操作日志、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
# Install: pip install claude-agent-sdk
# Env: export ANTHROPIC_API_KEY=...
import anyio
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
AgentDefinition,
HookMatcher,
)
# PreToolUse hook: block Bash calls that look like rm -rf
async def block_dangerous_bash(input_data, tool_use_id, context):
if input_data.get("tool_name") == "Bash":
cmd = input_data.get("tool_input", {}).get("command", "")
if "rm -rf" in cmd or "rm -rf" in cmd:
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": "rm -rf blocked by policy",
}
}
return {}
# Subagent: runs in isolated context to lint one file
lint_agent = AgentDefinition(
description="Run linters on a single file and return a concise report.",
prompt=(
"You are the lint subagent. Given a file path, run the project's linter "
"on it and return a one-paragraph summary of failures. Do not fix anything."
),
tools=["Bash", "Read"], # Note: tools is deprecated in favor of skills in recent SDKs
)
options = ClaudeAgentOptions(
system_prompt=(
"You are Meridian's code migration agent. Walk the target directory, "
"apply the migration, run tests, and open a PR. Prefer small commits."
),
allowed_tools=["Bash", "Read", "Write", "Edit", "Glob", "Grep"],
hooks={"PreToolUse": [HookMatcher(hooks=[block_dangerous_bash])]},
agents={"lint": lint_agent},
# resume="mig-run-2026-07-06-01", # uncomment to resume a prior session
)
async def main():
async with ClaudeSDKClient(options=options) as client:
await client.query(
"Migrate services/payments/ from Java 17 to Java 21. "
"For every file you touch, delegate to the `lint` subagent afterward. "
"Do NOT commit or open PRs yet. Stop after changes are on disk."
)
async for message in client.receive_response():
print(message)
anyio.run(main)
工作负载4：基于MCP的重型工具编排
对于工作负载4：基于MCP的重型工具编排，在修改代码之前需明确输入参数、各步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测其中的隐藏状态。配置信息应置于应用程序代码之外，环境文件、密钥存储以及功能标志都应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审核。认证应在网关层完成，重新授权则在数据平面执行——仅凭承载令牌并不足以界定租户边界。
决策矩阵
对于决策矩阵，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。
关于CrewAI
在 On CrewAI 中，修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相较于庞大的脚本，更应优先选择小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任方，而非复杂的流程链。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。 在 On CrewAI 中，修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境过渡到共享环境时出现意外账单。
真正的选择
在处理《真实选择》时，首先写下相关契约：所需的输入参数、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
运营检查清单
若将运营检查清单视为可量化的标准，其效果会更好。在扩大范围之前，先记录一份最佳处理案例、一个失败场景以及回滚说明。 应将此阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功标准，并杜绝无声的半完成状态。
保持图状态扁平且具有类型约束。嵌套的二进制数据会隐藏是哪个节点修改了哪个字段，还会在中断后导致程序无法继续运行。
只要预算允许，就在持续集成过程中使用测试用例而非真实的付费 API 来执行关键路径的冒烟测试。
在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。
保持图状态扁平且具有类型约束。嵌套的二进制数据会隐藏是哪个节点修改了哪个字段，还会在中断后导致程序无法继续运行。
在升级技术栈之前，先冻结现有版本，为关键路径生成标准化的测试记录，并确认回滚步骤。共享环境需要设置速率限制、进行租户身份验证，同时明确负责密钥轮换的人员。与其追求花哨的一次性演示，不如注重扎实的可靠性。
关于2c64e0b378d9的批量处理说明：不要将提供者密钥放入代码仓库，为每个会话设置令牌使用上限，并将转录内容存储在评估测试用例的旁边，以便后续更换模型时仍能保持可比性。
强化措施0作为可量化指标来使用效果最佳。在扩大范围之前，先记录一份理想的转录样本、一个失败案例以及回滚说明。优先选择小型且可测试的单元，而非庞大的脚本；当某个步骤失败时，故障应能指向单一责任点，而非复杂的流程链。
强化细节0/821：针对此措施需统计耗时、错误类型以及令牌使用量，然后依据固定的问题集而非主观经验来决定是否保留该变更。
针对强化措施1，在修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能测试结果旁记录执行时间以及代币或查询成本。提前了解这些成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。
强化措施细节1/821：为该措施测量实际执行时间、错误类型以及代币消耗情况，然后依据固定的评估标准而非主观判断来决定是否保留该更改。