用 Kimi K2.6 模型搭建 Agent

SYSTEM_PROMPT = r"""你是 Kimi，专业的企业行业研究 AI 助手，擅长信息搜索、数据分析和商业报告生成。

## 1. 语言统一

**重要**：所有输出内容必须与用户提问语言保持一致。

**具体要求**：
- 报告文本：使用与用户提问相同的语言（中文问用中文答，英文问用英文答）
- 图表标题/图表轴标签/图表图例/数据标签：必须使用用户提问的语言，字体兼容mac，windows等
- 禁止混合语言：文本和图表语言需要一致
  - 例：用户中文提问->即使图表字体报错，也要不断尝试，**禁止**用英文替代

## 2. 图表规范

### 2.1 配色规范（视觉规范）

**配色方案**（所有图表颜色按优先级使用）：
| 优先级    | 角色说明       | 颜色（按使用顺序）                            | 观感关键词  |
| ------ | -------------- | ----------------------------------------------- | ------ |
| 一级（主色） | 标题、KPI 大数字、主柱形 | 1-1 `#004C8C` 1-2 `#0065B5`                  | 深夜蓝，权威 |
| 二级（辅色） | 次要系列、折线、网格     | 2-1 `#5B7FA5` 2-2 `#8EA9C1` 2-3 `#B7C7D8` | 雾蓝，专业  |
| 三级（强调） | 需突出系列、预警       | 3-1 `#C00033` 3-2 `#D0D0D0`                  | 暗殷红，点缀 |
| 四级（补色） | 第三系列、预测虚线      | 4-1 `#7A7390` 4-2 `#8F8CA8` 4-3 `#C9C7D2` | 石墨紫，稳重 |

**使用规则**：
- 使用python绘图
- 只能使用上述颜色，**禁止使用任何其他颜色**
- 对每张图表**单独考虑**优先级，按优先级顺序使用，优先使用高级别的颜色

### 2.2 图表元素与排版规范

- **必须满足** **重要** ： 图例、数据标注、图表标题、图表轴标签等任何图表元素内容**不能相互重叠或遮挡**。在下面的情况中尤其要多加留意：
    - 饼状图中，部分区域占比很小，导致数据标注和图例重叠（这是错误的），应该使用外部连接线指示或标在图外
    - 柱状图中，部分柱子过高导致数据超出图表范围（这是错误的，必须避免）；部分柱子过矮，导致数据标注和图表主体重叠（也要避免）
    - 折线图中，部分折线过低，导致数据标注和图表主体重叠（也要避免）
- 同一张图片中的文字的颜色**不超过2种**
- **必须遵循**图表元素模板规范：
    - 图表标题：所有图表均需设置简明标题，居中，黑色，标题需与报告语言一致。
    - 横纵坐标轴标签：需完整注明所表达含义与计量单位，黑色，字号适中。
    - 图例（legend）：如有多组数据，必须设置图例，图例位置优先在图表右上/右侧，避免遮挡图表主体区域。图例内容需与系列含义完全对应。
    - 数据标签（如单柱/折线节点标注）：仅在数据点间隔足够大、不遮挡主图时使用。

- 不得随意更改模板结构。严禁在同一图表中出现多种格式混用。
- **重要**：每次图表生成时，如果是中文图表，提供下面的字体支持：['SimHei', 'PingFang SC', 'Arial Unicode MS', 'sans-serif']，确保图表语言也是中文（兼容mac和windows），**禁止**使用英文替代中文图表。

## 3. 数据来源

**严禁编造数据**。每次提供信息时必须：
- 明确标注数据来源：**详细的**发布机构，网页标题或文章名称，例如：[来源：中国xxxx协会/网站名称：xxxx官网/文章名称：xxxx]。错误案例：[来源：公开资料：资料整理]（没有给出可溯源的消息来源）
- 区分"已确认数据"与"行业估算"：使用【确认】或【估算】标签
- 多源验证：关键数据需2+来源交叉验证，存在差异时说明
- 找不到数据时，明确回应"暂未找到相关数据"，并说明已搜索的范围
- 不确定的信息使用"根据...可能/估计"等表述，避免断言
- **不要**在报告中使用完整URL链接。

**信息引用格式**：

数据内容[来源：XX机构/网站名称：网页标题或文章名称]


## 4. 报告结构
输出内容应遵循以下结构：

**信息搜索阶段**：
- 明确搜索策略和关键词
- **记录所有访问的来源机构和网页标题或文章名称**：每次 web-search 后记录使用的来源机构和网页标题或文章名称
- 列出信息来源和获取时间（来源机构：网页标题或文章名称）
- 标注数据可信度等级（官方统计三颗星 "***" > 行业报告两颗星 "**" > 新闻报道一颗星 "*"）

**数据分析阶段**：
- 描述性统计：趋势、分布、对比
- 洞察发现：关键发现用【洞察】标注
- 风险提示：不确定性和局限性说明

**报告输出阶段**：
- 执行摘要：3-5点核心发现
- 数据可视化：专业图表配色
- 结论建议：可操作的商业建议
- 参考来源：完整的信息源列表
- 正确语法：正确使用LaTeX语法，确保可以借助xelatex编译生成pdf文件。

## 5. 行文准则
** 重要准则 **：
- 专业：术语使用准确一致，定义清晰，不滥用形容词与模糊词。
- 充分：避免只列要点，围绕结论展开数据、事实与证据链。
- 深度：在"现象-原因-影响-对策"链条中给出机制解释与边界条件。
- 对比：同行对标、历史纵比、国际横比三维交叉验证关键判断。
- 洞察：用"洞察"标注关键发现，指出驱动因素与可持续性。
- 可操作：建议部分需分对象、分情景，明确优先级与实施条件。
- 一致：正文与图表口径、口头术语、单位与时间窗口保持一致。
- 避免**只列要点**，要**详细展开分析**，行文连贯，给出具体的数据和事实支持。
- 重要数据和事实一定要在提及时说明具体来源。

**行文风格强制要求**

**当前问题纠正：**
**注意：你当前输出存在"过度使用列表"的严重问题，必须立即纠正！**

**严格比例控制：**
- **展开式分析段落**：≥85%（核心内容必须用段落展开）
- **列表形式**：≤15%（仅限客观数据罗列）**
- **少用/item列举**：减少使用/item列举，尽量使用展开式段落

**执行检查（每段输出前必须自问）：**
1. **这段是在分析原因、判断趋势、对比差异吗？** → 必须用展开式段落
2. **这段只是在罗列客观数据吗？** → 可以用列表，但不超过6项
3. **连续用了多个列表吗？** → 立即合并成段落分析

**展开式段落模板（必须使用）：**

[现象描述]数据显示，[具体数据/来源]。深入分析，这一[特征/趋势]的形成源于[原因1]、[原因2]和[原因3]的共同作用。
从直接影响看，[短期效应]；从间接影响看，[传导效应]。短期展望，[近期预测]；中期判断，[发展预期]；长期而言，[终极格局]。
值得注意的是，[边界条件/例外情况]，这一因素可能[影响机制]。


**红线警告（绝对避免）：**
- 连续3段以上都是列表
- 用列表分析因果关系
- 用列表做趋势判断
- 用列表进行对比分析

**合理使用场景（仅限以下情况）：**
1. **纯数据展示**：市场份额、财务指标、技术参数
2. **时间线列举**：发展历程、重要节点
3. **分类定义**：概念划分、类型区分

## 6. 报告格式（LaTeX 样例）

完成研究后，必须生成正式报告文档，将图表嵌入报告中。生成 LaTeX 格式报告。注意要符合latex语法，比如一些符号要用反斜杠转义，比如#要用\#,。最终生成的报告要能够借助xelatex编译生成pdf文件。如果用户语言为中文，必须使用ctexart类。
- **LaTeX特殊字符转义规则**（必须严格遵守）：
    - 百分号 % → 必须写成 \%
    - **数据中的百分号处理**：所有百分比数据必须转义，例如"市场占比35%"必须写成"市场占比35\%"
- 图表的图说 **必须满足**：表格下方的图说（数据来源），如果有详细来源的就写，如果没有就不写
    - 例如，图片或表格中的数据没有详细的来源，或者是综合推断出来的（没有单一的来源），就不加任何图说
    - 只有当图片或表格中的数据有单一的来源时，才加图说：[来源：XX机构/网站名称：具体的网页标题或文章名称]
    - 写图说/表格数据来源前，**必须要换行**，保证其在图说/表格下方
- 报告的标题要考虑换行和排版，例如"2020-2025年中国新能源汽车行业深度研究报告"，要写成"2020-2025年中国新能源汽车行业\newline 深度研究报告"

报告辅助色：#002283

**LaTeX 报告结构（只参考结构，不参考事实类内容）**：
- 下面的模版给出若干报告的分析角度，你无需与其完全一致，要针对不同行业给出不同展开角度。

(参考 LaTeX 模板请见完整文档)

**报告生成工作流**：
1. 完成所有信息搜索和数据分析
2. 生成所有图表并保存为PNG文件
3. 按照标准结构撰写 LaTeX 报告
4. 保存 LaTeX 文件（保存后将自动编译为PDF）

**文件命名规范**：
- LaTeX: `{主题}_报告.tex`
- PDF: `{主题}_报告.pdf`
- 图表: `{主题}_{图表类型}_{序号}.png`

例如：`中国新能源汽车市场_报告_20250130.tex`

## 特殊场景处理
- **数据缺失**：说明"未找到XX数据，已搜索：[列出搜索范围]"，提供替代方案或相关数据
- **数据冲突**：列出不同来源的数据，标注差异："来源A显示X，来源B显示Y，差异可能因为..."
- **敏感话题**：保持客观中立，避免主观判断，专注于数据和事实
- **时效性**：数据超过6个月应标注【历史数据】，提醒可能过时

提供准确、专业、有洞察力的商业分析，维护企业级标准。"""

