大模型负责“想”，你的代码负责“做”——Function Calling就是连接思考与行动的桥梁。

在前两篇文章中，我们搭建了AI智能体的整体架构，构建了多业务知识库让AI拥有了“记忆”。但一个只会“想”和“记”的AI，距离一个真正的数字员工还有一步之遥——它还不能“动手” 。

今天这篇文章，我们就来深入讲解Function Calling（函数调用）与任务编排的实战技术。大模型本身无法直接调用外部API、操作数据库或执行业务逻辑——它只能输出文本。Function Calling的实质是：大模型在响应中表达调用特定工具的意图（而非直接执行），你的应用程序解析这个意图，实际执行函数，再将结果返回给模型。

你可以把Function Calling理解为大模型的 “派单系统” ——模型不亲自干活，它只负责判断“这个问题需要调哪个工具、传什么参数”。真正执行代码、调用API、操作系统的，是你自己的后端程序。

一、Function Calling的核心机制：从“对话”到“行动”

1.1 基本工作流

一个完整的Function Calling周期包含五个步骤：

1. 【定义工具】开发者用JSON Schema描述工具的name、description和parameters
2. 【发送请求】用户提问时，将工具定义随prompt一起发给大模型
3. 【模型决策】大模型判断是否需要调用工具，如需要则返回tool_calls（含函数名和参数JSON）
4. 【执行工具】你的后端解析tool_calls，实际执行对应的函数/API调用
5. 【返回结果】将执行结果传回模型，模型生成最终的自然语言回复

1.2 OpenAI风格的基础实现

以下是一个基于OpenAI API的完整Function Calling示例：

import openai
import json

client = openai.OpenAI()

# Step 1: 定义工具（JSON Schema格式）
tools = [{
    "type": "function",
    "function": {
        "name": "query_order",
        "description": "查询跨境电商订单的状态和物流信息",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "订单号，格式如 EB12345678"
                },
                "platform": {
                    "type": "string",
                    "enum": ["amazon", "walmart", "shopee"],
                    "description": "订单所属平台"
                }
            },
            "required": ["order_id"]
        },
        "strict": True
    }
}]

# Step 2: 发送请求
messages = [
    {"role": "system", "content": "你是一个跨境电商客服助手，可以帮助用户查询订单信息。"},
    {"role": "user", "content": "帮我查一下订单EB12345678的状态"}
]

response = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=messages,
    tools=tools,
    tool_choice="auto"  # auto表示让模型自行决定是否调用工具
)

# Step 3: 检查模型是否要求调用工具
message = response.choices[0].message

if message.tool_calls:
    for tool_call in message.tool_calls:
        # 解析参数
        function_name = tool_call.function.name
        arguments = json.loads(tool_call.function.arguments)
        
        print(f"模型请求调用: {function_name}")
        print(f"参数: {arguments}")
        
        # Step 4: 实际执行工具（这里模拟调用业务API）
        if function_name == "query_order":
            order_id = arguments["order_id"]
            # 实际场景中，这里会调用 Amazon SP-API 或 ERP 接口
            result = {
                "order_id": order_id,
                "status": "已发货",
                "tracking_number": "1Z999AA10123456784",
                "estimated_delivery": "2026-07-22",
                "carrier": "UPS"
            }
            
            # Step 5: 将结果返回给模型
            messages.append(message)  # 添加助手的tool_calls消息
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })
            
            # 让模型基于工具结果生成最终回复
            final_response = client.chat.completions.create(
                model="gpt-4.1-mini",
                messages=messages
            )
            print(final_response.choices[0].message.content)

1.3 Anthropic Claude的工具调用

Anthropic Claude也提供了类似的能力，称为 “Tool Use” 。Claude会在需要时返回tool_use块，由你的应用执行：

import anthropic

client = anthropic.Anthropic()

# 定义工具
tools = [{
    "name": "check_inventory",
    "description": "查询指定SKU在指定仓库的库存数量",
    "input_schema": {
        "type": "object",
        "properties": {
            "sku": {
                "type": "string",
                "description": "商品SKU编码，如 SKU-A100"
            },
            "warehouse": {
                "type": "string",
                "enum": ["US-West", "US-East", "EU-Frankfurt"],
                "description": "仓库代码"
            }
        },
        "required": ["sku"]
    }
}]

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=tools,
    messages=[
        {"role": "user", "content": "帮我查一下SKU-A100在美西仓库还有多少库存"}
    ]
)

