85 篇博文 含有标签「AI」

while True:
    thought = llm.generate(f"当前状态：{state}，下一步做什么？")
    action = parse_action(thought)       # 执行工具调用
    result = execute_tool(action)
    state += f"\n执行结果：{result}"

system_prompt = """
你是一个专业的电商客服助手，负责帮助用户查询订单、处理退换货、查询库存。

你的职责：
- 准确理解用户意图，必要时调用相应函数获取真实数据
- 不要编造订单号、物流信息等事实性数据
- 对于无法满足的需求，明确说明原因并提供替代方案

安全规则：
- 涉及订单修改、退款等操作时，必须先确认用户身份
- 不要直接调用需要高权限的函数，先请求用户确认

响应风格：
- 用中文回答，语气友好专业
- 对于数据类查询，用表格形式呈现
- 对于操作类任务，列出步骤并询问是否确认执行
"""

# 形态 1：直接明确
{"role": "user", "content": "帮我查订单 20241201001 的物流状态"}

# 形态 2：模糊但可推导
{"role": "user", "content": "看看我昨天买的手机发货了吗"}
  # → 需要从历史上下文推导订单号

# 形态 3：多参数隐含
{"role": "user", "content": "我要从北京飞上海，明天早上出发"}
  # → destination="上海", origin="北京", date=明天日期
  # → time_range="morning"（隐含约束）

{
  "role": "assistant",
  "content": "好的，我帮您查询订单 20241201001 的物流状态。"
}

{
  "role": "assistant",
  "content": None,  # 注意：tool_calls 模式下 content 为 None
  "tool_calls": [
    {
      "id": "call_abc123",  # 调用 ID，用于后续关联结果
      "type": "function",
      "function": {
        "name": "get_order_status",
        "arguments": '{"order_id": "20241201001"}'
      }
    }
  ]
}

{
  "role": "tool",
  "tool_call_id": "call_abc123",  # 必须匹配 Assistant 的 tool_calls[].id
  "content": '{"status": "shipped", "tracking_number": "SF1234567890", "estimated_delivery": "2024-12-05"}'
}

# ❌ 错误示例
{
  "role": "tool",
  "tool_call_id": "call_abc123",
  "content": "订单不存在"  # 太模糊，模型不知道为什么
}

# ✅ 正确示例
{
  "role": "tool",
  "tool_call_id": "call_abc123",
  "content": '{"error": "ORDER_NOT_FOUND", "message": "订单号 20241201001 不存在或已被删除", "suggestions": ["检查订单号是否输入正确", "联系客服确认"]}'
}

# Assistant 调用了两个函数：get_order_status 和 get_user_profile
# 你按顺序返回两个 Tool 消息

messages.append({
  "role": "tool",
  "tool_call_id": "call_abc123",
  "content": '{"order_id": "20241201001", "status": "shipped", ...}'
})

messages.append({
  "role": "tool",
  "tool_call_id": "call_def456",
  "content": '{"user_id": "u12345", "name": "张三", "level": "VIP", ...}'
})

function_def = {
  "name": "search_products",
  "description": "根据关键词、分类、价格范围搜索商品。支持模糊匹配和排序。",
  "strict": False,
  "parameters": {
    "type": "object",
    "properties": {
      # 详见下文
    },
    "additionalProperties": False
  }
}

{
  "search_type": {
    "type": "string",
    "description": "搜索方式：按订单号、运单号或手机号",
    "enum": ["order_id", "tracking_number", "phone"]
  }
}

{
  "price_min": {
    "type": "number",
    "description": "价格下限（元），不填表示无下限",
    "minimum": 0,
    "exclusiveMaximum": 1000000
  }
}

{
  "category_ids": {
    "type": "array",
    "description": "商品分类 ID 列表，最多 10 个",
    "items": {
      "type": "integer",
      "description": "分类 ID"
    },
    "minItems": 1,
    "maxItems": 10,
    "uniqueItems": true
  }
}

{
  "address": {
    "type": "object",
    "description": "收货地址",
    "properties": {
      "province": {"type": "string", "description": "省份"},
      "city": {"type": "string", "description": "城市"},
      "district": {"type": "string", "description": "区县"},
      "detail": {"type": "string", "description": "详细地址"}
    },
    "required": ["province", "city", "detail"]
  }
}

# examples 的强大之处
{
  "filter": {
    "type": "object",
    "description": "搜索过滤条件",
    "properties": {
      "brand": {"type": "string"},
      "price_range": {
        "type": "object",
        "properties": {
          "min": {"type": "number"},
          "max": {"type": "number"}
        }
      },
      "in_stock": {"type": "boolean"}
    },
    "examples": [
      {"brand": "Apple", "price_range": {"min": 5000, "max": 10000}},
      {"in_stock": true}
    ]
  }
}