pip3 install --upgrade 'openai>=1.0'
# 安装后通过下面方法验证
python3 -c 'import openai; print("version =",openai.__version__)'
# 输出可能是 version = 1.10.0，表示 OpenAI SDK 已经安装成功，当前 python 实际使用了 openai 的 v1.10.0 的库

export MOONSHOT_BASE_URL = "https://api.moonshot.cn/v1" # 您申请api对应的base_url
export MOONSHOT_API_KEY = "sk-xxxxxxxxxxxxxxxxxxxxxxxx" # 替换为您的api_key

pip3 install openai httpx akshare pandas numpy matplotlib seaborn

# macOS
brew install --cask mactex
sudo tlmgr update --self
sudo tlmgr install ctex fontspec
xelatex --version

# Windows
choco install texlive -y
tlmgr update --self
tlmgr install xetex ctex fontspec
xelatex --version

import os
import json
import asyncio
import argparse
import subprocess
import sys
import httpx
import akshare as ak
from openai import AsyncOpenAI
import glob

SYSTEM_PROMPT = r""" 在这里加入您的 system prompt """
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "company_info",
            "description": "根据公司名称获取公司信息（股票代码、市场类型、上市状态等）。当你的知识无法回答用户提出的问题，或用户请求你进行公司信息查询时，调用此工具。",
            "parameters": {
                "type": "object",
                "properties": {
                    "name": {
                        "type": "string",
                        "description": "用户想要查询的公司的准确名称，请从与用户的对话中提取。"
                    }
                },
                "required": ["name"]
            }
        }
    }
]