# 检查是否有tool_use块
for content in response.content:
    if content.type == "tool_use":
        print(f"Claude请求调用: {content.name}")
        print(f"参数: {content.input}")
        # 执行工具并返回结果...

Claude的Tool Use与OpenAI的实现思路一致，区别在于返回格式和API调用方式不同。在跨境电商场景中，你可以根据团队的技术栈和对模型的选择偏好来决定使用哪家。

二、跨境电商工具体系：从业务需求到工具定义

要让AI Agent真正在跨境电商场景中“干活”，首先需要把业务能力抽象为标准化的工具（Tools） 。每个工具本质上是一个带有清晰描述和参数规范的函数接口。

2.1 跨境电商核心工具清单

根据第一篇文章中的业务场景分析，我们需要为AI Agent配备以下工具：

工具名称	功能描述	风险等级	对应业务线
query_order	查询订单状态和物流信息	低	客服
query_inventory	查询商品库存	低	供应链
search_products	搜索商品信息	低	客服/运营
create_return	创建退货单	高	客服
adjust_price	调整商品价格	高	广告/运营
adjust_budget	调整广告预算	高	广告
sync_inventory	同步库存到各平台	高	供应链
query_ad_performance	查询广告投放数据	低	广告

2.2 使用LangChain的@tool装饰器定义工具

LangChain提供了多种定义工具的方式，其中最简洁的是使用@tool装饰器：

from langchain.tools import tool
from typing import Optional

# 低风险工具：查询订单
@tool
def query_order(order_id: str, platform: Optional[str] = "amazon") -> str:
    """
    查询跨境电商订单的状态和物流信息。
    
    Args:
        order_id: 订单号，格式如 EB12345678
        platform: 订单所属平台，可选 amazon/walmart/shopee，默认 amazon
    
    Returns:
        包含订单状态、物流单号、预计送达时间的JSON字符串
    """
    # 实际场景中，这里会调用平台API
    # 例如 Amazon SP-API: 
    # response = requests.get(
    #     f"https://sellingpartnerapi.amazon.com/orders/v0/orders/{order_id}",
    #     headers={"x-amz-access-token": get_access_token()}
    # )
    
    # 模拟返回
    return f'{{"order_id": "{order_id}", "status": "已发货", "tracking": "1Z999AA10123456784", "eta": "2026-07-22"}}'


# 低风险工具：查询库存
@tool
def query_inventory(sku: str, warehouse: Optional[str] = "US-West") -> str:
    """
    查询指定SKU在指定仓库的库存数量。
    
    Args:
        sku: 商品SKU编码
        warehouse: 仓库代码，默认 US-West
    
    Returns:
        包含可用库存、预留库存的JSON字符串
    """
    # 实际场景中，这里会调用ERP/WMS的库存API
    return f'{{"sku": "{sku}", "available": 156, "reserved": 23, "warehouse": "{warehouse}"}}'


# 高风险工具：调整价格（需要人工审核）
@tool
def adjust_price(sku: str, new_price: float, platform: str = "amazon") -> str:
    """
    调整商品在指定平台上的销售价格。
    
    ⚠️ 此操作涉及资金变动，执行前需经过人工审核。
    
    Args:
        sku: 商品SKU编码
        new_price: 新价格（美元）
        platform: 目标平台
    
    Returns:
        价格调整结果
    """
    # 实际场景中，这里会调用平台的定价API
    # 注意：此类高风险操作应通过人工审核流程
    return f'{{"sku": "{sku}", "old_price": 29.99, "new_price": {new_price}, "platform": "{platform}", "status": "pending_review"}}'


