业务人员通常不缺仪表盘，真正缺的是从“发现异常”到“采取行动”之间的可靠路径。Amazon Quick 可以承担业务入口：展示供应链指标、提供企业知识上下文，并接收规划人员的自然语言请求；NVIDIA NeMo Agent Toolkit 则在后端组织专用智能体工作流，把风险数据、业务规则和缓解措施组合成一份可审查的建议。
这类架构的价值不只是给仪表盘加一个聊天框，而是把分析结果变成受约束、可追踪的业务流程。例如，规划人员看到某个供应商的延迟概率上升后，可以继续询问影响范围、替代供应商和建议调拨量，而不必手工切换多个系统。
把业务入口与智能体执行层分开
可以把整个流程拆成两个职责清晰的平面：
- Amazon Quick 负责交互与上下文：呈现仪表盘、知识内容、当前筛选条件以及用户问题。
- NeMo Agent Toolkit 负责工作流编排：调用风险分析、库存查询、供应商查询和推荐生成等工具。
- 企业系统保留事实与执行权：ERP、采购、库存和物流系统仍然是记录来源；智能体默认只生成建议，不直接修改订单。
一次典型请求可能包含产品、区域、时间窗口和仪表盘当前选中的供应商。后端工作流先读取确定性数据，再让模型解释风险并组织建议。顺序很重要：如果先让模型自由推断，再补查数据，最终答案更容易混入过期库存、虚构供应商或无法执行的调拨量。
推荐的处理链路如下：
- 接收 Amazon Quick 页面传来的业务上下文。
- 校验产品编号、供应商编号和用户权限。
- 查询风险、库存、在途订单及可替代供应商。
- 用业务规则过滤不可行方案。
- 让模型比较剩余方案，并解释成本、时效和风险。
- 返回结构化建议、证据和待审批动作。
供应链智能体不是一个大提示词
专用工作流应当把任务拆成可测试的工具，而不是把数据库内容全部塞进提示词。以供应中断为例，可以设置以下能力：
get_risk_signal：获取供应商延迟概率、港口事件或交付波动。get_inventory_position：读取现货、安全库存和在途数量。find_alternative_suppliers：根据物料认证、区域和产能筛选候选供应商。calculate_mitigation：用确定性代码计算缺口、调拨量和预计覆盖天数。compose_recommendation：把计算结果转换成规划人员可读的建议。
其中，数量计算和资格过滤适合使用普通代码；模型更适合做比较、解释和信息归纳。不要让语言模型独立决定采购量，更不要让它在缺少审批的情况下创建采购订单。
工作流返回值也不应只有一段文本。更稳妥的响应结构应包含风险等级、证据、建议动作、假设和审批要求。例如：
{
"risk_level": "high",
"summary": "未来 14 天预计短缺 320 件",
"recommended_actions": [
{
"type": "inventory_transfer",
"quantity": 200,
"from": "WH-SIN",
"to": "WH-NRT"
},
{
"type": "expedite_purchase_order",
"purchase_order": "PO-10428",
"quantity": 120
}
],
"evidence": ["risk-signal-8821", "inventory-snapshot-2025-03-08"],
"requires_approval": true
}
这种结构便于 Amazon Quick 展示建议卡片，也便于审计系统记录“使用了哪些数据、提出了什么动作、由谁批准”。
可以这样实践：搭建一个可调用的最小工作流
下面是一个可直接运行的演示服务。它不宣称复现 Amazon Quick 或 NeMo Agent Toolkit 的具体接口，而是明确展示两者之间需要建立的 HTTP 契约。接入实际环境时，可以把 analyze_risk 替换为由 NeMo Agent Toolkit 编排的工具链，并把示例数据替换为经过授权的企业 API。
创建 app.py：
from typing import Literal
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field
app = FastAPI(title="Supply Chain Mitigation Agent")
class RiskRequest(BaseModel):
product_id: str = Field(min_length=1)
supplier_id: str = Field(min_length=1)
region: str = Field(min_length=2)
horizon_days: int = Field(default=14, ge=1, le=90)
demand_units: int = Field(gt=0)
available_units: int = Field(ge=0)
in_transit_units: int = Field(ge=0)
class Action(BaseModel):
type: Literal["monitor", "expedite", "source_alternative"]
quantity: int
reason: str
def analyze_risk(request: RiskRequest) -> dict:
projected_supply = request.available_units + request.in_transit_units
shortage = max(request.demand_units - projected_supply, 0)
if shortage == 0:
risk_level = "low"
actions = [
Action(
type="monitor",
quantity=0,
reason="Projected supply covers demand for the selected horizon.",
)
]
else:
risk_level = "high" if shortage >= request.demand_units * 0.25 else "medium"
expedite_quantity = min(shortage, request.in_transit_units)
actions = []
if expedite_quantity:
actions.append(
Action(
type="expedite",
quantity=expedite_quantity,
reason="Use existing in-transit supply before creating new demand.",
)
)
remaining = shortage - expedite_quantity
if remaining:
actions.append(
Action(
type="source_alternative",
quantity=remaining,
reason="Close the residual shortage with an approved supplier.",
)
)
return {
"risk_level": risk_level,
"shortage_units": shortage,
"recommended_actions": [action.model_dump() for action in actions],
"assumptions": [
"Demand remains unchanged during the selected horizon.",
"In-transit units arrive within the horizon if expedited."
],
"requires_approval": shortage > 0,
}
@app.post("/agent/supply-risk")
def supply_risk(
request: RiskRequest,
x_user_id: str | None = Header(default=None),
):
if not x_user_id:
raise HTTPException(status_code=401, detail="Missing user identity")
result = analyze_risk(request)
return {
"request_context": {
"user_id": x_user_id,
"product_id": request.product_id,
"supplier_id": request.supplier_id,
"region": request.region,
},
**result,
}
安装依赖并启动服务：
python -m venv .venv
source .venv/bin/activate
python -m pip install fastapi "uvicorn[standard]"
uvicorn app:app --host 0.0.0.0 --port 8000
在另一个终端模拟 Amazon Quick 发起的请求：
curl --fail-with-body http://localhost:8000/agent/supply-risk \
-H 'Content-Type: application/json' \
-H 'X-User-Id: planner-42' \
-d '{
"product_id": "SKU-1007",
"supplier_id": "SUP-23",
"region": "ap-northeast",
"horizon_days": 14,
"demand_units": 1000,
"available_units": 500,
"in_transit_units": 300
}'
这个输入会产生 200 件的预计缺口。示例工作流会优先建议加急已有在途库存，并把响应标记为需要审批。生产环境还应加入供应商认证、运输能力、最低订购量、货币和成本上限等约束。
在 NeMo Agent Toolkit 中可以沿用同样的职责边界：将风险查询、库存查询和方案计算注册为独立工具，再由工作流选择调用顺序。由于不同版本的组件名称和配置格式可能变化，具体注册代码应以项目安装版本为准；关键不是 YAML 字段名称，而是保证工具输入输出采用可验证的结构，并且每次调用都有身份、超时和审计信息。
提示词只负责解释，不负责创造事实
当确定性计算完成后，可以把结构化结果交给模型生成业务说明。下面的提示词可以作为工作流中的推荐生成节点：
You are a supply-chain mitigation assistant.
Use only the facts in TOOL_RESULTS. Do not invent inventory, suppliers,
lead times, prices, or purchase orders.
Return:
1. A two-sentence risk summary.
2. Up to three mitigation actions ordered by urgency.
3. Evidence used for each action.
4. Assumptions and missing data.
5. Whether human approval is required.
If TOOL_RESULTS conflict or are stale, do not recommend an execution action.
Ask the planner to refresh or verify the relevant source instead.
TOOL_RESULTS:
{{tool_results}}
这类约束能减少模型把“可能的替代供应商”写成“已经认证的供应商”。不过，提示词不是安全边界。服务端仍需验证模型输出，尤其是数量、供应商编号、采购订单编号和执行动作。
上线前检查：先做建议系统，再谈自动执行
引入这套架构时，可以先选择一个数据边界清晰的场景，例如单一区域的缺货风险，而不是一次覆盖全部供应链决策。上线前至少检查以下项目：
- Amazon Quick 传入的用户身份能否映射到后端数据权限。
- 仪表盘筛选条件是否随请求一起传递，并在响应中回显。
- 工具结果是否带有时间戳、来源标识和数据新鲜度。
- 计算逻辑能否用固定测试数据独立验证。
- 模型是否只能从认证供应商和允许动作中选择。
- 超时、工具失败和数据冲突时，系统是否停止给出执行性建议。
- 高成本调拨、加急和采购动作是否必须人工审批。
- 每次建议是否保存输入、工具调用、模型版本、输出和审批记录。
Amazon Quick 与 NeMo Agent Toolkit 的组合适合把业务上下文和专用智能体工作流连接起来，但它不应绕过原有控制体系。更稳妥的采用顺序是：先让系统解释风险，再生成结构化建议，随后接入审批，只有在数据质量、评估结果和审计能力都稳定后，才考虑有限度的自动执行。