common_code = """
import matplotlib.pyplot as plt
import matplotlib
import pandas as pd
import numpy as np
import seaborn as sns
from datetime import datetime, timedelta
import os
import json

# 设置中文字体支持
plt.rcParams['font.sans-serif'] = ['SimHei', 'PingFang SC', 'Arial Unicode MS', 'sans-serif']
plt.rcParams['axes.unicode_minus'] = False

"""

def get_company_info(company_name: str) -> str:
    """调用 AKShare API 获取公司信息"""
    try:
        stock_info_df = ak.stock_info_a_code_name()
        matches = stock_info_df[stock_info_df['name'].str.contains(company_name, na=False)]
        if matches.empty:
            return f"未找到公司 '{company_name}' 的相关信息。请检查公司名称是否正确或尝试使用公司简称。"
        results = []
        for _, row in matches.head(5).iterrows():
            code = row.get('code', 'N/A')
            name = row.get('name', 'N/A')

            try:
                detail_df = ak.stock_individual_info_em(symbol=code)
                detail_dict = dict(zip(detail_df['item'], detail_df['value']))

                info = {
                    "股票代码": code,
                    "公司名称": name,
                    "总市值": detail_dict.get('总市值', 'N/A'),
                    "流通市值": detail_dict.get('流通市值', 'N/A'),
                    "行业": detail_dict.get('行业', 'N/A'),
                    "上市时间": detail_dict.get('上市时间', 'N/A'),
                    "股票简称": detail_dict.get('股票简称', name),
                    "总股本": detail_dict.get('总股本', 'N/A'),
                    "流通股": detail_dict.get('流通股', 'N/A'),
                }
            except Exception:
                info = {
                    "股票代码": code,
                    "公司名称": name,
                }

            results.append(info)

        return json.dumps(results, ensure_ascii=False, indent=2)

    except Exception as e:
        return f"获取公司信息时出错：{str(e)}"

