一、引言：为什么跨境电商需要智能体中台
跨境电商行业有一个长期痛点：重复劳动多、系统割裂严重。客服每天回复大量同质化的物流查询、退换货咨询；运营团队要在Amazon、Walmart、独立站等多个平台之间来回切换，手动调广告、改价格、同步库存；供应链团队需要从ERP导出数据，再人工核对各平台的订单和库存差异。

传统RPA（机器人流程自动化）能解决一部分重复操作，但无法处理需要理解和推理的复杂任务。大模型的出现改变了这一点——LLM不仅能理解自然语言，还能通过Function Calling调用外部系统、通过RAG检索知识库、通过Agent工作流自主规划和执行任务。

本文将以LangChain + LangGraph为核心技术栈，从零搭建一个跨境电商智能体中台，涵盖知识库构建、Agent工作流编排、多系统对接、质量保障四个核心模块。

二、整体架构设计
智能体中台采用分层架构，从上到下依次为：

接入层：提供REST API和WebSocket接口，支持客服对话、运营指令、系统事件触发等多种交互方式

编排层：基于LangGraph构建的多Agent协作引擎，负责意图识别、任务分解、工作流调度

能力层：包括RAG检索、工具调用（Function Calling）、Prompt模板管理

数据层：向量数据库（Chroma）、关系数据库（PostgreSQL）、以及对接Amazon SP-API、Walmart API、ERP等外部系统的适配器

python

项目结构

ecommerce-agent-platform/
├── agent/
│ ├── graph.py # LangGraph 主图定义
│ ├── nodes/ # 各Agent节点
│ │ ├── classifier.py # 意图分类
│ │ ├── rag.py # RAG检索节点
│ │ └── tools.py # 工具调用节点
├── tools/
│ ├── amazon.py # Amazon API 工具
│ ├── erp.py # ERP 系统工具
│ └── order.py # 订单处理工具
├── knowledge/
│ ├── ingest.py # 知识入库
│ └── retrieval.py # 向量检索
├── evaluation/
│ ├── quality.py # 质量评估
│ └── fallback.py # 人工兜底
└── main.py # FastAPI 入口
三、知识库建设：RAG让大模型“懂业务”
大模型的通病是幻觉——它不知道你公司的具体退换货政策、不知道产品的真实库存、不知道广告投放的历史数据。RAG（检索增强生成）通过在生成回答之前先从知识库中检索相关信息，让大模型基于事实作答。

3.1 知识入库
我们将客服知识库、产品知识库、广告策略库、供应链规则库统一纳入RAG体系。以客服知识库为例：

python

knowledge/ingest.py

import os
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_community.document_loaders import DirectoryLoader, TextLoader

1. 加载文档

