4sapicom生成智能聊天机器人商业价值

import openai
import logging
from typing import Optional, Dict, List
from dataclasses import dataclass
from configparser import ConfigParser
from tenacity import retry, stop_after_attempt, wait_random_exponential
from datetime import datetime

# 日志模块化配置：支持分级监控与问题追溯
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s",
    handlers=[
        logging.FileHandler(f"chatbot_log_{datetime.now().strftime('%Y%m%d')}.log", encoding="utf-8"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("EnterpriseChatbot")

# 配置结构化管理：支持多环境动态切换
@dataclass
class BotConfig:
    base_url: str
    api_key: str
    timeout: int
    default_model: str
    max_tokens: int
    temperature: float
    retry_attempts: int

def load_config(config_path: str = "bot_config.ini") -> BotConfig:
    config = ConfigParser()
    config.read(config_path, encoding="utf-8")
    return BotConfig(
        base_url=config.get("API", "BASE_URL", fallback="https://4sapi.com"),
        api_key=config.get("API", "API_KEY"),
        timeout=config.getint("API", "TIMEOUT", fallback=30),
        default_model=config.get("MODEL", "DEFAULT_MODEL", fallback="davinci"),
        max_tokens=config.getint("MODEL", "MAX_TOKENS", fallback=150),
        temperature=config.getfloat("MODEL", "TEMPERATURE", fallback=0.7),
        retry_attempts=config.getint("API", "RETRY_ATTEMPTS", fallback=3)
    )

# 客户端初始化：整合高可用与安全性设计
config = load_config()
client = openai.OpenAI(
    base_url=config.base_url,
    api_key=config.api_key,
    timeout=config.timeout
)

# 核心对话函数：支持上下文管理与智能容错
@retry(
    stop=stop_after_attempt(config.retry_attempts),
    wait=wait_random_exponential(min=1, max=10),
    reraise=True
)
def intelligent_chat(
    user_input: str,
    history: Optional[List[Dict[str, str]]] = None,
    model: Optional[str] = None,
    max_tokens: Optional[int] = None,
    temperature: Optional[float] = None
) -> str:
    """
    智能对话核心函数：支持多轮上下文衔接、动态参数配置与异常重试
    :param user_input: 当前用户输入
    :param history: 对话历史记录，格式[{"user":"xxx","assistant":"xxx"}]
    :param model: 模型名称（覆盖默认配置）
    :param max_tokens: 响应长度限制（覆盖默认配置）
    :param temperature: 生成随机性（覆盖默认配置）
    :return: 结构化响应文本
    """
    # 上下文拼接：保障多轮对话连贯性
    prompt = ""
    if history:
        for msg in history:
            prompt += f"用户：{msg['user']}\n助手：{msg['assistant']}\n"
    prompt += f"用户：{user_input}\n助手："

    # 动态参数配置：适配不同场景需求
    model = model or config.default_model
    max_tokens = max_tokens or config.max_tokens
    temperature = temperature or config.temperature

    try:
        response = client.Completion.create(
            engine=model,
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=temperature,
            stop=["用户："]  # 精准截断，避免无效输出
        )
        result = response.choices[0].text.strip()
        logger.info(f"对话成功 - 模型：{model} | 用户输入：{user_input} | 响应：{result}")
        return result
    except Exception as e:
        logger.error(f"对话异常 - 模型：{model} | 用户输入：{user_input} | 错误信息：{str(e)}", exc_info=True)
        raise

# 行业场景适配示例
if __name__ == "__main__":
    # 场景1：金融客服多轮咨询
    finance_history = [
        {"user": "我想办理个人经营性贷款，需要准备哪些材料？", "assistant": "个人经营性贷款需准备身份证、营业执照、近6个月银行流水、经营场所证明等材料"}
    ]
    finance_input = "如果流水不足6个月，有其他替代材料吗？"
    finance_response = intelligent_chat(finance_input, finance_history, temperature=0.4)
    print(f"金融智能客服：{finance_response}")

    # 场景2：教育机构课程推荐
    education_input = "高二学生想提升数学成绩，有适合的线上课程吗？"
    education_response = intelligent_chat(education_input, model="gpt-3.5-turbo", max_tokens=200)
    print(f"教育智能顾问：{education_response}")

    # 场景3：工业设备维护咨询
    industrial_history = [
        {"user": "生产线的机械臂出现定位偏差，可能是什么原因？", "assistant": "机械臂定位偏差可能源于传感器故障、伺服电机异常、程序参数漂移等原因"}
    ]
    industrial_input = "如何快速排查传感器故障？"
    industrial_response = intelligent_chat(industrial_input, industrial_history, temperature=0.3)
    print(f"工业智能助手：{industrial_response}")