从 create_agent 到生产级 Agent：用 LangChain 构建可控、可观测、可扩展的智能体应用

一、为什么今天做 Agent，不能再停留在“会调用几个工具”？

过去一年里，很多团队做 Agent 的方式都很像：

  1. 给模型塞一个 system prompt
  2. 绑定几个工具
  3. 跑一个 ReAct loop
  4. 祈祷它别失控

这种做法在 Demo 阶段没问题，但一旦进入业务系统，很快就会遇到四个现实问题：

LangChain 新版 Agent 体系真正重要的地方，不是“又包装了一层 Agent API”，而是把这些工程问题系统化了：create_agent 提供了生产可用的 Agent 实现，而底层运行时来自 LangGraph——也就是说，Agent 不再只是一个 while loop，而是一个基于图的状态机运行时。模型节点、工具节点、middleware、状态持久化、streaming、human-in-the-loop，都被纳入统一架构。(LangChain Docs)

这篇文章我会围绕一个真实方向展开：搭建一个“技术支持型 Agent”。它具备以下能力：

这才是一个能上生产的 Agent 雏形。

二、先建立一个正确的心智模型：LangChain Agent = LangGraph Runtime + Model + Tools + State

LangChain 官方文档对 Agent 的描述很准确：Agent 会把语言模型和工具结合起来，让系统能够围绕任务进行推理、决定调用哪个工具，并在循环中不断逼近目标，直到模型给出最终答案或达到迭代上限。更关键的是，create_agent 底层构建的是一个 graph-based runtime，节点包括 model node、tools node 以及 middleware 等。(LangChain Docs)

所以你写 Agent 时，最好不要把它理解成：

“给 LLM 绑了一堆 Python 函数”

而应该理解成：

“我在定义一个带状态的图执行系统，LLM 只是其中的一个推理节点。”

这个心智模型会直接影响你的代码组织方式。

三、最小可运行版本：一个真正可执行的 LangChain Agent

先从一个最小版本开始。下面的代码重点不是“功能炫”，而是展示官方推荐入口 create_agent 的基本姿势。

python
# 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 设计对

很多 Agent 项目失败，不是模型不够强，而是 Tool 设计得太烂。

常见错误一：把一个大而全的 SDK 暴露给模型

反例：

python
@tool
def operate_system(action: str, payload: dict) -> str:
    """执行任意系统操作"""
    ...

这个工具的问题是：

更合理的做法：面向任务拆工具

python
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 设计的三个工程原则

1）描述要解决“何时调用”的问题

Tool 的 docstring 不是写给人看的，是写给模型看的。它要回答：

2）参数要解决“怎么调用”的问题

能枚举就不要开放文本；能拆字段就不要塞进一个 JSON blob。

3）返回值要解决“怎么继续推理”的问题

返回自然语言可以，但更推荐可解析文本或结构化数据。

五、静态模型够用吗？真正的线上 Agent 往往需要动态模型路由

官方文档已经把这个点写得很明确：模型既可以静态配置，也可以在运行时动态选择。动态模型选择依赖 middleware 中的 @wrap_model_call，它可以根据当前 state 和上下文修改实际调用的模型。(LangChain Docs)

这非常关键，因为 Agent 不是每一步都值得用最贵的模型。

一个实用策略：简单轮次用小模型，复杂轮次升配

python
# 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)

我建议再进一步：别只看消息数

真实项目里，你可以把路由条件做得更像“成本控制器”：

例如：

python
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 的工具集合，不应该是“系统拥有的所有能力”，而应该是“当前轮允许暴露给模型的最小能力集”。

场景：游客、普通用户、管理员看到的工具不同

python
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 个，而且业务差异明显，建议你不要直接全暴露。你应该至少做以下分层：

否则模型会越来越像“面对一堵工具菜单墙在犹豫”。

七、结构化输出不是“锦上添花”，而是 Agent 进入应用层的门票

官方结构化输出文档有一句非常重要的话：create_agent 会自动处理结构化输出，结构化结果会被捕获、校验，并放在最终 state 的 structured_response 字段里。response_format 支持 ToolStrategy、ProviderStrategy 或者直接传 schema 类型，让 LangChain 自动选择最佳策略。(LangChain Docs)

这意味着什么？

意味着你终于不需要在前端或后端用正则从自然语言里抠字段了。

例子：让 Agent 输出故障分析卡片

python
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"])

为什么这比“让模型输出 JSON”强很多？

因为这里不是 prompt 级 JSON 约束，而是运行时级 schema 约束。官方文档明确指出：

这对生产非常重要。你的前端可以稳定消费 IncidentReport，你的后端可以直接存库，你的自动化流程可以依据 need_human_escalation 做分支判断。

一个更实战的技巧：把 Agent 当“结构化决策器”

比如审批、告警归类、工单分发、用户意图抽取，这些任务最适合结构化输出。

python
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)

这和很多人理解的“把历史消息拼回去”有本质区别：

短期记忆不是字符串拼接，而是线程级状态管理。

最简单的做法：加一个内存 checkpointer

python
# 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)

python
# 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 足够大，也会带来更慢、更贵、效果更差的问题。因此常见策略包括：

这意味着你在工程上要接受一个现实：

记忆不是越多越好，而是越“与当前任务相关”越好。

九、把“业务状态”放进 AgentState，而不只是 messages

官方文档说明，默认 Agent 使用 AgentState 管理短期记忆，核心字段是 messages；但你完全可以扩展 state_schema，把更多业务字段纳入状态。(LangChain Docs)

这点非常强，因为很多 Agent 问题本质上是“业务上下文散落在外面”。

例子：给 Agent 加上用户 ID、租户、当前工单上下文

python
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)

十、让 Tool 读取和写回状态：这是多步 Agent 的关键能力