# 高风险工具：调整广告预算
@tool
def adjust_ad_budget(campaign_id: str, new_budget: float, platform: str = "amazon_ads") -> str:
    """
    调整广告活动的日预算。
    
    ⚠️ 此操作涉及广告费用支出，执行前需经过人工审核。
    
    Args:
        campaign_id: 广告活动ID
        new_budget: 新日预算（美元）
        platform: 广告平台
    
    Returns:
        预算调整结果
    """
    return f'{{"campaign_id": "{campaign_id}", "old_budget": 50.0, "new_budget": {new_budget}, "status": "pending_review"}}'

关键点：@tool装饰器会自动将函数的docstring作为工具描述传给大模型。描述写得越清晰，模型调用工具的准确率就越高。对于高风险工具，应在描述中明确标注需要审核。

三、ReAct模式：让AI“边想边做”

单次Function Call只能完成一步操作。但真实的业务场景往往是多步的——用户说“帮我查一下订单EB12345678，如果已经发货了，顺便把同款的库存也查一下”，这需要依次调用query_order和query_inventory两个工具。

ReAct（Reasoning + Acting）模式正是为此而生。它让模型交替进行 “思考→行动→观察” 的循环，直到完成任务。

3.1 ReAct的核心循环

用户提问
    ↓
【Thought】模型思考：需要做什么？先做什么？
    ↓
【Action】模型选择并调用一个工具
    ↓
【Observation】系统返回工具执行结果
    ↓
【Thought】模型思考：结果符合预期吗？还需要做什么？
    ↓
（循环直到任务完成）
    ↓
【Final Answer】模型输出最终答案

3.2 LangChain ReAct Agent实现

from langchain.agents import create_react_agent, AgentExecutor
from langchain.chat_models import init_chat_model
from langchain.prompts import PromptTemplate

# 1. 初始化大模型
llm = init_chat_model(
    model="deepseek-chat",
    model_provider="deepseek",
    temperature=0.3,  # 降低温度，让决策更稳定
)

# 2. 注册所有可用工具
tools = [query_order, query_inventory, search_products, adjust_price, adjust_ad_budget]

# 3. 定义ReAct提示词模板
react_prompt = PromptTemplate.from_template("""
你是一个跨境电商AI助手，可以使用以下工具来帮助用户：

{tools}

工具名称: {tool_names}

请按照以下格式回答，每一步都要经过"思考→行动→观察"：

Question: 用户的问题
Thought: 你需要思考当前应该做什么
Action: 你要调用的工具名称，必须是 [{tool_names}] 中的一个
Action Input: 工具的输入参数（JSON格式）
Observation: 工具返回的结果
... (这个循环可以重复多次)
Thought: 我现在知道最终答案了
Final Answer: 给用户的最终回复

开始！

Question: {input}
Thought: {agent_scratchpad}
""")

# 4. 创建Agent
agent = create_react_agent(llm, tools, react_prompt)

# 5. 创建Agent执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 打印详细执行过程
    max_iterations=5,  # 防止无限循环
    handle_parsing_errors=True,
)

# 6. 执行任务
result = agent_executor.invoke({
    "input": "帮我查一下订单EB12345678的状态，然后查一下同款商品SKU-A100在美西仓库的库存"
})

print(result["output"])

3.3 执行过程可视化

当上述代码运行时，你会看到类似这样的输出：

> Entering new AgentExecutor chain...

Thought: 用户想查询订单状态和库存信息。我需要先查订单，再查库存。
Action: query_order
Action Input: {"order_id": "EB12345678", "platform": "amazon"}
Observation: {"order_id": "EB12345678", "status": "已发货", "tracking": "1Z999AA10123456784", "eta": "2026-07-22"}

Thought: 订单已发货。现在需要查询SKU-A100的库存。
Action: query_inventory
Action Input: {"sku": "SKU-A100", "warehouse": "US-West"}
Observation: {"sku": "SKU-A100", "available": 156, "reserved": 23, "warehouse": "US-West"}

Thought: 我已经获得了订单状态和库存信息，可以回答用户了。
Final Answer: 您的订单EB12345678已于2026-07-22发货，物流单号1Z999AA10123456784。同款商品SKU-A100在美西仓库现有库存156件（其中23件已被预留）。

> Finished chain.

ReAct模式的价值在于：它将复杂的多步任务拆解为可追溯的“思考-行动-观察”序列。每一步都有据可查，便于调试和优化。

