SERIES · Langchain Agent 应用开发
2026-03-15 · 30 min read · by GUMP
LangChain 新版 Agent 将 create_agent 与 LangGraph 图运行时结合，把工具、状态、middleware、流式与人类介入统一进可控、可观测、可扩展的生产架构。
Next in Series
LangChain Models 深度梳理：从基础调用到可配置模型 ->create_agent 到生产级 Agent：用 LangChain 构建可控、可观测、可扩展的智能体应用过去一年里，很多团队做 Agent 的方式都很像：
这种做法在 Demo 阶段没问题，但一旦进入业务系统，很快就会遇到四个现实问题：
LangChain 新版 Agent 体系真正重要的地方，不是“又包装了一层 Agent API”，而是把这些工程问题系统化了：create_agent 提供了生产可用的 Agent 实现，而底层运行时来自 LangGraph——也就是说，Agent 不再只是一个 while loop，而是一个基于图的状态机运行时。模型节点、工具节点、middleware、状态持久化、streaming、human-in-the-loop，都被纳入统一架构。(LangChain Docs)
这篇文章我会围绕一个真实方向展开：搭建一个“技术支持型 Agent”。它具备以下能力：
这才是一个能上生产的 Agent 雏形。
LangChain 官方文档对 Agent 的描述很准确：Agent 会把语言模型和工具结合起来，让系统能够围绕任务进行推理、决定调用哪个工具，并在循环中不断逼近目标，直到模型给出最终答案或达到迭代上限。更关键的是，create_agent 底层构建的是一个 graph-based runtime，节点包括 model node、tools node 以及 middleware 等。(LangChain Docs)
所以你写 Agent 时，最好不要把它理解成：
“给 LLM 绑了一堆 Python 函数”
而应该理解成：
“我在定义一个带状态的图执行系统，LLM 只是其中的一个推理节点。”
这个心智模型会直接影响你的代码组织方式。
先从一个最小版本开始。下面的代码重点不是“功能炫”，而是展示官方推荐入口 create_agent 的基本姿势。
# pip install -U langchain langchain-openai
import os
from langchain.agents import create_agent
from langchain.tools import tool
os.environ["OPENAI_API_KEY"] = "your-api-key"
@tool
def search_docs(query: str) -> str:
"""搜索内部知识库或文档系统。"""
# 真实项目中，这里应该对接企业搜索 / RAG / Elasticsearch / API
return f"[mock-search-result] 与 '{query}' 相关的技术文档如下：..."
@tool
def get_service_status(service_name: str) -> str:
"""查询某个服务的运行状态。"""
status_map = {
"payment": "healthy",
"order": "degraded",
"search": "healthy",
}
return status_map.get(service_name, "unknown")
agent = create_agent(
model="openai:gpt-5",
tools=[search_docs, get_service_status],
system_prompt=(
"你是一名企业级技术支持 Agent。"
"遇到事实性问题先考虑调用工具，不要编造服务状态。"
"回答时先给结论，再给依据。"
),
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "帮我看看 search 服务是否正常，如果不正常顺便查下相关文档。"
}
]
}
)
print(result)这段代码利用了官方文档中明确给出的 create_agent(model, tools=...) 入口，以及工具可以直接使用普通 Python 函数或 @tool 装饰器定义的机制。官方也说明了，如果工具列表为空，那么这个 agent 本质上就是单节点 LLM，不具备工具调用能力。(LangChain Docs)
第一，model="openai:gpt-5" 这种写法不是随便拼的。官方文档明确说，create_agent 支持直接传模型标识字符串，并自动推断 provider；同时也支持你直接传模型实例，以便控制 temperature、max_tokens、timeout 等参数。(LangChain Docs)
第二，Tool 的本质是带输入 schema 的可调用函数。也就是说，Tool 设计不是“能调用就行”，而是“要让模型容易选、容易填参、容易从返回值继续推理”。这决定了工具描述要足够清楚，参数要收敛，返回值要尽量结构化。(LangChain Docs)
很多 Agent 项目失败，不是模型不够强，而是 Tool 设计得太烂。
反例：
@tool
def operate_system(action: str, payload: dict) -> str:
"""执行任意系统操作"""
...这个工具的问题是：
from langchain.tools import tool
@tool
def search_incident(keyword: str) -> str:
"""搜索与故障、告警、事故复盘相关的记录。"""
return f"[mock-incident] 已找到与 {keyword} 相关的事故记录"
@tool
def get_k8s_deployment_status(namespace: str, deployment: str) -> str:
"""查询某个 Kubernetes deployment 的状态。"""
return (
f"namespace={namespace}, deployment={deployment}, "
f"replicas=3, available=2, status=degraded"
)
@tool
def create_jira_ticket(title: str, description: str, severity: str) -> str:
"""创建故障工单。severity 取值：low, medium, high, critical。"""
return f"[mock-jira] ticket created: OPS-1024"这才是 Agent 友好的工具层。
Tool 的 docstring 不是写给人看的，是写给模型看的。它要回答：
能枚举就不要开放文本；能拆字段就不要塞进一个 JSON blob。
返回自然语言可以，但更推荐可解析文本或结构化数据。
官方文档已经把这个点写得很明确：模型既可以静态配置，也可以在运行时动态选择。动态模型选择依赖 middleware 中的 @wrap_model_call，它可以根据当前 state 和上下文修改实际调用的模型。(LangChain Docs)
这非常关键，因为 Agent 不是每一步都值得用最贵的模型。
# pip install -U langchain langchain-openai
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain_openai import ChatOpenAI
basic_model = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
advanced_model = ChatOpenAI(model="gpt-4.1", temperature=0)
@wrap_model_call
def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:
"""
根据会话复杂度动态选择模型：
- 对话短、无复杂工具调用 -> 小模型
- 对话长、问题复杂 -> 大模型
"""
messages = request.state["messages"]
message_count = len(messages)
# 一个很朴素但有效的启发式
if message_count > 10:
return handler(request.override(model=advanced_model))
return handler(request.override(model=basic_model))
agent = create_agent(
model=basic_model, # 默认模型
tools=[search_incident, get_k8s_deployment_status, create_jira_ticket],
middleware=[dynamic_model_selection],
system_prompt="你是 SRE 支持 Agent。"
)这段模式几乎就是官方示例思路的工程化版本：根据 request.state["messages"] 判断当前上下文复杂度，然后 request.override(model=...)。官方也特别提醒了一点：如果你要把动态模型选择和结构化输出一起用，不要传入已经 bind_tools 过的预绑定模型。(LangChain Docs)
真实项目里，你可以把路由条件做得更像“成本控制器”：
例如：
def estimate_complexity(messages: list[dict]) -> int:
text = " ".join(
m.get("content", "")
for m in messages
if isinstance(m.get("content", ""), str)
).lower()
score = 0
keywords = [
"分析日志", "sql", "报错栈", "traceback", "根因", "复盘",
"multi-step", "workflow", "architecture"
]
for kw in keywords:
if kw.lower() in text:
score += 2
if len(messages) > 12:
score += 3
return score然后在 middleware 里按 score 分层选择模型。这样，你的 Agent 才开始具备真正的成本-质量平衡能力。
官方文档提到动态工具时，给了两个非常重要的判断：
这实际上揭示了一个关键原则：
Agent 的工具集合，不应该是“系统拥有的所有能力”，而应该是“当前轮允许暴露给模型的最小能力集”。
from dataclasses import dataclass
from typing import Callable
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain.tools import tool
@tool
def read_dashboard(metric_name: str) -> str:
"""读取监控指标。"""
return f"{metric_name}=42"
@tool
def restart_service(service_name: str) -> str:
"""重启服务。仅限有权限的用户。"""
return f"{service_name} restarted"
@tool
def delete_job(job_id: str) -> str:
"""删除任务。高风险操作。"""
return f"job {job_id} deleted"
@dataclass
class Context:
user_role: str # viewer / operator / admin
@wrap_model_call
def context_based_tools(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
if request.runtime is None or request.runtime.context is None:
user_role = "viewer"
else:
user_role = request.runtime.context.user_role
if user_role == "admin":
# 全量工具
return handler(request)
if user_role == "operator":
tools = [t for t in request.tools if t.name != "delete_job"]
return handler(request.override(tools=tools))
# viewer 默认只读
tools = [t for t in request.tools if t.name.startswith("read_")]
return handler(request.override(tools=tools))
agent = create_agent(
model="gpt-4.1",
tools=[read_dashboard, restart_service, delete_job],
middleware=[context_based_tools],
context_schema=Context,
system_prompt="你是企业运维助手，必须遵守权限控制。"
)
result = agent.invoke(
{
"messages": [
{"role": "user", "content": "帮我删除 job-123"}
]
},
context=Context(user_role="viewer")
)
print(result)这类模式和官方文档里基于 Runtime Context / State / Store 的动态工具过滤是一致的。它的本质是把“权限控制”前移到了模型可见能力层，而不是等模型调用了危险工具再兜底。(LangChain Docs)
如果你的 Agent 工具数超过 8～12 个，而且业务差异明显，建议你不要直接全暴露。你应该至少做以下分层：
否则模型会越来越像“面对一堵工具菜单墙在犹豫”。
官方结构化输出文档有一句非常重要的话：create_agent 会自动处理结构化输出，结构化结果会被捕获、校验，并放在最终 state 的 structured_response 字段里。response_format 支持 ToolStrategy、ProviderStrategy 或者直接传 schema 类型，让 LangChain 自动选择最佳策略。(LangChain Docs)
这意味着什么？
意味着你终于不需要在前端或后端用正则从自然语言里抠字段了。
from pydantic import BaseModel, Field
from typing import Literal
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class IncidentReport(BaseModel):
summary: str = Field(description="故障摘要")
severity: Literal["low", "medium", "high", "critical"] = Field(description="故障等级")
suspected_root_cause: str = Field(description="疑似根因")
next_action: list[str] = Field(description="下一步行动项")
need_human_escalation: bool = Field(description="是否需要人工升级处理")
agent = create_agent(
model="gpt-5",
tools=[search_incident, get_k8s_deployment_status],
response_format=ToolStrategy(IncidentReport),
system_prompt=(
"你是故障分析 Agent。"
"先基于工具结果分析，再输出结构化结论。"
),
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "search 服务最近 5 分钟有大量报错，请帮我分析，并给出建议。"
}
]
}
)
print(result["structured_response"])因为这里不是 prompt 级 JSON 约束，而是运行时级 schema 约束。官方文档明确指出：
ProviderStrategy 会优先利用 provider 的原生 structured output 能力ToolStrategy 会通过 tool calling 方式达成相同目标这对生产非常重要。你的前端可以稳定消费 IncidentReport，你的后端可以直接存库，你的自动化流程可以依据 need_human_escalation 做分支判断。
比如审批、告警归类、工单分发、用户意图抽取，这些任务最适合结构化输出。
from pydantic import BaseModel, Field
from typing import Literal
class TicketRoutingDecision(BaseModel):
team: Literal["sre", "backend", "frontend", "data", "security"]
priority: Literal["p1", "p2", "p3", "p4"]
reason: str你让 Agent 返回这个对象，后面流程引擎就可以直接消费，而不是拿自然语言再判断一次。
官方文档把短期记忆讲得很清楚：短期记忆用于让应用在单个 thread / conversation 中记住之前交互；在 LangChain Agent 里，它是 agent state 的一部分，通过 checkpointer 持久化，step 开始时读入，step 完成时更新。(LangChain Docs)
这和很多人理解的“把历史消息拼回去”有本质区别：
短期记忆不是字符串拼接，而是线程级状态管理。
# pip install -U langgraph
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent = create_agent(
model="gpt-5",
tools=[search_docs],
checkpointer=checkpointer,
system_prompt="你是技术支持 Agent。"
)
# 第 1 轮
agent.invoke(
{
"messages": [
{"role": "user", "content": "我叫 Bob，我们在排查 search 服务故障。"}
]
},
{"configurable": {"thread_id": "session-001"}}
)
# 第 2 轮：同一 thread
result = agent.invoke(
{
"messages": [
{"role": "user", "content": "继续刚才的话题，帮我总结一下当前上下文。"}
]
},
{"configurable": {"thread_id": "session-001"}}
)
print(result)这正是官方短期记忆示例的核心模式：checkpointer + thread_id。(LangChain Docs)
官方建议生产环境使用数据库支撑的 checkpointer，例如 Postgres。文档给出了 langgraph-checkpoint-postgres 的做法。(LangChain Docs)
# pip install langgraph-checkpoint-postgres
from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup()
agent = create_agent(
model="gpt-5",
tools=[search_docs],
checkpointer=checkpointer,
system_prompt="你是支持 Agent。"
)
result = agent.invoke(
{
"messages": [
{"role": "user", "content": "记录一下：当前客户是 ACME，故障级别倾向 P1。"}
]
},
{"configurable": {"thread_id": "case-20260315-001"}}
)官方文档直接指出，长上下文会让模型“分心”，即使 context window 足够大，也会带来更慢、更贵、效果更差的问题。因此常见策略包括：
这意味着你在工程上要接受一个现实：
记忆不是越多越好，而是越“与当前任务相关”越好。
官方文档说明，默认 Agent 使用 AgentState 管理短期记忆，核心字段是 messages；但你完全可以扩展 state_schema，把更多业务字段纳入状态。(LangChain Docs)
这点非常强，因为很多 Agent 问题本质上是“业务上下文散落在外面”。
from typing import Optional
from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver
class SupportAgentState(AgentState):
user_id: str
tenant_id: str
active_ticket_id: Optional[str] = None
incident_stage: Optional[str] = None
agent = create_agent(
model="gpt-5",
tools=[search_incident, create_jira_ticket],
state_schema=SupportAgentState,
checkpointer=InMemorySaver(),
system_prompt=(
"你是企业支持 Agent。"
"需要结合当前工单阶段和租户上下文来回答问题。"
)
)因为一旦你把状态显式化，很多高级能力就顺理成章了：
官方文档也明确提到，工具可以通过 runtime 读取短期记忆，还可以直接返回 state updates 修改 agent state。(LangChain Docs)
from langchain.agents import AgentState
from langchain.tools import tool, ToolRuntime
class CustomState(AgentState):
user_id: str
active_ticket_id: str | None = None
@tool
def read_current_ticket(runtime: ToolRuntime[CustomState]) -> str:
"""读取当前会话正在处理的工单 ID。"""
ticket_id = runtime.state.get("active_ticket_id")
if not ticket_id:
return "当前没有激活的工单。"
return f"当前工单是 {ticket_id}"根据官方文档，runtime 作为 ToolRuntime 参数存在，但不会暴露给模型，因此它可以安全地让工具访问内部 state。(LangChain Docs)
@tool
def set_active_ticket(ticket_id: str) -> dict:
"""设置当前会话的激活工单。"""
return {
"active_ticket_id": ticket_id,
"messages": [
{
"role": "tool",
"content": f"已将当前激活工单设置为 {ticket_id}"
}
]
}这一招非常适合做：
如果你的 Agent 要执行“先检索，再分析，再创建工单，再回写结果”这样的链路，状态读写能力几乎是必需品。
官方 middleware 文档说得很直白：middleware 可以让你更紧密地控制 agent 内部发生的事情，比如日志、分析、调试、prompt 转换、工具选择、输出格式、重试、fallback、early termination、rate limit、guardrails、PII detection。(LangChain Docs)
这句话基本可以翻译成：
Prompt 负责“告诉模型怎么想”，Middleware 负责“控制系统怎么跑”。
你可以把 middleware 想成 Agent 生命周期的拦截器。官方 custom middleware 文档指出，middleware 支持 node-style hooks 和 wrap-style hooks，可在特定执行点或包裹模型/工具调用时生效。(LangChain Docs)
比如：
import time
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
@wrap_model_call
def metrics_middleware(request: ModelRequest, handler) -> ModelResponse:
start = time.time()
response = handler(request)
cost = time.time() - start
# 这里可接入 Prometheus / OpenTelemetry / 自定义日志系统
print(
"[metrics]",
{
"message_count": len(request.state["messages"]),
"tool_count": len(request.tools),
"latency_sec": round(cost, 3),
}
)
return response这类 middleware 不花哨，但非常有价值。因为线上 Agent 迟早要回答这些问题：
官方 Guardrails 文档把防护分成 deterministic guardrails 和 model-based guardrails，同时支持 before-agent 和 after-agent 等机制，还提供了 PII middleware、human-in-the-loop 等能力。before-agent guardrail 可以在 invocation 一开始做认证、限流、拦截不当请求等工作。(LangChain Docs)
阻止明显违规或越权请求。
高风险工具必须审计、确认或改为 HITL。
防止泄露敏感信息、内部字段、PII 等。
from typing import Any
from langchain.agents.middleware import AgentMiddleware, hook_config
from langgraph.runtime import Runtime
class ContentFilterMiddleware(AgentMiddleware):
def __init__(self, banned_keywords: list[str]):
self.banned_keywords = banned_keywords
@hook_config(can_jump_to=["end"])
def before_agent(self, state: dict[str, Any], runtime: Runtime) -> dict[str, Any] | None:
messages = state.get("messages", [])
last_user_text = ""
for msg in reversed(messages):
if msg.get("role") == "user":
last_user_text = msg.get("content", "")
break
lowered = last_user_text.lower()
if any(kw in lowered for kw in self.banned_keywords):
return {
"messages": [
{
"role": "assistant",
"content": "请求被安全策略拦截，请联系管理员。"
}
]
}
return None然后接到 agent 上：
agent = create_agent(
model="gpt-4.1",
tools=[read_dashboard, restart_service],
middleware=[
ContentFilterMiddleware(banned_keywords=["drop database", "delete all", "hack"])
],
system_prompt="你是运维支持 Agent。"
)这类模式与官方 before-agent guardrail 用法一致：在真正进入 Agent 推理之前先做验证。(LangChain Docs)
restart_service、delete_job 这类工具启用人工审批如果你线上还有合规要求，那 guardrails 应该被视为系统主路径，而不是补丁。
LangGraph 官方关于 workflows and agents 的说明很值得引用：workflow 是预定义代码路径、按既定顺序执行；agent 是动态的，会自己定义过程与工具使用方式。(LangChain Docs)
这句话看似简单，但实践里非常重要。
例如：
例如：
也就是说：
例如故障处理系统：
这比“全链路都交给 Agent 随便发挥”稳得多。
下面给一个相对完整、可落地的示例，把前面的点串起来：
create_agent# pip install -U langchain langchain-openai langgraph pydantic
from dataclasses import dataclass
from typing import Callable, Literal, Optional
from pydantic import BaseModel, Field
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import (
wrap_model_call,
ModelRequest,
ModelResponse,
AgentMiddleware,
hook_config,
)
from langchain.agents.structured_output import ToolStrategy
from langchain.tools import tool, ToolRuntime
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
# ----------------------------
# 1) 模型
# ----------------------------
basic_model = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
advanced_model = ChatOpenAI(model="gpt-4.1", temperature=0)
# ----------------------------
# 2) 状态
# ----------------------------
class SupportState(AgentState):
user_id: str
tenant_id: str
active_ticket_id: Optional[str] = None
incident_stage: Optional[str] = None
@dataclass
class RequestContext:
user_role: str # viewer / operator / admin
# ----------------------------
# 3) 工具
# ----------------------------
@tool
def search_knowledge_base(query: str) -> str:
"""搜索知识库、FAQ、事故复盘、架构文档。"""
return f"[KB] 找到与 '{query}' 相关文档 3 篇"
@tool
def get_service_health(service_name: str) -> str:
"""查询服务健康状态。"""
mock = {
"search": "status=degraded,error_rate=12%,latency_p95=2.4s",
"payment": "status=healthy,error_rate=0.1%,latency_p95=180ms",
"order": "status=healthy,error_rate=0.2%,latency_p95=230ms",
}
return mock.get(service_name, "status=unknown")
@tool
def create_incident_ticket(title: str, description: str, severity: str) -> dict:
"""创建故障工单，severity 取值: low/medium/high/critical。"""
ticket_id = "INC-20260315-1024"
return {
"active_ticket_id": ticket_id,
"incident_stage": "created",
"messages": [
{"role": "tool", "content": f"工单已创建，ticket_id={ticket_id}"}
]
}
@tool
def read_active_ticket(runtime: ToolRuntime[SupportState]) -> str:
"""读取当前线程中的激活工单。"""
ticket_id = runtime.state.get("active_ticket_id")
if not ticket_id:
return "当前没有激活工单。"
return f"当前工单: {ticket_id}"
@tool
def restart_service(service_name: str) -> str:
"""重启服务。高风险操作，仅 operator/admin 可用。"""
return f"{service_name} restarted"
# ----------------------------
# 4) 结构化输出
# ----------------------------
class SupportDecision(BaseModel):
summary: str = Field(description="当前问题总结")
severity: Literal["low", "medium", "high", "critical"] = Field(description="严重等级")
should_create_ticket: bool = Field(description="是否应创建故障工单")
should_restart_service: bool = Field(description="是否建议重启服务")
next_actions: list[str] = Field(description="后续行动项")
evidence: list[str] = Field(description="结论依据")
# ----------------------------
# 5) Middleware：模型路由
# ----------------------------
@wrap_model_call
def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:
messages = request.state["messages"]
text = " ".join(
m.get("content", "")
for m in messages
if isinstance(m.get("content", ""), str)
).lower()
complexity = 0
if len(messages) > 10:
complexity += 2
if any(kw in text for kw in ["根因", "traceback", "日志", "sql", "架构"]):
complexity += 2
if "critical" in text or "p1" in text:
complexity += 2
model = advanced_model if complexity >= 3 else basic_model
return handler(request.override(model=model))
# ----------------------------
# 6) Middleware：权限裁剪工具
# ----------------------------
@wrap_model_call
def filter_tools_by_role(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
if request.runtime is None or request.runtime.context is None:
role = "viewer"
else:
role = request.runtime.context.user_role
if role == "admin":
return handler(request)
if role == "operator":
tools = [t for t in request.tools if t.name != "create_incident_ticket"]
return handler(request.override(tools=tools))
# viewer: 只读
readonly = {"search_knowledge_base", "get_service_health", "read_active_ticket"}
tools = [t for t in request.tools if t.name in readonly]
return handler(request.override(tools=tools))
# ----------------------------
# 7) Guardrail：危险输入拦截
# ----------------------------
class DangerousInputMiddleware(AgentMiddleware):
def __init__(self, banned_keywords: list[str]):
self.banned_keywords = banned_keywords
@hook_config(can_jump_to=["end"])
def before_agent(self, state: dict, runtime: Runtime) -> dict | None:
messages = state.get("messages", [])
last_user = ""
for msg in reversed(messages):
if msg.get("role") == "user":
last_user = msg.get("content", "")
break
lowered = last_user.lower()
if any(kw in lowered for kw in self.banned_keywords):
return {
"messages": [
{
"role": "assistant",
"content": "请求触发安全策略，已拒绝执行。"
}
]
}
return None
# ----------------------------
# 8) 组装 Agent
# ----------------------------
agent = create_agent(
model=basic_model,
tools=[
search_knowledge_base,
get_service_health,
create_incident_ticket,
read_active_ticket,
restart_service,
],
middleware=[
DangerousInputMiddleware(["delete all", "drop database", "shutdown cluster"]),
dynamic_model_selection,
filter_tools_by_role,
],
state_schema=SupportState,
context_schema=RequestContext,
response_format=ToolStrategy(SupportDecision),
checkpointer=InMemorySaver(),
system_prompt=(
"你是一个企业级技术支持 Agent。\n"
"规则：\n"
"1. 事实性判断优先使用工具，不要编造。\n"
"2. 高风险操作只给建议，不擅自执行。\n"
"3. 输出必须结合证据，给出 next_actions。\n"
"4. 如果信息不足，明确指出缺口。"
),
)
# ----------------------------
# 9) 调用
# ----------------------------
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "search 服务现在报错率很高，请帮我分析要不要创建故障工单，顺便给出下一步建议。"
}
]
},
context=RequestContext(user_role="viewer"),
config={"configurable": {"thread_id": "support-thread-001"}},
)
print("structured_response =", result["structured_response"])因为它不是单纯“做出来了”，而是考虑了：
这套组合拳，和 LangChain 官方当前 Agent 设计理念是高度一致的。(LangChain Docs)
LangGraph 官方强调了 durable execution、memory、human-in-the-loop、debugging with LangSmith 等底层能力。换句话说，生产 Agent 的关键不只是“推理成功”，而是可恢复、可观察、可干预。(LangChain Docs)
Tool 是给模型消费的，不是给工程师炫技的。能拆就拆，能限制就限制。
只要后续系统要消费结果，就别让模型输出自由文本。
长上下文不是资产，很多时候是噪声。该 trim、summarize 就做。官方也把 summarization middleware 作为预置能力之一。(LangChain Docs)
尤其是删数据、改配置、发通知、扣费等动作。Agent 可以建议，但不一定应该直接执行。
真正成熟的系统通常是：
如果你以前熟悉的是旧时代 LangChain 里各种 agent executor、memory patch、callback patch 的玩法，那么当前这套体系最值得注意的是：
create_agent 成为官方标准入口，LangChain v1 明确推荐它，且相比旧的 create_react_agent 更易用，也更容易通过 middleware 做深度定制。(LangChain Docs)如果要用一句话总结 LangChain 这套新 Agent 体系，我会这样说：
它的核心价值，不是让 LLM 获得“工具使用能力”，而是让 Agent 应用第一次有了接近后端系统的工程骨架。
这个骨架包括：
所以，今天再谈 Agent 开发，已经不能只谈 Prompt 和 Tool 了。真正决定一个 Agent 能不能上线的，是它是否具备可控性、可维护性、可观测性和可扩展性。
而 LangChain 当前这套 create_agent + LangGraph runtime + middleware + state 的设计，已经把这些关键问题摆到了台面上。(LangChain Docs)