functions = [
  {
    "name": "search_products",
    "description": "搜索商品，支持关键词、分类、价格、品牌等多维度筛选，返回商品列表和分页信息。",
    "strict": False,
    "parameters": {
      "type": "object",
      "properties": {
        "keyword": {
          "type": "string",
          "description": "搜索关键词，支持商品名称、型号的模糊匹配。为空则返回所有符合条件的商品。"
        },
        "category_id": {
          "type": "integer",
          "description": "商品分类 ID，1-手机数码，2-家用电器，3-服装鞋帽，4-食品生鲜，5-图书文具"
        },
        "price_range": {
          "type": "object",
          "description": "价格范围约束",
          "properties": {
            "min": {
              "type": "number",
              "description": "最低价格（元），必须 ≥ 0",
              "minimum": 0
            },
            "max": {
              "type": "number",
              "description": "最高价格（元）",
              "minimum": 0
            }
          },
          "examples": [
            {"min": 0, "max": 100},
            {"min": 1000, "max": 5000}
          ]
        },
        "brand": {
          "type": "string",
          "description": "品牌名称，精确匹配。如：Apple、华为、小米"
        },
        "in_stock": {
          "type": "boolean",
          "description": "是否只显示有货商品。true-仅显示有货，false-显示全部（包含无货）"
        },
        "sort_by": {
          "type": "string",
          "description": "排序方式",
          "enum": ["relevance", "price_asc", "price_desc", "sales_desc", "rating_desc"]
        },
        "page": {
          "type": "integer",
          "description": "页码，从 1 开始",
          "minimum": 1,
          "default": 1
        },
        "page_size": {
          "type": "integer",
          "description": "每页数量，建议 10-50",
          "minimum": 1,
          "maximum": 100,
          "default": 20
        }
      },
      "required": [],
      "additionalProperties": False
    }
  }
]

# LLM 返回
{
  "id": "chatcmpl-abc123",
  "object": "chat.completion",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "好的，请问您要搜索哪个品牌的手机？预算大概是多少？"
      },
      "finish_reason": "stop"
    }
  ]
}

def handle_text_response(response):
    assistant_msg = response["choices"][0]["message"]
    content = assistant_msg["content"]
    
    # 直接返回给用户
    return content

# LLM 返回
{
  "id": "chatcmpl-abc123",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": None,
        "tool_calls": [
          {
            "id": "call_xxx123",
            "type": "function",
            "function": {
              "name": "search_products",
              "arguments": '{"keyword": "iPhone", "price_range": {"min": 5000, "max": 10000}}'
            }
          }
        ]
      },
      "finish_reason": "tool_calls"
    }
  ]
}

import json

def handle_function_call(response, functions_registry):
    assistant_msg = response["choices"][0]["message"]
    tool_calls = assistant_msg["tool_calls"]
    
    # 保存 Assistant 消息到对话历史
    messages.append(assistant_msg)
    
    # 执行每个函数调用
    tool_responses = []
    for tool_call in tool_calls:
        call_id = tool_call["id"]
        function_name = tool_call["function"]["name"]
        arguments_str = tool_call["function"]["arguments"]
        
        # 1. 解析参数（可能失败）
        try:
            arguments = json.loads(arguments_str)
        except json.JSONDecodeError as e:
            # 参数解析失败，返回错误信息给模型
            tool_responses.append({
                "role": "tool",
                "tool_call_id": call_id,
                "content": json.dumps({
                  "error": "INVALID_ARGUMENTS_JSON",
                  "message": f"参数 JSON 解析失败: {str(e)}",
                  "raw_arguments": arguments_str
                })
            })
            continue
        
        # 2. 执行函数（可能失败）
        try:
            function_impl = functions_registry[function_name]
            result = function_impl(**arguments)
            # 将结果转为 JSON 字符串
            result_str = json.dumps(result) if not isinstance(result, str) else result
        except KeyError:
            # 函数不存在
            tool_responses.append({
                "role": "tool",
                "tool_call_id": call_id,
                "content": json.dumps({
                  "error": "FUNCTION_NOT_FOUND",
                  "message": f"函数 {function_name} 不存在",
                  "available_functions": list(functions_registry.keys())
                })
            })
            continue
        except Exception as e:
            # 函数执行失败
            tool_responses.append({
                "role": "tool",
                "tool_call_id": call_id,
                "content": json.dumps({
                  "error": "FUNCTION_EXECUTION_ERROR",
                  "message": str(e),
                  "function": function_name,
                  "arguments": arguments
                })
            })
            continue
        
        # 3. 成功执行，返回结果
        tool_responses.append({
            "role": "tool",
            "tool_call_id": call_id,
            "content": result_str
        })
    
    # 4. 将 Tool 结果加入对话历史，继续下一轮
    messages.extend(tool_responses)
    return messages  # 传回给 LLM API 继续对话