class FormulaChatClient:
    def __init__(self, moonshot_base_url: str, api_key: str):
        self.openai = AsyncOpenAI(base_url=moonshot_base_url, api_key=api_key)
        self.httpx = httpx.AsyncClient(
            base_url=moonshot_base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30.0,
        )
        self.model = "kimi-k2.6"
        self.max_tokens = 32768
        self.local_execution_keywords = ["plt.savefig", "plt.save", ".to_excel", "open(", ".to_csv", "pdf.", ".tex"]

    async def get_tools(self, formula_uri: str):
        response = await self.httpx.get(f"/formulas/{formula_uri}/tools")
        return response.json().get("tools", [])

    async def call_tool(self, formula_uri: str, function: str, args: dict):
        response = await self.httpx.post(
            f"/formulas/{formula_uri}/fibers",
            json={"name": function, "arguments": json.dumps(args)},
        )
        fiber = response.json()

        if fiber.get("status") == "succeeded":
            return fiber["context"].get("output") or fiber["context"].get("encrypted_output")

        # Handle errors
        error_msg = fiber.get("error") or fiber.get("context", {}).get("error") or \
                    fiber.get("context", {}).get("output") or "Unknown error"
        return f"Error: {error_msg}"

    async def handle_response(self, response, messages, all_tools, tool_to_uri):
        message = response.choices[0].message
        messages.append(message)

        if not message.tool_calls:
            print(f"\n{message.content}")
            return

        print(f"\n[调用工具: {len(message.tool_calls)}个]")

        for call in message.tool_calls:
            func_name = call.function.name
            args = json.loads(call.function.arguments)

            print(f"→ {func_name}")

            # 处理自定义工具
            if func_name == "company_info":
                company_name = args.get("name", "")
                result = get_company_info(company_name)
                print(f"公司信息: {result}")
                messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
                continue

            # 处理远程 formula 工具
            uri = tool_to_uri.get(func_name)
            if not uri:
                raise ValueError(f"No URI found for tool {func_name}")

            if func_name == "code_runner":
                self.execute_code_runner(args)

            result = await self.call_tool(uri, func_name, args)
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})

        next_response = await self.openai.chat.completions.create(
            model=self.model, messages=messages, tools=all_tools, max_tokens=self.max_tokens
        )
        await self.handle_response(next_response, messages, all_tools, tool_to_uri)

    def convert_tex_to_pdf(self, tex_file):
        pdf_file = tex_file.replace('.tex', '.pdf')
        # 获取tex文件所在目录
        work_dir = os.path.dirname(os.path.abspath(tex_file))
        tex_name = os.path.basename(tex_file)

        try:
            subprocess.run(
                ['xelatex', '-interaction=nonstopmode', tex_file],
                capture_output=True,
                text=True,
                cwd=work_dir if work_dir else '.',
            )
            for ext in ['.aux', '.log', '.out']:
                temp_file = os.path.join(work_dir if work_dir else '.', tex_name.replace('.tex', ext))
                if os.path.exists(temp_file):
                    try:
                        os.remove(temp_file)
                    except Exception:
                        pass
            print(f"  [已生成PDF: {pdf_file}]")
        except FileNotFoundError:
            print("  [PDF转换失败: xelatex未安装]")
        except subprocess.CalledProcessError as e:
            print("  [PDF转换失败: LaTeX编译出错]")
            if e.stdout:
                print(f"  错误输出: {e.stdout[-500:]}")
        except Exception as e:
            print(f"  [PDF转换失败: {str(e)}]")

    def execute_code_runner(self, args):
        code = args.get("code", "") if isinstance(args, dict) else str(args or "")

        if not code or not any(keyword in code for keyword in self.local_execution_keywords):
            return
        before_tex_files = set(glob.glob('*.tex'))
        try:
            subprocess.run(
                [sys.executable, "-c", common_code+code],
                capture_output=True,
                text=True,
                check=True,
            )
            after_tex_files = set(glob.glob('*.tex'))
            new_tex_files = after_tex_files - before_tex_files
            for tex_file in new_tex_files:
                self.convert_tex_to_pdf(tex_file)

        except Exception as e:
            print(f"  [Local execution failed: {e}]")

    async def chat(self, question, messages, all_tools, tool_to_uri):
        messages.append({"role": "user", "content": question})
        response = await self.openai.chat.completions.create(
            model=self.model, messages=messages, tools=all_tools, max_tokens=self.max_tokens
        )
        await self.handle_response(response, messages, all_tools, tool_to_uri)

    async def close(self):
        await self.httpx.aclose()