四、任务编排：让Agent按正确的顺序做事

4.1 顺序编排 vs 并行编排

跨境电商场景中，有些任务必须按顺序执行（串行），有些则可以同时执行（并行）。

顺序编排示例（必须先查库存，才能决定是否调价）：

1. query_inventory(sku="SKU-A100") → 库存充足
2. adjust_price(sku="SKU-A100", new_price=25.99) → 执行调价

并行编排示例（两个独立查询可以同时进行）：

1. query_order(order_id="EB12345678") ─┐
2. query_inventory(sku="SKU-A100")    ─┼→ 同时执行，汇总结果

4.2 基于LangGraph的任务编排

LangGraph是LangChain生态中专门用于构建有状态、可编排的Agent工作流的框架。它通过定义“节点”和“边”来精确控制执行流程：

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Dict, Any
import json

# 定义状态结构
class AgentState(TypedDict):
    messages: List[Dict[str, Any]]
    current_tool: str
    tool_results: List[Dict[str, Any]]
    need_review: bool
    review_status: str  # pending / approved / rejected

# 定义节点函数
def parse_intent(state: AgentState) -> AgentState:
    """解析用户意图，决定需要调用哪些工具"""
    # 这里调用LLM分析用户输入
    # 简化示例
    state["current_tool"] = "query_order"
    return state

def execute_tool(state: AgentState) -> AgentState:
    """执行工具调用"""
    tool_name = state["current_tool"]
    
    # 工具路由
    if tool_name == "query_order":
        result = query_order.invoke({"order_id": "EB12345678"})
    elif tool_name == "query_inventory":
        result = query_inventory.invoke({"sku": "SKU-A100"})
    elif tool_name == "adjust_price":
        # 高风险工具，标记需要审核
        state["need_review"] = True
        result = adjust_price.invoke({"sku": "SKU-A100", "new_price": 25.99})
    else:
        result = f"未知工具: {tool_name}"
    
    state["tool_results"].append({
        "tool": tool_name,
        "result": result
    })
    return state

def check_review(state: AgentState) -> str:
    """检查是否需要人工审核"""
    if state.get("need_review", False):
        return "wait_review"
    return "generate_response"

def wait_human_review(state: AgentState) -> AgentState:
    """等待人工审核（实际场景中会挂起任务，等待回调）"""
    # 实际实现中，这里会：
    # 1. 将待审核任务推送到队列/看板
    # 2. 通过Slack/邮件通知审核人
    # 3. 等待审核结果回调
    state["review_status"] = "approved"  # 模拟审核通过
    return state

def generate_response(state: AgentState) -> AgentState:
    """基于工具执行结果生成最终回复"""
    # 将工具结果传给LLM生成自然语言回复
    return state

# 构建工作流图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("parse_intent", parse_intent)
workflow.add_node("execute_tool", execute_tool)
workflow.add_node("wait_review", wait_human_review)
workflow.add_node("generate_response", generate_response)

# 添加边（定义执行顺序）
workflow.set_entry_point("parse_intent")
workflow.add_edge("parse_intent", "execute_tool")
workflow.add_conditional_edges(
    "execute_tool",
    check_review,
    {
        "wait_review": "wait_review",
        "generate_response": "generate_response"
    }
)
workflow.add_edge("wait_review", "generate_response")
workflow.add_edge("generate_response", END)

# 编译并执行
app = workflow.compile()
result = app.invoke({
    "messages": [{"role": "user", "content": "把SKU-A100的价格调到25.99"}],
    "tool_results": [],
    "need_review": False,
    "review_status": "pending"
})

五、人工审核（Human-in-the-Loop）：为高风险操作上保险

并非所有工具调用都应该自动执行。在跨境电商场景中，涉及资金、价格、库存等敏感操作必须经过人工确认。

核心原则：查询类操作自动执行，修改类操作必须审核。

5.1 审核机制的架构设计

Agent准备执行工具
    ↓
检查工具的风险等级
    ↓
┌─ 低风险 → 自动执行 → 返回结果
│
└─ 高风险 → 创建审批任务 → 推送到审核队列
              ↓
         通知审核人（Slack/邮件/看板）
              ↓
         审核人决策：批准 / 拒绝 / 修改参数
              ↓
         执行或取消 → 返回结果

