从 LangChain 到 OpenClaw：AI Agent 工程化的五层拼图与生产落地全攻略

真正能上线的 Agent，从来不是“模型 + Prompt”这么简单。它本质上是一套分层系统：底层要有可靠的推理运行时，中层要有工作流和知识编排，上层要有能力封装、渠道接入与工程化交付，外围还要补齐安全、观测、弹性和治理。本文以 LangChain、Dify、Agent Skills、OpenClaw、Harness 为线索，系统讲清 AI Agent 从原理到架构、从代码到生产的完整落地路径。

一、先把问题说透：为什么很多 Agent Demo 一上线就失灵

过去一年，很多团队都经历过类似路径：

1. 1. 用一个大模型加几个 Tool，很快做出一个“会对话、会查数据、会调接口”的 Demo。
2. 2. 在测试环境里效果惊艳，产品、运营、老板都觉得“这事快成了”。
3. 3. 一旦接入真实业务流量，问题开始集中爆发。

典型故障几乎总是这几类：

• • 对话轮次一长，上下文爆炸，响应时间和 Token 成本同时失控。
• • Tool 调用缺少边界控制，Agent 把“查询订单”一路推演成“自动退款”。
• • 模型输出不稳定，JSON 时好时坏，下游服务一片报错。
• • 接入多个业务系统后，幂等、超时、重试、补偿没人兜底。
• • 研发能写，运营不会改；产品想调流程，必须找工程师发版。
• • 出了线上问题，日志里只有一句 “LLM call failed”，根本追不到是模型、向量检索、审批服务还是消息通道出了问题。

这说明一个关键事实：

Agent 的核心难点，不是“让模型回答问题”，而是“让智能行为在工程系统中可控地运行”。

所以，成熟的 Agent 体系必须同时解决五类问题：

• • 运行时问题：模型如何思考、调用工具、维护状态。
• • 平台问题：流程怎么编排，知识怎么治理，业务怎么配置。
• • 能力复用问题：领域知识和动作模板如何沉淀，而不是每个项目从零写 Prompt。
• • 触达问题：用户在 Slack、飞书、Web、CLI 等多个入口发起请求时，如何共享上下文与策略。
• • 交付问题：如何测试、发布、灰度、回滚、观测、审计。

这正是本文五层拼图的由来：

• • LangChain / LangGraph：运行时层
• • Dify：平台层
• • Agent Skills：能力层
• • OpenClaw：渠道与执行入口层
• • Harness：交付与治理层

二、五层拼图不是“选型列表”，而是一张生产架构图

很多文章把这些工具写成横向对比，仿佛只能五选一。实际恰恰相反，它们更多是纵向协作关系。

可以先看一张简化分层图：

┌──────────────────────────────────────────────────────────────┐
│                    OpenClaw / 渠道接入层                     │
│     Slack / 飞书 / Teams / Web / CLI / Mobile / Bot         │
├──────────────────────────────────────────────────────────────┤
│                     Dify / 平台编排层                        │
│      Workflow / RAG / PromptOps / Model Routing / API       │
├──────────────────────────────────────────────────────────────┤
│                LangChain + LangGraph / 运行时层              │
│   Agent Loop / State Graph / Tool Calling / Memory / HITL   │
├──────────────────────────────────────────────────────────────┤
│                   Agent Skills / 能力封装层                  │
│  Domain Prompt / SOP / Script / Policy / Reference / Asset  │
├──────────────────────────────────────────────────────────────┤
│                Harness / 工程化交付与治理层                  │
│   CI/CD / Verification / Canary / Rollback / Audit / SRE    │
└──────────────────────────────────────────────────────────────┘

如果用一句话概括：

LangChain 决定 Agent “怎么运行”，Dify 决定业务“怎么配置”，Skills 决定能力“怎么复用”，OpenClaw 决定结果“怎么到达用户”，Harness 决定系统“怎么稳定上线”。

这五层分别解决不同控制面：

三、第一层：LangChain / LangGraph 是 Agent 的运行时内核

3.1 为什么说 Agent 不是函数调用，而是状态机

很多人把 Agent 理解成：

用户输入 -> LLM -> 输出答案

而生产级 Agent 更接近：

用户输入
  -> 意图识别
  -> 状态装载
  -> 检索上下文
  -> 规划动作
  -> 调用工具
  -> 验证结果
  -> 可能人工审批
  -> 写入记忆
  -> 输出答案

这已经不是一个普通函数，而是一个可中断、可恢复、可追踪的有状态流程。

从原理上看，Agent 至少包含五个核心部件：

1. 1. Planner：决定下一步做什么。
2. 2. Tool Executor：真正执行外部动作。
3. 3. Memory Manager：管理短期上下文、长期记忆与摘要。
4. 4. Policy Layer：审批、脱敏、限流、重试、审计。
5. 5. State Store：让流程可恢复、可重放、可追踪。

LangChain 提供高层 Agent 抽象，LangGraph 更像底层状态图运行时。对于简单场景，create_agent 很高效；对于多步骤、长流程、可恢复任务，图式编排更稳。

3.2 单 Agent 的上限，往往不是模型，而是上下文和副作用

一个只读型 FAQ Agent 很容易做；难的是具备真实业务动作的 Agent，例如：

