76 篇博文 含有标签「LLM」

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

word = UnstructuredLoader("简历.docx")
docs = word.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=50, chunk_overlap=20)
chunks = splitter.split_documents(docs)

embeddings = DashScopeEmbeddings(
    dashscope_api_key=os.environ.get("dashscope"),
    model="text-embedding-v1",
)
vectordb = QdrantVectorStore.from_documents(
    chunks, embeddings,
    collection_name="resume",
    url="http://localhost:6333",
)

chain = {
    "context": vec_store.as_retriever() | format_docs,
    "question": RunnablePassthrough(),
} | prompt | llm | StrOutputParser()

chain.invoke("请输入姓名. 格式如下\n姓名: ?")
chain.invoke("总结专业技能情况. 格式如下\n专业技能: ?")
chain.invoke("根据工作年份总结工作经验. 格式如下\n工作经验: ?年")

from langsmith import Client as LangSmithClient
client = LangSmithClient()
prompt = client.pull_prompt("rlm/rag-prompt", dangerously_pull_public_prompt=True)

import os

from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_unstructured import UnstructuredLoader
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langsmith import Client as LangSmithClient

from llm import DeepSeek, QdrantVecStoreFromDocs


def clearstr(s: str) -> str:
    filter_chars = ['\n', '\r', '\t', '\u3000', ' ']
    for char in filter_chars:
        s = s.replace(char, '')
    return s


def format_docs(docs) -> str:
    return "\n\n".join(clearstr(doc.page_content) for doc in docs)


def load_doc():
    word = UnstructuredLoader("C:\\Users\\Administrator\\Downloads\\xxx.docx")
    docs = word.load()
    splitter = RecursiveCharacterTextSplitter(chunk_size=50, chunk_overlap=20)
    chunks = splitter.split_documents(docs)
    vec_store = QdrantVecStoreFromDocs(chunks, collection_name="resume")
    llm = DeepSeek()
    client = LangSmithClient()
    prompt = client.pull_prompt("rlm/rag-prompt", dangerously_pull_public_prompt=True)

    chain = {
        "context": vec_store.as_retriever() | format_docs,
        "question": RunnablePassthrough(),
    } | prompt | llm | StrOutputParser()

    ret = chain.invoke("请输入姓名. 格式如下\n姓名: ?")
    print(ret)
    ret = chain.invoke("总结专业技能情况,内容可能包含golang、AI Agent、python、rag等.格式如下\n专业技能: ?")
    print(ret)
    ret = chain.invoke("根据各大公司工作过的年份总结工作经验有多少年.格式如下\n工作经验: ?年")
    print(ret)


def main():
    print("Hello from rag!")
    load_doc()


if __name__ == "__main__":
    main()

import os
from typing import List, Optional

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.documents import Document
from langchain_openai import ChatOpenAI
from langchain_qdrant import QdrantVectorStore


def DeepSeek():
    return ChatOpenAI(
        model="deepseek-v4-flash",
        api_key=os.environ.get("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    )


def get_embeddings():
    return DashScopeEmbeddings(
        dashscope_api_key=os.environ.get("dashscope"),
        model="text-embedding-v1",
    )


def QdrantVecStoreFromDocs(docs: List[Document], collection_name: str,
                           qdrant_url: Optional[str] = None,
                           qdrant_api_key: Optional[str] = None):
    if qdrant_url is None:
        qdrant_url = "http://localhost:6333"
    embeddings = get_embeddings()
    vectordb = QdrantVectorStore.from_documents(
        docs, embeddings, collection_name=collection_name,
        url=qdrant_url, api_key=qdrant_api_key,
    )
    return vectordb

RAGPrompt = """
You are an assistant for question-answering tasks.
Use the following pieces of retrieved context to answer the question.
If you don't know the answer, just say that you don't know,
don't try to make up an answer.
Use three sentences maximum and keep the answer concise.
Question: {question}
Context: {context}
Answer:
"""