5.2 使用LangGraph实现人工中断

LangGraph原生支持检查点（Checkpointer） 机制，可以在任意节点暂停执行，等待外部恢复：

from langgraph.checkpoint import MemorySaver
from langgraph.graph import StateGraph, END
import json

# 定义审核状态
class ReviewState(TypedDict):
    tool_call: Dict[str, Any]
    status: str  # "pending", "approved", "rejected"
    reviewer_comment: str
    modified_params: Dict[str, Any]

# 创建检查点存储器（生产环境可用Redis或数据库持久化）
memory = MemorySaver()

def high_risk_tool_node(state: ReviewState) -> ReviewState:
    """高风险工具节点：暂停并等待审核"""
    # 工具调用信息保存到状态
    state["status"] = "pending"
    # 这里会触发中断，等待外部恢复
    return state

def approve_node(state: ReviewState) -> ReviewState:
    """审核通过后的执行节点"""
    state["status"] = "approved"
    # 实际执行工具
    print(f"执行工具: {state['tool_call']['name']}")
    print(f"参数: {state['tool_call'].get('modified_params', state['tool_call']['params'])}")
    return state

def reject_node(state: ReviewState) -> ReviewState:
    """审核拒绝节点"""
    state["status"] = "rejected"
    print(f"工具调用被拒绝: {state['tool_call']['name']}")
    print(f"审核意见: {state.get('reviewer_comment', '无')}")
    return state

# 构建审核工作流
review_graph = StateGraph(ReviewState)
review_graph.add_node("high_risk_tool", high_risk_tool_node)
review_graph.add_node("approve", approve_node)
review_graph.add_node("reject", reject_node)

review_graph.set_entry_point("high_risk_tool")
review_graph.add_conditional_edges(
    "high_risk_tool",
    lambda s: s["status"],
    {
        "approved": "approve",
        "rejected": "reject",
    }
)
review_graph.add_edge("approve", END)
review_graph.add_edge("reject", END)

# 编译时启用检查点
app = review_graph.compile(checkpointer=memory)

# 模拟审核流程
def submit_for_review(tool_call: dict):
    """提交工具调用进行审核"""
    config = {"configurable": {"thread_id": "review_001"}}
    
    # 初始状态
    state = {
        "tool_call": tool_call,
        "status": "pending",
        "reviewer_comment": "",
        "modified_params": {}
    }
    
    # 第一次执行会停在high_risk_tool节点（status=pending）
    result = app.invoke(state, config)
    
    # 模拟审核人操作（实际场景中通过API或Dashboard触发）
    # 审核人批准
    result["status"] = "approved"
    result["modified_params"] = {"new_price": 26.99}  # 审核人修改了参数
    
    # 恢复执行
    final = app.invoke(result, config)
    return final

# 测试
tool_call = {
    "name": "adjust_price",
    "params": {"sku": "SKU-A100", "new_price": 25.99, "platform": "amazon"}
}
submit_for_review(tool_call)

5.3 审核通知与集成

在实际生产环境中，审核通常通过以下方式触达审核人：

import requests
import smtplib
from email.mime.text import MIMEText

class ReviewNotifier:
    """审核通知服务"""
    
    def notify_slack(self, tool_call: dict, review_id: str):
        """通过Slack发送审核通知"""
        webhook_url = "https://hooks.slack.com/services/xxx/yyy/zzz"
        message = {
            "text": f"🚨 需要人工审核\n\n"
                    f"工具: {tool_call['name']}\n"
                    f"参数: {json.dumps(tool_call['params'], ensure_ascii=False)}\n"
                    f"审核链接: https://your-platform.com/review/{review_id}"
        }
        requests.post(webhook_url, json=message)
    
    def notify_email(self, tool_call: dict, review_id: str, reviewers: list):
        """通过邮件发送审核通知"""
        # 实际实现中，使用SMTP或邮件服务API
        pass
    
    def create_dashboard_task(self, tool_call: dict, review_id: str):
        """在内部审核看板创建任务"""
        # 实际实现中，调用内部任务系统的API
        pass