1）Tool 读取状态

python
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)

2）Tool 写回状态

python
@tool
def set_active_ticket(ticket_id: str) -> dict:
    """设置当前会话的激活工单。"""
    return {
        "active_ticket_id": ticket_id,
        "messages": [
            {
                "role": "tool",
                "content": f"已将当前激活工单设置为 {ticket_id}"
            }
        ]
    }

这一招非常适合做：

如果你的 Agent 要执行“先检索，再分析，再创建工单，再回写结果”这样的链路，状态读写能力几乎是必需品。

十一、Middleware 才是生产 Agent 的核心扩展点

官方 middleware 文档说得很直白：middleware 可以让你更紧密地控制 agent 内部发生的事情，比如日志、分析、调试、prompt 转换、工具选择、输出格式、重试、fallback、early termination、rate limit、guardrails、PII detection。(LangChain Docs)

这句话基本可以翻译成：

Prompt 负责“告诉模型怎么想”，Middleware 负责“控制系统怎么跑”。

你应该把哪些逻辑放进 middleware？

适合放进去的

不适合放进去的

一个实用的 before/after 模式

你可以把 middleware 想成 Agent 生命周期的拦截器。官方 custom middleware 文档指出，middleware 支持 node-style hooks 和 wrap-style hooks，可在特定执行点或包裹模型/工具调用时生效。(LangChain Docs)

比如：

python
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：不是“安全部门的需求”，而是 Agent 系统的基本卫士

官方 Guardrails 文档把防护分成 deterministic guardrails 和 model-based guardrails，同时支持 before-agent 和 after-agent 等机制，还提供了 PII middleware、human-in-the-loop 等能力。before-agent guardrail 可以在 invocation 一开始做认证、限流、拦截不当请求等工作。(LangChain Docs)

我建议你至少做三层 guardrails

第 1 层：输入拦截

阻止明显违规或越权请求。

第 2 层：工具调用拦截

高风险工具必须审计、确认或改为 HITL。

第 3 层：输出净化

防止泄露敏感信息、内部字段、PII 等。

一个简单的关键词输入拦截

python
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 上：

python
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)

更高阶的做法

如果你线上还有合规要求，那 guardrails 应该被视为系统主路径，而不是补丁。

十三、Agent 和 Workflow 的边界：不要什么都做成 Agent

LangGraph 官方关于 workflows and agents 的说明很值得引用：workflow 是预定义代码路径、按既定顺序执行；agent 是动态的，会自己定义过程与工具使用方式。(LangChain Docs)

这句话看似简单，但实践里非常重要。

什么时候该用 Workflow？

例如：

什么时候该用 Agent？

例如：

我更推荐的生产模式：Workflow 外壳 + Agent 内核

也就是说：

例如故障处理系统：

  1. Workflow 接收告警
  2. Agent 负责日志分析和根因猜测
  3. Workflow 决定是否升级、派单、通知
  4. Agent 生成结构化工单摘要
  5. Workflow 写入 Jira / PagerDuty

这比“全链路都交给 Agent 随便发挥”稳得多。

十四、一个更完整的生产级示例：技术支持 Agent

下面给一个相对完整、可落地的示例，把前面的点串起来：

python
# 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)

十五、生产实践中的几个硬经验

1）不要把 Agent 当黑盒

LangGraph 官方强调了 durable execution、memory、human-in-the-loop、debugging with LangSmith 等底层能力。换句话说，生产 Agent 的关键不只是“推理成功”，而是可恢复、可观察、可干预。(LangChain Docs)

2）不要让工具成为无限接口

Tool 是给模型消费的，不是给工程师炫技的。能拆就拆，能限制就限制。

3）结构化输出尽量前置

只要后续系统要消费结果，就别让模型输出自由文本。

4）记忆要控量，不要恋旧

长上下文不是资产，很多时候是噪声。该 trim、summarize 就做。官方也把 summarization middleware 作为预置能力之一。(LangChain Docs)

5）高风险动作最好走 Workflow 或 HITL

尤其是删数据、改配置、发通知、扣费等动作。Agent 可以建议，但不一定应该直接执行。

6）Agent 不等于“全自动”

真正成熟的系统通常是：

十六、这套 LangChain Agent 体系最值得关注的升级点

如果你以前熟悉的是旧时代 LangChain 里各种 agent executor、memory patch、callback patch 的玩法，那么当前这套体系最值得注意的是：

  1. create_agent 成为官方标准入口，LangChain v1 明确推荐它，且相比旧的 create_react_agent 更易用，也更容易通过 middleware 做深度定制。(LangChain Docs)
  2. 底层统一到 LangGraph runtime，Agent 终于有了严肃的状态机和运行时基础。(LangChain Docs)
  3. Middleware 成为核心扩展机制，不再把工程控制逻辑塞到 prompt 或 callback 里。(LangChain Docs)
  4. Structured output 成为一等公民，这对应用开发尤其关键。(LangChain Docs)
  5. Memory 和 state 融合，从“聊天记录”升级成“线程级状态持久化”。(LangChain Docs)

十七、结语：真正的 Agent 开发，重点从来不是“让模型会调用工具”

如果要用一句话总结 LangChain 这套新 Agent 体系，我会这样说：

它的核心价值，不是让 LLM 获得“工具使用能力”，而是让 Agent 应用第一次有了接近后端系统的工程骨架。

这个骨架包括：

所以，今天再谈 Agent 开发，已经不能只谈 Prompt 和 Tool 了。真正决定一个 Agent 能不能上线的，是它是否具备可控性、可维护性、可观测性和可扩展性。

而 LangChain 当前这套 create_agent + LangGraph runtime + middleware + state 的设计，已经把这些关键问题摆到了台面上。(LangChain Docs)