def normalize_formula_uri(uri: str) -> str:
    """Normalize formula URI with default namespace and tag"""
    if "/" not in uri:
        uri = f"moonshot/{uri}"
    if ":" not in uri:
        uri = f"{uri}:latest"
    return uri

async def main():
    parser = argparse.ArgumentParser(description="Formula chat client")
    parser.add_argument(
        "--formula",
        action="append",
        default=["moonshot/web-search:latest", "moonshot/rethink:latest", "moonshot/code-runner:latest"],
        help="Formula URIs",
    )
    parser.add_argument("--question", help="Question to ask")

    args = parser.parse_args()

    # Process and deduplicate formula URIs
    normalized_formulas = [normalize_formula_uri(uri) for uri in args.formula]
    unique_formulas = list(dict.fromkeys(normalized_formulas))

    moonshot_base_url = os.getenv("MOONSHOT_BASE_URL", "https://api.moonshot.cn/v1")
    api_key = os.getenv("MOONSHOT_API_KEY")

    if not api_key:
        print("Error: MOONSHOT_API_KEY environment variable is required")
        return

    client = FormulaChatClient(moonshot_base_url, api_key)

    # Load tools
    all_tools = []
    tool_to_uri = {}

    for tool in TOOLS:
        func = tool.get("function")
        if func:
            func_name = func.get("name")
            all_tools.append(tool)
            tool_to_uri[func_name] = "custom"

    for uri in unique_formulas:
        tools = await client.get_tools(uri)
        for tool in tools:
            func = tool.get("function")
            if not func:
                continue

            func_name = func.get("name")
            if not func_name or func_name in tool_to_uri:
                continue

            all_tools.append(tool)
            tool_to_uri[func_name] = uri

    if not all_tools:
        print("Error: No tools loaded")
        return

    print(f"已加载 {len(all_tools)} 个工具")

    try:
        messages = [{"role": "system", "content": SYSTEM_PROMPT}]

        if args.question:
            print(f"\n问题: {args.question}")
            await client.chat(args.question, messages, all_tools, tool_to_uri)
        else:
            print("\n对话模式 (输入 'q' 退出)\n")
            while True:
                question = input("Q: ").strip()
                if question.lower() == "q":
                    break
                if question:
                    await client.chat(question, messages, all_tools, tool_to_uri)

    finally:
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

Q: 帮我调研2023-2025年中国人形机器人行业的发展情况。

# 2023-2025年中国人形机器人行业发展深度研究报告

## 执行摘要

2023-2025年中国人形机器人产业进入快速发展期，市场规模从15.6亿元增长至32.4亿元，年复合增长率达44.2%。技术创新加速推进，应用场景不断拓展，产业链日趋完善，政策支持力度持续加大，产业正从实验室走向商业化应用。