六、完整实战：从用户提问到工具执行的全链路

让我们把以上所有技术串起来，还原一个完整的跨境电商场景。

场景：美国站运营人员在系统中输入：“把SKU-A100的价格从29.99降到25.99，然后在亚马逊广告里把对应campaign的日预算从50提到80。”

完整链路：

1. 【用户输入】→ 发送给Agent

2. 【感知层】解析意图
   → 意图: 批量操作（调价 + 调广告预算）
   → 实体: {sku: "SKU-A100", new_price: 25.99, campaign_id: "CAM-123", new_budget: 80}

3. 【ReAct循环】开始
   Thought: 用户要求两个操作，都需要先验证可行性
   Action: query_inventory
   Action Input: {"sku": "SKU-A100"}
   Observation: {"available": 156} ✅ 库存充足

   Thought: 库存充足，可以调价。但调价和调预算都是高风险操作，需要审核。
   Action: adjust_price
   Action Input: {"sku": "SKU-A100", "new_price": 25.99}
   → 触发审核（HITL中断）

4. 【人工审核】
   → Slack通知: "🚨 SKU-A100调价申请，新价格25.99，请审批"
   → 审核人点击批准 → 状态变为approved

5. 【恢复执行】
   → 调价执行成功
   → 继续下一个操作
   Action: adjust_ad_budget
   Action Input: {"campaign_id": "CAM-123", "new_budget": 80}
   → 触发审核

6. 【第二轮审核】
   → 审核人批准 → 预算调整执行成功

7. 【生成回复】
   Final Answer: "已完成SKU-A100的价格调整（29.99 → 25.99），以及CAM-123广告活动的日预算调整（50 → 80）。两项操作均已审核通过并生效。"

七、最佳实践与常见陷阱

7.1 工具描述要“让模型看懂”

工具的description和参数的description是模型决定是否调用、如何调用的唯一依据。写得越清晰，调用越准确：

# ❌ 不好的描述
@tool
def adjust_price(sku: str, price: float) -> str:
    """调整价格"""
    pass

# ✅ 好的描述
@tool
def adjust_price(sku: str, new_price: float, platform: str = "amazon") -> str:
    """
    调整商品在指定电商平台的销售价格。
    
    适用场景：
    - 运营人员主动要求调价
    - 根据竞品价格自动调价（需审核）
    - 促销活动期间的临时调价
    
    注意：此操作涉及收入变动，将自动触发人工审核流程。
    """
    pass

7.2 设置合理的迭代上限

ReAct循环如果没有上限，可能陷入死循环：

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,  # 最多5轮思考-行动
    early_stopping_method="generate",  # 达到上限时直接生成答案
)

7.3 工具执行结果要结构化

返回给模型的结果尽量用JSON格式，方便模型解析和引用：

# ❌ 不好的返回
return "订单已发货，物流单号是1Z999AA10123456784，预计7月22日送达"

# ✅ 好的返回
return json.dumps({
    "order_id": "EB12345678",
    "status": "shipped",
    "tracking_number": "1Z999AA10123456784",
    "eta": "2026-07-22",
    "carrier": "UPS"
})

7.4 敏感操作的参数二次确认

在审核环节，不仅要让审核人选择“批准/拒绝”，还应允许修改参数：

# 审核人界面应提供：
# 1. 批准（使用原参数）
# 2. 批准并修改参数（如把25.99改为26.99）
# 3. 拒绝（附带原因）

八、总结

至此，我们已经完成了AI智能体三大核心能力的建设：

1. 第一天：系统架构设计——感知、决策、执行三层分离
2. 第二天：知识库构建——RAG检索、多业务隔离、多语言支持
3. 第三天：工具调用与任务编排——Function Calling、ReAct模式、人工审核

一个能“想”、能“记”、能“做”的AI智能体，已经具备了上岗的基本条件。

但最后一个问题尚未解决：如何保证AI的输出质量？如何在大模型“胡说八道”时及时止损？

下一篇，我们将深入AI输出的质量评估、异常检测与人工兜底机制——给这个能干的AI智能体装上“刹车”和“保险”，让它既高效又可靠。