• • 查订单
• • 发起退款
• • 查询库存
• • 修改收货地址
• • 触发人工审批
• • 通知物流系统

这些动作会带来三个工程问题：

• • 副作用：一旦写库、发消息、调支付接口，就必须考虑幂等。
• • 不确定性：模型可能反复调用工具，也可能输出错误参数。
• • 长耗时：库存、物流、审批这类调用可能横跨秒级到分钟级。

因此，生产级 LangChain 方案不应该只是“把 Tool 注册进去”，而应该建立明确边界：

• • Agent 只负责决策，不直接持有关键写操作权限。
• • 所有写操作必须经由受控应用服务。
• • 工具调用结果必须结构化、可验证、可审计。
• • 高风险动作必须支持人工确认和异步补偿。

3.3 一个更接近生产的 LangChain 服务骨架

下面这段代码不是玩具 Demo，而是一个更接近真实生产的最小骨架。重点在四件事：

1. 1. 结构化输出
2. 2. 工具超时与重试
3. 3. 幂等键
4. 4. 审批与审计边界

from __future__ import annotations

import asyncio
import json
import time
import uuid
from dataclasses import dataclass
from typing import Any

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI


class ChatRequest(BaseModel):
    session_id: str
    user_id: str
    message: str
    tenant_id: str


class ChatResponse(BaseModel):
    session_id: str
    answer: str
    trace_id: str
    requires_approval: bool = False
    approval_task_id: str | None = None


@dataclass
class ToolContext:
    tenant_id: str
    user_id: str
    trace_id: str
    request_id: str


class OrderGateway:
    async def query_order(self, tenant_id: str, order_id: str) -> dict[str, Any]:
        await asyncio.sleep(0.05)
        return {
            "order_id": order_id,
            "status": "SHIPPED",
            "amount": 299.0,
            "currency": "CNY",
            "can_refund": True,
        }

    async def create_refund_request(
        self,
        tenant_id: str,
        order_id: str,
        reason: str,
        idempotency_key: str,
) -> dict[str, Any]:
        await asyncio.sleep(0.05)
        return {
            "refund_request_id": f"rf-{order_id}",
            "status": "PENDING_APPROVAL",
            "reason": reason,
            "idempotency_key": idempotency_key,
        }


order_gateway = OrderGateway()
app = FastAPI(title="agent-service")


def with_timeout(seconds: float):
    def decorator(fn):
        async def wrapper(*args, **kwargs):
            return await asyncio.wait_for(fn(*args, **kwargs), timeout=seconds)
        return wrapper
    return decorator


@tool
@with_timeout(2.0)
async def query_order(order_id: str, tenant_id: str) -> str:
    """查询订单详情，只读工具。"""
    result = await order_gateway.query_order(tenant_id=tenant_id, order_id=order_id)
    return json.dumps(result, ensure_ascii=False)


@tool
@with_timeout(2.0)
async def apply_refund(order_id: str, reason: str, tenant_id: str, request_id: str) -> str:
    """提交退款申请，该工具只创建审批单，不直接退款。"""
    result = await order_gateway.create_refund_request(
        tenant_id=tenant_id,
        order_id=order_id,
        reason=reason,
        idempotency_key=request_id,
    )
    return json.dumps(result, ensure_ascii=False)


llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

agent = create_agent(
    model=llm,
    tools=[query_order, apply_refund],
    system_prompt=(
        "你是电商订单助理。"
        "必须优先调用工具获取事实。"
        "涉及退款时只能创建退款申请，不得声称已经完成退款。"
        "如果工具返回 PENDING_APPROVAL，必须明确告知用户等待审批。"
    ),
)


@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest) -> ChatResponse:
    trace_id = str(uuid.uuid4())
    request_id = f"{req.session_id}-{int(time.time() * 1000)}"

    try:
        result = await agent.ainvoke(
            {
                "messages": [{"role": "user", "content": req.message}],
                "tenant_id": req.tenant_id,
                "request_id": request_id,
            }
        )
    except asyncio.TimeoutError as exc:
        raise HTTPException(status_code=504, detail="agent timeout") from exc
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f"agent failed: {exc}") from exc

    final_message = result["messages"][-1].content
    requires_approval = "PENDING_APPROVAL" in str(result)

    return ChatResponse(
        session_id=req.session_id,
        answer=final_message,
        trace_id=trace_id,
        requires_approval=requires_approval,
        approval_task_id=request_id if requires_approval else None,
    )

这段代码体现了一个非常关键的设计原则：

Agent 可以发起“业务意图”，但不能绕过业务系统直接执行“最终动作”。

也就是说，“退款”不等于“直接调用支付系统出款”，而应该是“创建退款申请 -> 进入审批流 -> 审批通过后由事务型服务执行”。

这会让 Agent 从“危险的自动脚本”变成“受控的智能入口”。

3.4 为什么复杂场景必须上 LangGraph

当你遇到以下需求时，单次 ReAct Loop 就不够了：

• • 一个任务要跑 10 分钟以上
• • 中途需要人工介入
• • 失败后要从中间步骤恢复
• • 多个 Specialist Agent 协作
• • 同一个会话里要维护长期状态

这时，图式状态机比“无限循环 + if/else”更可靠。一个典型状态图如下：