AI Yi踩坑记录

from openai import OpenAI

client = OpenAI(
    api_key="sk-xxxxx"
)

response = client.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[
        {"role": "user", "content": "你好"}
    ]
)

from openai import OpenAI

client = OpenAI(
    base_url="https://api.lingyiwanwu.com/v1",
    api_key="your-yi-api-key"
)

response = client.chat.completions.create(
    model="yi-large",
    messages=[
        {"role": "user", "content": "你好"}
    ]
)

def customer_service_query(user_question: str) -> dict:
    client = OpenAI(
        base_url="https://api.lingyiwanwu.com/v1",
        api_key="your-yi-api-key"
    )

    system_prompt = """你是一个智能客服助手，请根据用户的问题提供清晰的回答。

要求：
1. 用简洁明了的中文回答
2. 如果问题复杂或涉及具体业务，建议转人工
3. 提供 1-2 个相关问题供用户参考
4. 保持友好和专业的语气

请以 JSON 格式回复，包含：
- answer: 你的回答
- related_questions: 相关问题列表
- need_human: 是否需要转人工（true/false）
"""

    response = client.chat.completions.create(
        model="yi-large",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_question}
        ],
        temperature=0.3,
        response_format={"type": "json_object"}
    )

    return json.loads(response.choices[0].message.content)

def document_summary_and_qa(document_text: str, question: str = None) -> dict:
    client = OpenAI(
        base_url="https://api.lingyiwanwu.com/v1",
        api_key="your-yi-api-key"
    )

    system_prompt = """你是一个文档分析助手。请分析提供的文档内容，完成以下任务：

1. 生成简洁准确的摘要（200字以内）
2. 提取 3-5 个关键信息点
3. 如果有问题，基于文档内容回答
4. 如果无法从文档中找到答案，明确说明

请以 JSON 格式回复，包含：
- summary: 文档摘要
- key_points: 关键信息点列表
- answer: 问题回答（如果有问题）或 null
- confidence: 回答置信度（1-10）
"""

    user_message = f"文档内容：\n{document_text}"
    if question:
        user_message += f"\n\n问题：{question}"

    response = client.chat.completions.create(
        model="yi-large",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_message}
        ],
        temperature=0.2,
        response_format={"type": "json_object"}
    )

    return json.loads(response.choices[0].message.content)

def code_assistant(code_snippet: str, problem_description: str) -> dict:
    client = OpenAI(
        base_url="https://api.lingyiwanwu.com/v1",
        api_key="your-yi-api-key"
    )

    system_prompt = """你是一个代码辅助专家。请分析提供的代码和问题，给出专业的建议。

要求：
1. 准确理解代码逻辑和问题所在
2. 提供可执行的解决方案
3. 解释问题原因和解决思路
4. 如果代码有优化空间，也提供建议

请以 JSON 格式回复，包含：
- diagnosis: 问题诊断
- solution: 解决方案（包含代码示例）
- explanation: 详细解释
- optimization: 优化建议（如果有）
"""

    response = client.chat.completions.create(
        model="yi-large",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"代码：\n{code_snippet}\n\n问题：{problem_description}"}
        ],
        temperature=0.3,
        response_format={"type": "json_object"}
    )

    return json.loads(response.choices[0].message.content)

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def safe_yi_call(client, model, messages, **kwargs):
    return client.chat.completions.create(
        model=model,
        messages=messages,
        timeout=30,
        **kwargs
    )

def chunk_document(document_text: str, max_chunk_size: int = 4000) -> list:
    """将文档分块，避免单次处理过长"""
    # 简单按段落分块，实际可以更智能
    paragraphs = document_text.split('\n\n')
    chunks = []
    current_chunk = ""
    current_size = 0

    for para in paragraphs:
        para_size = len(para)
        if current_size + para_size > max_chunk_size:
            chunks.append(current_chunk)
            current_chunk = para
            current_size = para_size
        else:
            current_chunk += "\n\n" + para if current_chunk else para
            current_size += para_size

    if current_chunk:
        chunks.append(current_chunk)

    return chunks

def robust_json_parse(response_text: str) -> dict:
    """健壮的 JSON 解析，带重试和容错"""
    try:
        return json.loads(response_text)
    except json.JSONDecodeError:
        # 尝试提取 JSON 部分
        try:
            json_start = response_text.find('{')
            json_end = response_text.rfind('}') + 1
            if json_start != -1 and json_end > json_start:
                return json.loads(response_text[json_start:json_end])
        except:
            pass

        # 失败则返回默认结构
        return {
            "error": "JSON 解析失败",
            "raw_response": response_text
        }