10 倍降本、100K 长文秒回、128K 上下文托管——DeepSeek-R1 登陆 Bedrock，云上大模型进入血战模式

import boto3
from botocore.exceptions import ClientError, ConnectTimeoutError
import json
from datetime import datetime
import re
def init_translation_client(aws_region="us-east-1", timeout=10):
    """
    初始化技术文档翻译专用 Bedrock 客户端
    :param aws_region: AWS 区域，默认 us-east-1
    :param timeout: 超时时间（秒），适配实时翻译场景
    :return: 初始化后的客户端对象
    """
    try:
        trans_client = boto3.client(
            service_name="bedrock",
            region_name=aws_region,
            config=boto3.session.Config(
                connect_timeout=timeout,
                read_timeout=timeout
            )
        )
        print(f"技术文档翻译客户端已在 {aws_region} 区域初始化")
        return trans_client
    except ConnectTimeoutError:
        raise Exception(f"连接超时：无法初始化翻译客户端，请检查网络（超时设置：{timeout}秒）")
    except ClientError as e:
        raise Exception(f"客户端初始化失败：{e.response['Error']['Message']}")
def translate_tech_document(
    doc_content,
    src_lang,
    tgt_lang,
    term_dict=None,
    preserve_format=True
):
    """
    翻译技术文档（支持中英互译，保留格式与专业术语）
    :param doc_content: 待翻译的技术文档内容
    :param src_lang: 源语言（如 'en' 英文，'zh' 中文）
    :param tgt_lang: 目标语言（如 'zh' 中文，'en' 英文）
    :param term_dict: 自定义术语库（字典格式，如 {"API": "应用程序接口"}），默认空
    :param preserve_format: 是否保留原文格式（代码块、列表等），默认 True
    :return: 翻译后的文档内容（字符串）
    """
    # 处理自定义术语库，生成术语提示
    term_prompt = ""
    if term_dict:
        term_list = [f"'{k}': '{v}'" for k, v in term_dict.items()]
        term_prompt = f"请严格遵循以下术语翻译规则：{', '.join(term_list)}。"

    # 处理格式保留提示
    format_prompt = "请保留原文格式，包括代码块（用```标记）、列表（用数字/符号标记）、标题层级。" if preserve_format else ""

    # 构建技术文档翻译专用提示词
    trans_prompt = f"""
    请将以下技术文档从{src_lang}翻译为{tgt_lang}：
    {term_prompt}
    {format_prompt}
    翻译要求：
    1. 准确传递技术细节，专业术语翻译一致（如遇未指定术语，保留原文并加注）；
    2. 译文符合目标语言表达习惯，避免直译导致的晦涩；
    3. 若包含代码片段，仅翻译注释，不修改代码逻辑。
    文档内容：{doc_content}
    """

    try:
        # 调用 DeepSeek-R1 模型执行翻译
        client = init_translation_client()
        response = client.invoke_model(
            body=json.dumps({"prompt": trans_prompt}),
            modelId="DeepSeek-R1",
            parameters={
                "task": "technical_translation",
                "source_language": src_lang,
                "target_language": tgt_lang,
                "temperature": 0.1  # 低温度确保术语翻译一致性
            }
        )
        # 解析并验证翻译结果
        translated_content = response["body"].read().decode("utf-8")
        if not translated_content.strip():
            raise ValueError("翻译结果为空，请检查文档内容或模型配置")
        return translated_content
    except ClientError as e:
        raise Exception(f"模型调用失败：{e.response['Error']['Message']}")
    except json.JSONDecodeError:
        raise Exception("响应格式错误：无法解析模型返回的 JSON 数据")