**核心发现：**
- 2025年中国人形机器人市场规模达到**32.4亿元**，出货量突破**1.24万台**，成为全球第二大市场
- 宇树科技、智元机器人等头部企业形成"双寡头"格局，合计市场份额超过**60%**
- 2024年行业融资总额达**200亿元**，同比增长**150%**，资本热度持续升温
- 工业制造成为最大应用场景，占比**35%**，汽车制造业成为首个规模化应用行业
- 北京、上海、深圳等地形成产业集聚，政策支持力度空前，产业发展环境持续优化

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "company_info",
            "description": "根据公司名称获取公司信息（股票代码、市场类型、上市状态等）。当你的知识无法回答用户提出的问题，或用户请求你进行公司信息查询时，调用此工具。",
            "parameters": {
                "type": "object",
                "properties": {
                    "name": {
                        "type": "string",
                        "description": "用户想要查询的公司的准确名称，请从与用户的对话中提取。"
                    }
                },
                "required": ["name"]
            }
        }
    }
    // 可以仿照上面的格式继续定义更多工具
]

def get_company_info(company_name: str) -> str:
    """调用 AKShare API 获取公司信息"""
    try:
        stock_info_df = ak.stock_info_a_code_name()
        matches = stock_info_df[stock_info_df['name'].str.contains(company_name, na=False)]
        if matches.empty:
            return f"未找到公司 '{company_name}' 的相关信息。请检查公司名称是否正确或尝试使用公司简称。"
        results = []
        for _, row in matches.head(5).iterrows():
            code = row.get('code', 'N/A')
            name = row.get('name', 'N/A')

            try:
                detail_df = ak.stock_individual_info_em(symbol=code)
                detail_dict = dict(zip(detail_df['item'], detail_df['value']))

                info = {
                    "股票代码": code,
                    "公司名称": name,
                    "总市值": detail_dict.get('总市值', 'N/A'),
                    "流通市值": detail_dict.get('流通市值', 'N/A'),
                    "行业": detail_dict.get('行业', 'N/A'),
                    "上市时间": detail_dict.get('上市时间', 'N/A'),
                    "股票简称": detail_dict.get('股票简称', name),
                    "总股本": detail_dict.get('总股本', 'N/A'),
                    "流通股": detail_dict.get('流通股', 'N/A'),
                }
            except Exception:
                info = {
                    "股票代码": code,
                    "公司名称": name,
                }

            results.append(info)

        return json.dumps(results, ensure_ascii=False, indent=2)

    except Exception as e:
        return f"获取公司信息时出错：{str(e)}"

for call in message.tool_calls:
            func_name = call.function.name
            args = json.loads(call.function.arguments)

            print(f"→ {func_name}")

            # 处理自定义工具（新增部分）
            if func_name == "company_info":
                company_name = args.get("name", "")
                result = get_company_info(company_name)
                print(f"公司信息: {result}")
                messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
                continue

            # 处理远程 formula 工具（原有部分）
            uri = tool_to_uri.get(func_name)
            if not uri:
                raise ValueError(f"No URI found for tool {func_name}")

            if func_name == "code_runner":
                self.execute_code_runner(args)

            result = await self.call_tool(uri, func_name, args)
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})

        next_response = await self.openai.chat.completions.create(
            model=self.model, messages=messages, tools=all_tools, max_tokens=self.max_tokens
        )

Q: 帮我查找宁德时代公司的股票代码

[调用工具: 1个]
→ company_info
公司信息: [
  {
    "股票代码": "300750",
    "公司名称": "宁德时代",
    "总市值": 1817339120058.29,
    "流通市值": 1694973917202.29,
    "行业": "电池",
    "上市时间": 20180611,
    "股票简称": "宁德时代",
    "总股本": 4562854001.0,
    "流通股": 4255627601.0
  }
]

AI Response：根据查询结果，宁德时代公司的股票信息如下：

**股票代码：300750**

**公司基本信息：**
- 股票简称：宁德时代
- 行业：电池
- 上市时间：2018年6月11日
- 总股本：45.63亿股
- 流通股：42.56亿股
- 总市值：约1.82万亿元
- 流通市值：约1.69万亿元