loader = DirectoryLoader(
“./data/knowledge_base/”,
glob=“**/*.md”,
loader_cls=TextLoader
)
documents = loader.load()

2. 文档切分（Chunking）

text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=[“\n\n”, “\n”, “。”, “！”, “？”, “；”, “，”, " ", “”]
)
chunks = text_splitter.split_documents(documents)

3. 向量化并存入Chroma

embeddings = OpenAIEmbeddings(model=“text-embedding-3-small”)
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory=“./data/chroma_db”
)
vectorstore.persist()
print(f"✅ 已入库 {len(chunks)} 个知识片段")
3.2 RAG检索与生成
python

knowledge/retrieval.py

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from langchain_chroma import Chroma

def create_rag_chain():
embeddings = OpenAIEmbeddings(model=“text-embedding-3-small”)
vectorstore = Chroma(
persist_directory=“./data/chroma_db”,
embedding_function=embeddings
)
retriever = vectorstore.as_retriever(search_kwargs={“k”: 5})

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

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)
return qa_chain

使用示例

qa = create_rag_chain()
result = qa.invoke(“客户问：退货需要什么条件？”)
print(result[“result”])

输出基于知识库的准确回答，而非大模型凭空编造

跨境电商场景中，知识库需要多语言支持。可以将FAQ、产品描述等分别用中英文向量化存储，检索时根据用户语言偏好选择对应的向量空间。

四、Agent工作流：从意图到执行的完整链路
单次RAG问答只能解决“查询类”问题。真正的智能体需要能自主规划和执行多步任务——比如“帮我把销量最好的三个SKU的广告预算提高20%”，这涉及查询销量、分析TOP3、调用广告平台API修改预算等多个步骤。

4.1 基于LangGraph的多Agent协作
LangGraph是LangChain生态中专门用于构建有状态、可编排的多Agent系统的框架。我们设计三个专用Agent：

Classifier Agent：识别用户意图（客服咨询 / 运营指令 / 供应链查询）

RAG Agent：处理知识检索类问题

Tool Agent：执行具体的系统操作（查订单、改广告、同步库存）

python

agent/graph.py

from typing import Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from pydantic import BaseModel
from typing import List, Optional

class AgentState(BaseModel):
messages: List[dict] # 对话历史
intent: Optional[str] # 识别出的意图
retrieved_docs: Optional[List] # RAG检索结果
tool_results: Optional[dict] # 工具调用结果
final_answer: Optional[str] # 最终回复
needs_human: bool = False # 是否需要人工介入

创建图

workflow = StateGraph(AgentState)

定义节点

def classify_intent(state: AgentState) -> AgentState:
“”“意图分类节点”“”
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model=“gpt-4o-mini”, temperature=0)

prompt = f"""分析用户意图，只返回以下类别之一：
- knowledge: 咨询知识库（政策、规则、产品信息）
- operation: 执行操作（查订单、改广告、调库存）
- other: 其他

用户消息：{state.messages[-1]['content']}
"""
response = llm.invoke(prompt)
state.intent = response.content.strip().lower()
return state

def rag_retrieve(state: AgentState) -> AgentState:
“”“RAG检索节点”“”
from knowledge.retrieval import create_rag_chain
qa = create_rag_chain()
result = qa.invoke(state.messages[-1][‘content’])
state.retrieved_docs = result.get(“source_documents”, [])
state.final_answer = result[“result”]
return state

def execute_tools(state: AgentState) -> AgentState:
“”“工具调用节点”“”
from tools.dispatcher import ToolDispatcher
dispatcher = ToolDispatcher()
result = dispatcher.execute(state.messages[-1][‘content’])
state.tool_results = result
# 根据工具执行结果生成最终回复
state.final_answer = format_tool_result(result)
return state

def check_quality(state: AgentState) -> AgentState:
“”“质量检查节点”“”
from evaluation.quality import QualityChecker
checker = QualityChecker()
score = checker.evaluate(state.final_answer)
if score < 0.7:
state.needs_human = True
return state
4.2 定义路由逻辑
python

agent/graph.py (续)

def route_after_classify(state: AgentState) -> Literal[“rag”, “tools”, “fallback”]:
“”“根据意图路由到不同节点”“”
if state.intent == “knowledge”:
return “rag”
elif state.intent == “operation”:
return “tools”
else:
return “fallback”

def route_after_execution(state: AgentState) -> Literal[“quality_check”, “human”]:
“”“执行后检查是否需要人工介入”“”
return “quality_check” if not state.needs_human else “human”

构建图

workflow.add_node(“classify”, classify_intent)
workflow.add_node(“rag”, rag_retrieve)
workflow.add_node(“tools”, execute_tools)
workflow.add_node(“quality_check”, check_quality)
workflow.add_node(“human”, lambda s: s) # 人工兜底节点

workflow.set_entry_point(“classify”)
workflow.add_conditional_edges(“classify”, route_after_classify)
workflow.add_edge(“rag”, “quality_check”)
workflow.add_edge(“tools”, “quality_check”)
workflow.add_conditional_edges(“quality_check”, route_after_execution)
workflow.add_edge(“human”, END)
workflow.add_edge(“quality_check”, END)

编译并添加记忆

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
4.3 运行Agent
python

main.py

from agent.graph import app

def process_user_message(user_id: str, message: str):
“”“处理用户消息，支持多轮对话”“”
config = {“configurable”: {“thread_id”: user_id}}
state = {“messages”: [{“role”: “user”, “content”: message}]}

result = app.invoke(state, config)

return {
    "answer": result.get("final_answer", "抱歉，我暂时无法处理这个请求"),
    "needs_human": result.get("needs_human", False),
    "intent": result.get("intent")
}

五、工具集成：让Agent对接真实业务系统
Agent的“手脚”就是工具（Tools） 。通过Function Calling，大模型可以决定调用哪个工具、传什么参数。

5.1 定义工具
以订单查询和广告预算调整为例：

python

tools/order.py

from langchain.tools import tool
import httpx

@tool
def query_order(order_id: str) -> str:
“”"
查询订单状态和物流信息。
Args:
order_id: 订单编号，如 ORD-2026-001
“”"
# 对接Amazon SP-API 或 ERP 系统[reference:10]
response = httpx.get(
f"https://api.erp.company.com/orders/{order_id}“,
headers={“Authorization”: f"Bearer {os.getenv(‘ERP_TOKEN’)}”}
)
if response.status_code == 200:
data = response.json()
return f"订单 {order_id} 状态：{data[‘status’]}，物流：{data[‘tracking’]}"
return f"未找到订单 {order_id}"

@tool
def adjust_ad_budget(product_ids: list, percentage: float) -> str:
“”"
调整指定商品的广告预算。
Args:
product_ids: 商品ID列表
percentage: 调整百分比，正数增加，负数减少
“”"
# 对接Amazon Advertising API[reference:11]
results = []
for pid in product_ids:
# 实际调用广告平台API
new_budget = calculate_new_budget(pid, percentage)
results.append(f"{pid}: 预算调整为 {new_budget}")
return “\n”.join(results)
5.2 工具调度器
python

tools/dispatcher.py

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from tools.order import query_order, adjust_ad_budget

class ToolDispatcher:
def init(self):
self.tools = [query_order, adjust_ad_budget]
self.llm = ChatOpenAI(model=“gpt-4o”, temperature=0)

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个跨境电商运营助手，根据用户需求调用合适的工具。"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}")
    ])
    
    self.agent = create_openai_tools_agent(self.llm, self.tools, prompt)
    self.executor = AgentExecutor(agent=self.agent, tools=self.tools, verbose=True)

def execute(self, query: str) -> dict:
    result = self.executor.invoke({"input": query})
    return {"output": result["output"]}

六、质量保障：评估、异常检测与人工兜底
AI Agent在生产环境中最大的风险是不可控。模型可能产生幻觉、工具调用可能失败、Token可能超预算。我们需要建立三层防护：

6.1 输出质量评估
使用LLM-as-Judge方式评估回答质量：

python

evaluation/quality.py

from langchain_openai import ChatOpenAI

class QualityChecker:
def init(self):
self.llm = ChatOpenAI(model=“gpt-4o-mini”, temperature=0)

def evaluate(self, answer: str) -> float:
    prompt = f"""请从以下维度评估回答质量，给出0-1之间的分数：
    1. 准确性：回答是否基于事实，有无幻觉
    2. 完整性：是否完整回答了用户问题
    3. 可用性：信息是否可直接使用
    
    回答内容：{answer}
    
    只返回一个0-1之间的数字。
    """
    response = self.llm.invoke(prompt)
    try:
        score = float(response.content.strip())
        return min(max(score, 0), 1)
    except:
        return 0.5

6.2 异常检测与兜底
python

evaluation/fallback.py

import logging
from typing import Optional

class FallbackManager:
def init(self):
self.quality_threshold = 0.7
self.max_retries = 2
self.logger = logging.getLogger(name)

def handle(self, agent_result: dict, user_message: str) -> dict:
    """处理Agent执行结果，必要时触发兜底"""
    
    # 场景1：质量分数过低
    if agent_result.get("quality_score", 0) < self.quality_threshold:
        self.logger.warning(f"质量分数过低: {agent_result['quality_score']}")
        return {
            "action": "human_escalation",
            "message": "这个问题比较复杂，已转接人工客服，请稍候...",
            "context": agent_result
        }
    
    # 场景2：工具调用异常
    if agent_result.get("tool_error"):
        self.logger.error(f"工具调用失败: {agent_result['tool_error']}")
        if agent_result.get("retry_count", 0) < self.max_retries:
            return {"action": "retry", "message": "系统暂时繁忙，正在重试..."}
        return {
            "action": "human_escalation",
            "message": "系统暂时无法处理，已为您转接人工...",
            "context": agent_result
        }
    
    # 场景3：Token超预算
    if agent_result.get("token_usage", 0) > 10000:
        return {
            "action": "simplify",
            "message": "您的请求涉及范围较广，我将精简回答..."
        }
    
    return {"action": "success", "message": agent_result.get("final_answer")}

6.3 全链路可观测性
使用LangSmith追踪Agent的每一步执行——调用了哪个模型、执行了什么工具、消耗了多少Token、每一步的耗时：

python

在环境变量中启用 LangSmith 追踪

os.environ[“LANGCHAIN_TRACING_V2”] = “true”
os.environ[“LANGCHAIN_ENDPOINT”] = “https://api.smith.langchain.com”
os.environ[“LANGCHAIN_API_KEY”] = “your-langsmith-key”
os.environ[“LANGCHAIN_PROJECT”] = “ecommerce-agent”

之后所有LangChain调用会自动上报到LangSmith

可以在 Web 控制台查看完整的调用链

七、部署与上线
python

main.py - FastAPI 服务

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent.graph import app
from evaluation.fallback import FallbackManager

app = FastAPI(title=“跨境电商智能体中台”)
fallback = FallbackManager()

class ChatRequest(BaseModel):
user_id: str
message: str
session_id: Optional[str] = None

class ChatResponse(BaseModel):
answer: str
needs_human: bool = False
session_id: str

@app.post(“/chat”, response_model=ChatResponse)
async def chat(request: ChatRequest):
try:
config = {“configurable”: {“thread_id”: request.user_id}}
state = {“messages”: [{“role”: “user”, “content”: request.message}]}

    result = app.invoke(state, config)
    
    # 质量检查与兜底
    handled = fallback.handle(result, request.message)
    
    return ChatResponse(
        answer=handled["message"],
        needs_human=handled.get("action") == "human_escalation",
        session_id=request.session_id or request.user_id
    )
except Exception as e:
    # 全局兜底
    return ChatResponse(
        answer="系统暂时繁忙，您的请求已记录，人工客服将尽快联系您。",
        needs_human=True,
        session_id=request.session_id or request.user_id
    )

启动服务：

bash
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
八、总结
本文从零搭建了一个跨境电商智能体中台，覆盖了：

知识库：通过RAG让大模型基于企业真实数据作答，降低幻觉

Agent工作流：基于LangGraph实现多Agent协作，支持意图识别→RAG检索→工具调用的完整链路

工具集成：通过Function Calling对接Amazon、ERP等业务系统

质量保障：建立质量评估、异常检测、人工兜底三层防护机制

这套架构已在多个电商场景中得到验证——智能客服可将70%的常见咨询自动化，运营Agent可将广告调优、库存同步等重复操作效率提升5倍以上。

未来可以继续演进的方向包括：引入多模态能力处理商品图片和视频、建设Agent记忆系统实现跨会话的上下文理解、以及通过强化学习让Agent在真实业务反馈中持续优化。

大模型正在从“聊天工具”演变为“数字员工”。让它们接管重复劳动，把人类释放到更需要创造力的工作中去——这正是智能体中台的价值所在。