# LLM 返回
{
  "choices": [
    {
      "message": {
        "role": "assistant",
        "content": None,
        "tool_calls": [
          {
            "id": "call_aaa",
            "function": {"name": "get_order_status", "arguments": '{"order_id": "123"}'}
          },
          {
            "id": "call_bbb",
            "function": {"name": "get_account_balance", "arguments": '{"user_id": "user_456"}'}
          }
        ]
      },
      "finish_reason": "tool_calls"
    }
  ]
}

def handle_multiple_function_calls(response, functions_registry):
    assistant_msg = response["choices"][0]["message"]
    tool_calls = assistant_msg["tool_calls"]
    
    # 保存 Assistant 消息
    messages.append(assistant_msg)
    
    # 并行执行所有函数（如果函数之间无依赖）
    # 如果有依赖关系，按顺序执行
    tool_responses = []
    
    for tool_call in tool_calls:
        call_id = tool_call["id"]
        function_name = tool_call["function"]["name"]
        arguments = json.loads(tool_call["function"]["arguments"])
        
        try:
            result = functions_registry[function_name](**arguments)
            result_str = json.dumps(result) if not isinstance(result, str) else result
        except Exception as e:
            result_str = json.dumps({"error": str(e)})
        
        tool_responses.append({
            "role": "tool",
            "tool_call_id": call_id,
            "content": result_str
        })
    
    # 按 tool_call_id 排序（保持与 tool_calls 顺序一致）
    tool_responses.sort(key=lambda x: x["tool_call_id"])
    
    messages.extend(tool_responses)
    return messages

def chat_with_functions(user_message, conversation_history, functions_registry):
    # 1. 构造当前对话上下文
    messages = conversation_history + [{"role": "user", "content": user_message}]
    
    # 2. 调用 LLM API
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=[{"type": "function", "function": func_def} for func_def in functions_registry.values()],
        tool_choice="auto"  # 让模型自动决定是否调用函数
    )
    
    assistant_msg = response.choices[0].message
    
    # 3. 判断响应类型
    if assistant_msg.tool_calls:
        # === 函数调用场景 ===
        conversation_history.append(assistant_msg)  # 保存 Assistant
        
        tool_responses = []
        for tool_call in assistant_msg.tool_calls:
            # 执行函数并收集结果
            result = execute_function(tool_call, functions_registry)
            tool_responses.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })
        
        conversation_history.extend(tool_responses)
        
        # 递归调用，继续对话
        return chat_with_functions(None, conversation_history, functions_registry)
    
    else:
        # === 纯文本响应场景 ===
        conversation_history.append(assistant_msg)
        return assistant_msg.content  # 返回给用户

def execute_function(tool_call, functions_registry):
    func_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)
    
    func = functions_registry[func_name]
    return func(**arguments)

def validate_and_cast(value, schema):
    if schema["type"] == "integer" and isinstance(value, str):
        try:
            return int(value)
        except ValueError:
            pass
    # 其他类型转换...
    return value

arguments = {k: validate_and_cast(v, param_schema) for k, v in arguments.items()}

# 在错误信息中给出明确建议
{
  "error": "INVALID_ORDER_ID",
  "message": "订单号格式错误，应为 10 位数字",
  "user_input": order_id,
  "suggestions": [
    "请检查订单号是否正确",
    "订单号示例：20241201001"
  ]
}

# 在 System Prompt 中增加：
# "如果同一个函数连续失败 3 次，停止调用并告知用户无法完成"

# ❌ 返回完整数据
{
  "products": [/* 100 个商品 */]
}

# ✅ 返回摘要
{
  "total": 100,
  "page": 1,
  "products": [/* 前 3 个 */],
  "note": "共 100 个商品，仅展示前 3 个，如需更多请翻页"
}

def trim_conversation(messages, max_tokens=4000):
    # 保留 System
    # 保留最近 5 轮对话
    # 保留所有 Tool 结果（因为可能被引用）
    # 删除早期的 Assistant / User 对话
    pass