# ------------------- 示例：翻译技术文档（英→中） -------------------
if __name__ == "__main__":
    # 1. 待翻译的英文技术文档（含代码片段与专业术语）
    sample_tech_doc = """
    # API Documentation for User Authentication
    ## 1. Overview
    This API is used to handle user login and token generation. It supports two authentication methods:
    - JWT (JSON Web Token)
    - OAuth 2.0

    ## 2. Request Example
    ```python
    import requests
    response = requests.post(
        url="https://api.example.com/auth/login",
        data={"username": "user123", "password": "***"}
    )
    print(response.json())  # Returns token and expiration time

import boto3
from botocore.exceptions import ClientError, ConnectTimeoutError
import json
from datetime import datetime
def init_finance_bedrock_client(aws_region="us-east-1", timeout=30):
    """
    初始化金融分析专用Bedrock客户端
    :param aws_region: AWS区域，默认us-east-1
    :param timeout: 超时时间（秒），适配财报长文本分析
    :return: 初始化后的客户端对象
    """
    try:
        finance_client = boto3.client(
            service_name="bedrock",
            region_name=aws_region,
            config=boto3.session.Config(
                connect_timeout=timeout,
                read_timeout=timeout
            )
        )
        print(f"金融分析Bedrock客户端已在{aws_region}区域初始化")
        return finance_client
    except ConnectTimeoutError:
        raise Exception(f"连接超时：无法初始化金融分析客户端，请检查网络或延长超时设置（当前{timeout}秒）")
    except ClientError as e:
        raise Exception(f"客户端初始化失败：{e.response['Error']['Message']}")
def extract_finance_insights(report_content, focus_metrics=None, forecast_quarters=3):
    """
    提取财报核心数据并生成趋势预测
    :param report_content: 财报文本内容
    :param focus_metrics: 重点关注指标列表（如["营收","净利润","毛利率"]），默认全量分析
    :param forecast_quarters: 预测季度数，默认3季度
    :return: 结构化分析结果（字符串）
    """
    # 处理重点指标参数，默认分析全量核心指标
    if not focus_metrics:
        focus_metrics = ["营收", "净利润", "毛利率", "资产负债率", "经营现金流", "研发投入"]

    # 构建金融场景专用提示词，明确分析要求
    finance_prompt = f"""
    请对以下财报文本进行深度分析，输出需满足：
    1. 核心数据提取：从{report_content}中提取{focus_metrics}等指标，需包含具体数值、单位及报告期
    2. 对比分析：计算各指标同比（与去年同期）、环比（与上一季度）变化率，标注增减趋势
    3. 趋势预测：基于近3年数据规律，预测未来{forecast_quarters}季度的关键指标走势，附预测依据
    4. 风险提示：识别潜在风险点（如现金流缺口、毛利率异常波动），按风险等级排序
    5. 输出格式：
       - 数据摘要：用表格形式呈现核心指标及对比数据
       - 趋势分析：文字描述+图表特征说明（如"营收呈季度性增长，Q4预计突破XX亿元"）
       - 风险清单：按"高/中/低"风险等级标注，附应对建议
    """

    try:
        # 调用DeepSeek-R1模型进行金融分析
        response = init_finance_bedrock_client().invoke_model(
            body=json.dumps({"prompt": finance_prompt}),
            modelId="DeepSeek-R1",
            parameters={
                "task": "financial_report_analysis",
                "response_format": "structured",
                "temperature": 0.1  # 极低温度确保数据准确性
            }
        )
        # 解析并验证响应内容
        analysis_content = response["body"].read().decode("utf-8")
        if not analysis_content.strip():
            raise ValueError("分析结果为空，请检查财报文本完整性或模型配置")
        return analysis_content
    except ClientError as e:
        raise Exception(f"模型调用失败：{e.response['Error']['Message']}")
    except json.JSONDecodeError:
        raise Exception("响应格式错误：无法解析模型返回的JSON数据")
# ------------------- 示例：分析某企业季度财报 -------------------
if __name__ == "__main__":
    # 1. 模拟财报文本（实际场景可从PDF/Word解析获取）
    sample_report = """
    2024年Q2季度财报：本季度营收125.6亿元，较去年同期增长18.3%，较上一季度增长5.2%；
    净利润28.9亿元，同比增长22.1%，环比下降1.7%；毛利率维持在42.5%，与上季度持平；
    资产负债率58.3%，较Q1下降2.1个百分点；经营现金流净额32.4亿元，同比增长15.6%。
    研发投入8.7亿元，占营收比重6.9%，较去年同期提升0.8个百分点。
    """

    # 2. 定义分析参数（重点关注营收、净利润及风险点）
    key_metrics = ["营收", "净利润", "毛利率", "资产负债率"]
    predict_quarters = 3  # 预测未来3个季度

    # 3. 执行财报分析
    try:
        finance_analysis = extract_finance_insights(
            report_content=sample_report,
            focus_metrics=key_metrics,
            forecast_quarters=predict_quarters
        )

        # 4. 输出并保存分析结果（按金融规范命名）
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        result_filename = f"Financial_Report_Analysis_{timestamp}.txt"

        print("="*60)
        print(f"财报分析结果（重点指标：{key_metrics} | 预测季度：{predict_quarters}）")
        print("="*60)
        print(finance_analysis)

        # 保存结果用于归档或进一步分析
        with open(result_filename, "w", encoding="utf-8") as f:
            f.write(f"分析时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write(f"重点指标：{key_metrics}\n")
            f.write(f"预测季度数：{predict_quarters}\n")
            f.write("="*40 + "\n")
            f.write(finance_analysis)

        print(f"\n分析结果已保存至：{result_filename}")
    except Exception as e:
        print(f"财报分析失败：{str(e)}")

import boto3
from botocore.exceptions import ClientError, ConnectTimeoutError
import json  # 新增JSON处理，便于结构化输出解析
def init_bedrock_client(region="us-east-1"):
    """
    初始化AWS Bedrock客户端
    :param region: AWS区域，默认us-east-1
    :return: 初始化后的Bedrock客户端对象
    """
    try:
        bedrock_client = boto3.client(
            service_name="bedrock",
            region_name=region,
            # 可选：添加超时配置，避免长期阻塞
            config=boto3.session.Config(connect_timeout=10, read_timeout=30)
        )
        print(f"Bedrock客户端在{region}区域初始化成功")
        return bedrock_client
    except ConnectTimeoutError:
        raise Exception(f"连接超时：无法在{region}区域初始化Bedrock客户端，请检查网络或区域配置")
    except ClientError as e:
        raise Exception(f"客户端初始化失败：{e.response['Error']['Message']}")
def generate_std_based_exercise_set(
    subject, 
    grade, 
    curriculum_std, 
    bedrock_client, 
    model_id="DeepSeek-R1"
):
    """
    基于课程标准生成带解题步骤的习题集
    :param subject: 学科（如"Mathematics"、"Chinese"）
    :param grade: 年级（如"Grade 5"、"Grade 3"）
    :param curriculum_std: 课程标准版本（如"人教版小学数学2022版"、"北师大版数学五年级上册"）
    :param bedrock_client: 已初始化的Bedrock客户端
    :param model_id: 模型ID，默认DeepSeek-R1
    :return: 结构化习题集（字符串格式，包含题目、步骤、答案）
    """
    # 构建精细化的模型请求prompt，明确习题结构与解题步骤要求
    prompt = f"""
    请依据{curriculum_std}课程标准，为{grade}的{subject}学科生成习题集，需满足以下要求：
    1. 题型分布：基础题5道（考查核心概念）+ 提高题3道（综合应用）
    2. 每道题需包含：题干 + 详细解题步骤（分点说明，适合学生理解） + 最终答案
    3. 内容范围：围绕{grade} {subject}的核心知识点（如五年级数学的"小数乘法"、"图形的面积计算"）
    4. 格式要求：用标题区分"基础题"和"提高题"，每道题标注题号，解题步骤用数字序号列出
    """
    try:
        # 调用DeepSeek-R1模型，指定教育内容生成任务
        model_response = bedrock_client.invoke_model(
            body=json.dumps({"prompt": prompt}),  # 改用JSON格式传递prompt，提升兼容性
            modelId=model_id,
            parameters={
                "task": "educational_content_generation",
                "response_format": "text",  # 指定文本输出格式
                "temperature": 0.3  # 低温度，确保内容准确性（教育场景需严谨）
            }
        )
        # 解析响应内容，处理编码问题
        exercise_content = model_response["body"].read().decode("utf-8")
        # 验证响应是否有效（避免空内容）
        if not exercise_content.strip():
            raise ValueError("模型返回空内容，请检查prompt或模型配置")
        return exercise_content
    except ClientError as e:
        raise Exception(f"模型调用失败：{e.response['Error']['Message']}")
    except json.JSONDecodeError:
        raise Exception("响应格式错误：无法解析模型返回的JSON数据")
# ------------------- 示例：生成五年级人教版数学习题集 -------------------
if __name__ == "__main__":
    # 1. 初始化Bedrock客户端
    bedrock_client = init_bedrock_client(region="us-east-1")
    # 2. 定义习题生成参数（贴合实际教学需求）
    target_subject = "Mathematics"  # 目标学科：数学
    target_grade = "Grade 5"        # 目标年级：五年级
    target_curriculum = "人教版小学数学2022版"  # 目标课程标准
    # 3. 生成习题集
    try:
        math_exercise_set = generate_std_based_exercise_set(
            subject=target_subject,
            grade=target_grade,
            curriculum_std=target_curriculum,
            bedrock_client=bedrock_client
        )
        # 4. 输出结果（可扩展为保存到文件或对接教学平台）
        print("="*50)
        print(f"{target_grade} {target_subject} 习题集（{target_curriculum}）")
        print("="*50)
        print(math_exercise_set)
        # 可选：将习题集保存为txt文件，便于教学使用
        with open(f"{target_grade}_{target_subject}_exercises.txt", "w", encoding="utf-8") as f:
            f.write(math_exercise_set)
        print(f"\n习题集已保存至：{target_grade}_{target_subject}_exercises.txt")
    except Exception as e:
        print(f"习题集生成失败：{str(e)}")

import boto3
from botocore.exceptions import ClientError, ConnectTimeoutError
import json
from datetime import datetime
def init_bedrock_research_client(aws_region="us-east-1", conn_timeout=15, read_timeout=40):
    """
    初始化科研场景专用 Bedrock 客户端
    :param aws_region: AWS 区域，默认 us-east-1
    :param conn_timeout: 连接超时时间（秒），默认 15
    :param read_timeout: 读取超时时间（秒），默认 40（科研内容生成需更长时间）
    :return: 初始化后的 Bedrock 客户端对象
    """
    try:
        research_bedrock_client = boto3.client(
            service_name="bedrock",
            region_name=aws_region,
            config=boto3.session.Config(
                connect_timeout=conn_timeout,
                read_timeout=read_timeout
            )
        )
        print(f"科研专用 Bedrock 客户端在 {aws_region} 区域初始化成功")
        return research_bedrock_client
    except ConnectTimeoutError:
        raise Exception(f"连接超时：无法在 {aws_region} 区域初始化 Bedrock 客户端，请检查网络或调整超时配置")
    except ClientError as e:
        raise Exception(f"客户端初始化失败：{e.response['Error']['Message']}")
def generate_scientific_lit_review(
    research_topic,
    discipline_scope,
    review_depth,
    bedrock_client,
    model_id="DeepSeek-R1"
):
    """
    生成结构化文献综述并挖掘跨学科知识关联
    :param research_topic: 研究主题（如 "Climate change impact on agriculture"）
    :param discipline_scope: 学科范围（如 "Agriculture, Environmental Science, Economics"）
    :param review_depth: 综述深度（如 "Basic Overview", "In-depth Analysis", "Critical Evaluation"）
    :param bedrock_client: 已初始化的 Bedrock 客户端
    :param model_id: 调用的模型 ID，默认 DeepSeek-R1
    :return: 包含文献综述与跨学科关联的结构化内容（字符串）
    """
    # 构建科研场景专用 Prompt，明确结构化要求与跨学科挖掘需求
    research_prompt = f"""
    基于以下要求生成文献综述并挖掘跨学科知识关联：
    1. 研究主题：{research_topic}
    2. 学科范围：{discipline_scope}
    3. 综述深度：{review_depth}
    4. 输出结构：
       - 研究背景（近5年研究现状概述）
       - 核心文献观点（标注至少3类文献来源类型）
       - 跨学科关联点（分析与学科范围内其他领域的交叉研究方向）
       - 研究空白（当前领域未充分探索的问题）
    5. 跨学科挖掘要求：明确各关联学科的具体研究视角（如经济学视角下的农业经济损失评估）
    """

    try:
        model_response = bedrock_client.invoke_model(
            body=json.dumps({"prompt": research_prompt}),
            modelId=model_id,
            parameters={
                "task": "scientific_literature_review",
                "response_format": "structured_text",
                "temperature": 0.2  # 低温度保障科研内容准确性
            }
        )
        # 解析响应内容
        lit_review_content = model_response["body"].read().decode("utf-8")
        # 验证内容有效性
        if not lit_review_content.strip():
            raise ValueError("模型返回空内容，请检查研究主题表述或模型配置")
        return lit_review_content
    except ClientError as e:
        raise Exception(f"模型调用失败：{e.response['Error']['Message']}")
    except json.JSONDecodeError:
        raise Exception("响应格式错误：无法解析模型返回的 JSON 数据")
# ------------------- 示例：生成气候变化对农业影响的文献综述 -------------------
if __name__ == "__main__":
    # 1. 初始化科研专用 Bedrock 客户端
    sci_bedrock_client = init_bedrock_research_client(aws_region="us-east-1")

    # 2. 定义科研需求参数
    target_research_topic = "Climate change impact on agriculture"  # 目标研究主题
    target_discipline_scope = "Agriculture, Environmental Science, Agricultural Economics"  # 学科范围
    target_review_depth = "In-depth Analysis"  # 综述深度：深度分析

    # 3. 生成文献综述与跨学科关联内容
    try:
        scientific_review_result = generate_scientific_lit_review(
            research_topic=target_research_topic,
            discipline_scope=target_discipline_scope,
            review_depth=target_review_depth,
            bedrock_client=sci_bedrock_client
        )

        # 4. 输出并保存结果（按科研规范命名文件）
        current_time = datetime.now().strftime("%Y%m%d_%H%M%S")
        review_filename = f"Literature_Review_{target_research_topic.replace(' ', '_')}_{current_time}.txt"

        print("=" * 60)
        print(f"研究主题：{target_research_topic}")
        print(f"学科范围：{target_discipline_scope}")
        print(f"综述深度：{target_review_depth}")
        print("=" * 60)
        print("文献综述与跨学科关联结果：\n")
        print(scientific_review_result)

        # 保存结果到文件（便于科研文档整理）
        with open(review_filename, "w", encoding="utf-8") as f:
            f.write(f"研究主题：{target_research_topic}\n")
            f.write(f"学科范围：{target_discipline_scope}\n")
            f.write(f"综述深度：{target_review_depth}\n")
            f.write(f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write("=" * 40 + "\n")
            f.write(scientific_review_result)

        print(f"\n结果已保存至：{review_filename}")
    except Exception as e:
        print(f"文献综述生成失败：{str(e)}")