Grok / xAI 大模型应用开发教程
零基础系统学习 Grok / xAI 大模型应用开发，从原理到实战，掌握实时数据访问、多模态理解、Function Calling、DeepSearch 等核心技能。
适合人群：AI 开发者、数据分析师、全栈工程师
预计学习时间：12-16 小时
前置要求：基础 Python 编程能力、了解 REST API 概念
目录
- 第一章：xAI 公司与 Grok 模型家族概述
- 第二章：Grok-3 架构与核心能力
- 第三章：Grok API 接入实战
- 第四章：实时数据访问——X 平台集成
- 第五章：多模态理解能力
- 第六章：Function Calling 函数调用
- 第七章：Grok Studio 创意模式
- 第八章：DeepSearch 深度搜索
- 第九章：Grok 与 OpenAI / Anthropic 对比分析
- 第十章：企业级集成方案
- 第十一章：实战项目一——实时舆情分析系统
- 第十二章：实战项目二——智能 AI 助手
- 第十三章：常见问题与排错指南
- 附录：资源汇总
第一章：xAI 公司与 Grok 模型家族概述
1.1 xAI 公司简介
xAI 是由 Elon Musk 于 2023 年创立的人工智能公司，总部位于美国旧金山湾区。公司的核心使命是"理解宇宙的真实本质"（Understand the true nature of the universe）。与 OpenAI、Google DeepMind、Anthropic 等公司并列，xAI 是全球顶级的 AI 研究机构之一。
xAI 的核心竞争力在于：
- 算力优势：在田纳西州孟菲斯建设了 Colossus 超级计算集群，动用超过 10 万块 Nvidia H100 GPU，训练算力为前代模型的 10 倍。
- 数据优势：与 X（原 Twitter）平台深度整合，拥有海量实时社交媒体数据的访问权限。
- 产品定位：强调"无审查"和"推理透明"，与主流 AI 助手形成差异化竞争。
1.2 Grok 模型家族演进
| 模型 | 发布时间 | 核心特点 |
|---|---|---|
| Grok-1 | 2023 年 11 月 | 首个模型，33B 参数，开源发布 |
| Grok-1.5 | 2024 年 3 月 | 128K 上下文窗口，推理能力提升 |
| Grok-2 | 2024 年 8 月 | 性能大幅跃升，支持图像理解 |
| Grok-3 | 2025 年 2 月 | 旗舰模型，AIME/GPQA 基准领先 |
| Grok-3 Mini | 2025 年 4 月 | 轻量版，性价比极高 |
1.3 当前可用模型详解
截至 2025 年，xAI API 平台提供以下四个核心模型：
grok-3-beta 旗舰模型，适合复杂推理、编码和数据分析。拥有最强的综合能力，适合对质量要求最高的场景。
grok-3-fast-beta 旗舰模型的速度优化版本，在保持高质量的同时注重低延迟响应，适合交互式应用场景。
grok-3-mini-beta 轻量模型，在性能与成本之间取得优秀平衡。适合实时应用和大规模批量处理。
grok-3-mini-fast-beta 超快轻量模型，专为高吞吐量任务设计，延迟最低、成本最经济。
此外，每个模型都有对应的推理版本（reasoning variant），支持 Think 模式，可返回详细的推理链。
第二章：Grok-3 架构与核心能力
2.1 模型架构概述
Grok-3 基于 Transformer 架构，采用了多项创新技术：
- 大规模 MoE（Mixture of Experts）：通过稀疏激活机制，在增大模型容量的同时控制推理成本。
- 超长上下文窗口：支持 131,072 tokens 的上下文长度（约 200 页文本），内测阶段已展示百万 token 潜力。
- 多模态融合：原生支持文本和图像输入，未来将扩展至视频和音频。
2.2 基准测试表现
Grok-3 在多项权威基准测试中表现优异：
| 基准测试 | Grok-3 Beta | Grok-3 Mini | GPT-o1 | Claude 3.5 Sonnet |
|---|---|---|---|---|
| AIME 2025（数学） | 93.3% | 95.8% | 79% | - |
| GPQA（科学） | 84.6% | - | 78% | - |
| LiveCodeBench（编码） | 79.4% | 80.4% | 72.9% | - |
| LMArena ELO | 1400 | - | - | - |
2.3 核心能力矩阵
推理能力：Grok-3 的推理版本支持 Think 模式，API 返回详细的推理链，展示模型的思考过程。在解决数学问题时，会列出推导步骤和自查逻辑。
编码能力：代码生成的准确性和速度优于同类模型，尤其擅长前端开发和逻辑优化。能生成完整的交互式网页应用。
长文档处理：131K 的上下文窗口使其能一次性处理完整代码库、长篇法律文档或多轮复杂对话。
实时数据访问：通过 DeepSearch 功能，可从 X 平台和互联网抓取实时数据。
第三章：Grok API 接入实战
3.1 获取 API Key
- 访问 xAI API 控制台
- 使用 X 账号或邮箱注册
- 新用户可获得 $50 免费额度
- 在 API Keys 页面创建新的 API Key
注意：妥善保管你的 API Key，不要将其提交到公开代码仓库。
3.2 安装 SDK
xAI 提供与 OpenAI SDK 兼容的接口，这意味着你可以直接使用 OpenAI 的 Python SDK 来调用 Grok API：
pip install openai
也可以使用 xAI 官方 SDK：
pip install xai-sdk
3.3 第一个 Grok API 调用
使用 OpenAI 兼容方式调用：
from openai import OpenAI
# 初始化客户端，base_url 指向 xAI API
client = OpenAI(
api_key="your-xai-api-key",
base_url="https://api.x.ai/v1"
)
# 发起对话请求
response = client.chat.completions.create(
model="grok-3-beta",
messages=[
{"role": "system", "content": "你是一个专业的技术助手。"},
{"role": "user", "content": "请解释什么是 Transformer 架构？"}
],
temperature=0.7,
max_tokens=1024
)
# 输出结果
print(response.choices[0].message.content)
3.4 流式输出
from openai import OpenAI
client = OpenAI(
api_key="your-xai-api-key",
base_url="https://api.x.ai/v1"
)
# 流式请求
stream = client.chat.completions.create(
model="grok-3-mini-beta",
messages=[
{"role": "user", "content": "用 Python 写一个快速排序算法"}
],
stream=True
)
# 逐块输出
for chunk in stream:
if chunk.choices[0].delta.content is not None:
print(chunk.choices[0].delta.content, end="")
3.5 使用 xAI 官方 SDK
import xai_sdk
import xai_sdk.chat as chat
# 初始化客户端
client = xai_sdk.Client(api_key="your-xai-api-key")
# 创建对话
conversation = client.chat.create(
model="grok-3-beta",
temperature=0.7
)
# 添加消息
conversation.append(chat.system("你是一个专业的数据分析师。"))
conversation.append(chat.user("分析一下2025年全球AI市场的趋势。"))
# 获取回复
response = conversation.sample()
print(response.content)
3.6 完整的对话管理类
from openai import OpenAI
from typing import List, Dict, Optional
import json
class GrokChatManager:
"""Grok 对话管理器，支持多轮对话和上下文管理。"""
def __init__(self, api_key: str, model: str = "grok-3-mini-beta"):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
self.model = model
self.conversation_history: List[Dict[str, str]] = []
def set_system_prompt(self, prompt: str):
"""设置系统提示词。"""
self.conversation_history = [
{"role": "system", "content": prompt}
]
def chat(self, user_message: str, temperature: float = 0.7,
max_tokens: int = 2048) -> str:
"""发送消息并获取回复。"""
self.conversation_history.append(
{"role": "user", "content": user_message}
)
response = self.client.chat.completions.create(
model=self.model,
messages=self.conversation_history,
temperature=temperature,
max_tokens=max_tokens
)
assistant_message = response.choices[0].message.content
self.conversation_history.append(
{"role": "assistant", "content": assistant_message}
)
return assistant_message
def chat_stream(self, user_message: str, temperature: float = 0.7):
"""流式输出回复。"""
self.conversation_history.append(
{"role": "user", "content": user_message}
)
stream = self.client.chat.completions.create(
model=self.model,
messages=self.conversation_history,
temperature=temperature,
stream=True
)
full_response = ""
for chunk in stream:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
full_response += content
yield content
self.conversation_history.append(
{"role": "assistant", "content": full_response}
)
def clear_history(self):
"""清空对话历史。"""
self.conversation_history = []
def get_history(self) -> List[Dict[str, str]]:
"""获取对话历史。"""
return self.conversation_history.copy()
def export_history(self, filepath: str):
"""导出对话历史到 JSON 文件。"""
with open(filepath, "w", encoding="utf-8") as f:
json.dump(self.conversation_history, f,
ensure_ascii=False, indent=2)
# 使用示例
if __name__ == "__main__":
manager = GrokChatManager(api_key="your-xai-api-key")
manager.set_system_prompt("你是一个友好的AI助手，擅长用简洁的语言解释复杂概念。")
# 多轮对话
print(manager.chat("什么是机器学习？"))
print(manager.chat("它和深度学习有什么区别？"))
print(manager.chat("能给我一个实际应用的例子吗？"))
3.7 API 参数详解
| 参数 | 类型 | 说明 | 默认值 |
|---|---|---|---|
| model | string | 模型名称 | 必填 |
| messages | array | 消息数组 | 必填 |
| temperature | float | 采样温度 (0-2) | 1.0 |
| max_tokens | int | 最大输出 token 数 | 模型默认 |
| top_p | float | 核采样概率 | 1.0 |
| stream | bool | 是否流式输出 | false |
| stop | string/array | 停止序列 | null |
| presence_penalty | float | 存在惩罚 (-2 到 2) | 0 |
| frequency_penalty | float | 频率惩罚 (-2 到 2) | 0 |
3.8 错误处理与重试机制
import time
from openai import OpenAI, APIError, RateLimitError, APIConnectionError
class RobustGrokClient:
"""带重试机制的 Grok 客户端。"""
def __init__(self, api_key: str, model: str = "grok-3-mini-beta",
max_retries: int = 3, base_delay: float = 1.0):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
self.model = model
self.max_retries = max_retries
self.base_delay = base_delay
def call_with_retry(self, messages: list, **kwargs) -> str:
"""带指数退避重试的 API 调用。"""
last_exception = None
for attempt in range(self.max_retries):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
**kwargs
)
return response.choices[0].message.content
except RateLimitError as e:
last_exception = e
delay = self.base_delay * (2 ** attempt)
print(f"速率限制，等待 {delay} 秒后重试...")
time.sleep(delay)
except APIConnectionError as e:
last_exception = e
delay = self.base_delay * (2 ** attempt)
print(f"连接错误，等待 {delay} 秒后重试...")
time.sleep(delay)
except APIError as e:
last_exception = e
if e.status_code >= 500:
delay = self.base_delay * (2 ** attempt)
print(f"服务器错误 ({e.status_code})，等待 {delay} 秒后重试...")
time.sleep(delay)
else:
raise # 4xx 错误不重试
raise last_exception
3.9 定价与成本控制
Grok API 按 token 计费，以下是各模型的定价（每百万 token）：
| 模型 | 输入价格 | 输出价格 |
|---|---|---|
| grok-3-beta | $2.00 | $8.00 |
| grok-3-fast-beta | $2.50 | $10.00 |
| grok-3-mini-beta | $0.40 | $1.60 |
| grok-3-mini-fast-beta | $0.50 | $2.00 |
成本优化建议：
- 选择合适模型：简单任务使用 mini 版本，复杂推理任务才用完整版
- 控制输出长度：合理设置
max_tokens避免不必要的输出 - 利用缓存：对相同或相似请求做结果缓存
- 批量处理：批量请求可享 25% 折扣
# 成本估算工具
def estimate_cost(input_tokens: int, output_tokens: int,
model: str = "grok-3-mini-beta") -> float:
"""估算 API 调用成本（美元）。"""
pricing = {
"grok-3-beta": (2.0, 8.0),
"grok-3-fast-beta": (2.5, 10.0),
"grok-3-mini-beta": (0.4, 1.6),
"grok-3-mini-fast-beta": (0.5, 2.0),
}
input_price, output_price = pricing.get(model, (0.4, 1.6))
cost = (input_tokens * input_price + output_tokens * output_price) / 1_000_000
return round(cost, 6)
# 示例
print(f"成本: ${estimate_cost(13000, 2000, 'grok-3-mini-beta')}")
# 输出: 成本: $0.0084
第四章：实时数据访问——X 平台集成
4.1 X 平台数据的独特价值
Grok 与其他大模型的最大差异之一，是其与 X（原 Twitter）平台的深度整合。X 平台每天产生数亿条推文，涵盖了全球实时热点、舆情动态、技术讨论和行业趋势。
这种实时数据访问能力使得 Grok 在以下场景中具有独特优势：
- 舆情监控：实时追踪品牌、产品或话题的公众情绪
- 趋势分析：发现正在兴起的话题和技术趋势
- 新闻摘要：基于实时信息源生成最新资讯摘要
- 市场研究：分析消费者反馈和市场动态
4.2 通过 API 访问实时数据
Grok API 的对话接口天然支持实时数据访问。当你的问题涉及当前事件或实时信息时，Grok 会自动利用其数据优势进行回答：
from openai import OpenAI
client = OpenAI(
api_key="your-xai-api-key",
base_url="https://api.x.ai/v1"
)
# 请求实时信息
response = client.chat.completions.create(
model="grok-3-beta",
messages=[
{
"role": "system",
"content": "你是一个实时信息分析助手，请基于最新数据回答问题。"
},
{
"role": "user",
"content": "分析一下最近一周关于人工智能的主要讨论热点。"
}
],
temperature=0.3 # 降低温度以获得更准确的事实性回答
)
print(response.choices[0].message.content)
4.3 构建实时数据查询管道
from openai import OpenAI
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional
import json
@dataclass
class TrendQuery:
"""趋势查询参数。"""
topic: str
time_range: str = "24h" # 24h, 7d, 30d
language: str = "zh"
sentiment_focus: bool = True
class RealTimeAnalyzer:
"""实时数据分析器。"""
def __init__(self, api_key: str):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
def analyze_trend(self, query: TrendQuery) -> dict:
"""分析指定话题的趋势。"""
prompt = f"""请分析以下话题在 X 平台上的最新讨论趋势：
话题：{query.topic}
时间范围：最近 {query.time_range}
语言偏好：{query.language}
请提供：
1. 讨论热度概述（高/中/低）
2. 主要观点和立场分布
3. 关键意见领袖的发言摘要
4. 情绪分析（正面/负面/中性比例）
5. 未来趋势预测
请以 JSON 格式返回结果。"""
response = self.client.chat.completions.create(
model="grok-3-beta",
messages=[
{"role": "system", "content": "你是一个专业的社交媒体分析师，擅长舆情分析和趋势预测。请始终以JSON格式返回分析结果。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=2048
)
result_text = response.choices[0].message.content
try:
# 提取 JSON 部分
json_start = result_text.find('{')
json_end = result_text.rfind('}') + 1
return json.loads(result_text[json_start:json_end])
except json.JSONDecodeError:
return {"raw_text": result_text}
def compare_topics(self, topics: List[str]) -> dict:
"""对比多个话题的热度。"""
topics_str = "、".join(topics)
prompt = f"请对比以下话题在 X 平台上的讨论热度和公众情绪：{topics_str}。以 JSON 格式返回对比结果。"
response = self.client.chat.completions.create(
model="grok-3-mini-beta",
messages=[
{"role": "system", "content": "你是社交媒体分析专家。返回JSON格式。"},
{"role": "user", "content": prompt}
],
temperature=0.3
)
result_text = response.choices[0].message.content
try:
json_start = result_text.find('{')
json_end = result_text.rfind('}') + 1
return json.loads(result_text[json_start:json_end])
except json.JSONDecodeError:
return {"raw_text": result_text}
# 使用示例
if __name__ == "__main__":
analyzer = RealTimeAnalyzer(api_key="your-xai-api-key")
# 分析单一话题
query = TrendQuery(topic="GPT-5 发布", time_range="7d")
result = analyzer.analyze_trend(query)
print(json.dumps(result, ensure_ascii=False, indent=2))
# 对比多个话题
comparison = analyzer.compare_topics([
"Grok-3", "GPT-4o", "Claude 3.5", "Gemini"
])
print(json.dumps(comparison, ensure_ascii=False, indent=2))
4.4 实时数据处理的最佳实践
- 提示词设计：明确指定数据来源、时间范围和分析维度
- 温度控制：事实性查询使用低温度 (0.1-0.3)，创意分析使用中温度 (0.5-0.7)
- 结果验证：实时数据可能存在噪声，建议交叉验证关键信息
- 速率控制：实时查询消耗更多资源，注意 API 速率限制
第五章：多模态理解能力
5.1 图像理解
Grok-3 支持图像输入，能够分析图表、截图、文档等多种视觉内容：
from openai import OpenAI
import base64
def analyze_image(api_key: str, image_path: str,
question: str) -> str:
"""分析本地图像。"""
client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
# 读取并编码图像
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode("utf-8")
# 根据文件扩展名确定 MIME 类型
ext = image_path.lower().split(".")[-1]
mime_map = {"jpg": "jpeg", "jpeg": "jpeg", "png": "png",
"gif": "gif", "webp": "webp"}
mime_type = f"image/{mime_map.get(ext, 'jpeg')}"
response = client.chat.completions.create(
model="grok-3-beta",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": question},
{
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{image_data}"
}
}
]
}
],
max_tokens=1024
)
return response.choices[0].message.content
def analyze_image_url(api_key: str, image_url: str,
question: str) -> str:
"""分析网络图像。"""
client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
response = client.chat.completions.create(
model="grok-3-beta",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": question},
{
"type": "image_url",
"image_url": {"url": image_url}
}
]
}
],
max_tokens=1024
)
return response.choices[0].message.content
# 使用示例
result = analyze_image(
api_key="your-xai-api-key",
image_path="chart.png",
question="请分析这张图表中的数据趋势，并总结关键发现。"
)
print(result)
5.2 多图对比分析
def compare_images(api_key: str, image_paths: list,
question: str) -> str:
"""对比分析多张图像。"""
client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
content = [{"type": "text", "text": question}]
for path in image_paths:
with open(path, "rb") as f:
image_data = base64.b64encode(f.read()).decode("utf-8")
ext = path.lower().split(".")[-1]
mime_type = f"image/{'jpeg' if ext in ['jpg', 'jpeg'] else ext}"
content.append({
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{image_data}"
}
})
response = client.chat.completions.create(
model="grok-3-beta",
messages=[{"role": "user", "content": content}],
max_tokens=2048
)
return response.choices[0].message.content
5.3 文档 OCR 与结构化提取
from openai import OpenAI
import base64
import json
class DocumentExtractor:
"""文档结构化提取器。"""
def __init__(self, api_key: str):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
def extract_invoice(self, image_path: str) -> dict:
"""从发票图像中提取结构化数据。"""
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode("utf-8")
prompt = """请从这张发票图像中提取以下信息，以JSON格式返回：
{
"invoice_number": "发票号码",
"date": "开票日期",
"seller": "销售方名称",
"buyer": "购买方名称",
"items": [
{
"name": "商品名称",
"quantity": 数量,
"unit_price": 单价,
"total": 金额
}
],
"subtotal": 小计,
"tax": 税额,
"total": 总计
}"""
response = self.client.chat.completions.create(
model="grok-3-beta",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_data}"
}
}
]
}
],
temperature=0.1
)
result = response.choices[0].message.content
try:
json_start = result.find('{')
json_end = result.rfind('}') + 1
return json.loads(result[json_start:json_end])
except json.JSONDecodeError:
return {"raw_text": result}
def extract_table(self, image_path: str) -> list:
"""从图像中提取表格数据。"""
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode("utf-8")
prompt = "请识别图像中的表格，将其转换为JSON数组格式，每个元素为一行数据。"
response = self.client.chat.completions.create(
model="grok-3-beta",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_data}"
}
}
]
}
],
temperature=0.1
)
result = response.choices[0].message.content
try:
json_start = result.find('[')
json_end = result.rfind(']') + 1
return json.loads(result[json_start:json_end])
except json.JSONDecodeError:
return [{"raw_text": result}]
第六章：Function Calling 函数调用
6.1 Function Calling 基础
Function Calling 允许 Grok 模型调用你预定义的函数，从而实现与外部系统的交互。这是构建 AI Agent 的核心能力。
from openai import OpenAI
import json
# 定义可用函数
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的当前天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如'北京'、'上海'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "search_products",
"description": "搜索商品信息",
"parameters": {
"type": "object",
"properties": {
"keyword": {
"type": "string",
"description": "搜索关键词"
},
"category": {
"type": "string",
"description": "商品类别"
},
"max_price": {
"type": "number",
"description": "最高价格"
}
},
"required": ["keyword"]
}
}
}
]
# 模拟函数实现
def get_weather(city: str, unit: str = "celsius") -> dict:
"""获取天气信息（模拟）。"""
weather_data = {
"北京": {"temp": 22, "condition": "晴", "humidity": 45},
"上海": {"temp": 26, "condition": "多云", "humidity": 65},
"深圳": {"temp": 30, "condition": "阵雨", "humidity": 80},
}
data = weather_data.get(city, {"temp": 20, "condition": "未知", "humidity": 50})
if unit == "fahrenheit":
data["temp"] = data["temp"] * 9 / 5 + 32
return {"city": city, **data, "unit": unit}
def search_products(keyword: str, category: str = None,
max_price: float = None) -> list:
"""搜索商品（模拟）。"""
products = [
{"name": "MacBook Pro 14", "category": "电脑", "price": 14999},
{"name": "iPhone 16 Pro", "category": "手机", "price": 8999},
{"name": "AirPods Pro 3", "category": "耳机", "price": 1899},
{"name": "iPad Air", "category": "平板", "price": 4799},
]
results = [p for p in products if keyword.lower() in p["name"].lower()]
if category:
results = [p for p in results if p["category"] == category]
if max_price:
results = [p for p in results if p["price"] <= max_price]
return results
# 函数映射
function_map = {
"get_weather": get_weather,
"search_products": search_products,
}
def chat_with_functions(user_message: str) -> str:
"""带函数调用的对话。"""
client = OpenAI(
api_key="your-xai-api-key",
base_url="https://api.x.ai/v1"
)
messages = [
{"role": "system", "content": "你是一个智能助手，可以查询天气和搜索商品。"},
{"role": "user", "content": user_message}
]
# 第一次调用：让模型决定是否需要调用函数
response = client.chat.completions.create(
model="grok-3-beta",
messages=messages,
tools=tools,
tool_choice="auto"
)
assistant_message = response.choices[0].message
# 检查是否有函数调用
if assistant_message.tool_calls:
messages.append(assistant_message)
# 执行每个函数调用
for tool_call in assistant_message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"调用函数: {func_name}({func_args})")
# 执行函数
func = function_map.get(func_name)
if func:
result = func(**func_args)
else:
result = {"error": f"未知函数: {func_name}"}
# 将结果添加到消息中
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
# 第二次调用：让模型基于函数结果生成回复
final_response = client.chat.completions.create(
model="grok-3-beta",
messages=messages
)
return final_response.choices[0].message.content
else:
return assistant_message.content
# 使用示例
print(chat_with_functions("北京今天天气怎么样？"))
print(chat_with_functions("帮我找一下价格5000以下的苹果产品"))
6.2 并行函数调用
Grok 支持在一次请求中返回多个函数调用，实现并行处理：
def handle_parallel_calls(user_message: str) -> str:
"""处理并行函数调用。"""
client = OpenAI(
api_key="your-xai-api-key",
base_url="https://api.x.ai/v1"
)
messages = [
{"role": "system", "content": "你是一个智能助手。"},
{"role": "user", "content": user_message}
]
response = client.chat.completions.create(
model="grok-3-beta",
messages=messages,
tools=tools,
tool_choice="auto"
)
assistant_message = response.choices[0].message
if assistant_message.tool_calls:
messages.append(assistant_message)
# 并行执行所有函数调用
for tool_call in assistant_message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
func = function_map.get(func_name)
if func:
result = func(**func_args)
else:
result = {"error": f"未知函数: {func_name}"}
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
final_response = client.chat.completions.create(
model="grok-3-beta",
messages=messages
)
return final_response.choices[0].message.content
return assistant_message.content
# 模型可能会同时调用 get_weather 和 search_products
print(handle_parallel_calls("北京天气怎么样？顺便帮我看看有什么耳机推荐"))
6.3 构建完整的 Agent 框架
from openai import OpenAI
from typing import Callable, Dict, Any
import json
class GrokAgent:
"""基于 Grok Function Calling 的 Agent 框架。"""
def __init__(self, api_key: str, model: str = "grok-3-beta"):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
self.model = model
self.tools: list = []
self.functions: Dict[str, Callable] = {}
self.system_prompt = ""
def set_system_prompt(self, prompt: str):
"""设置系统提示词。"""
self.system_prompt = prompt
def register_function(self, name: str, description: str,
parameters: dict, func: Callable):
"""注册一个可调用函数。"""
self.tools.append({
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters
}
})
self.functions[name] = func
def run(self, user_message: str, max_rounds: int = 5) -> str:
"""运行 Agent，自动处理函数调用循环。"""
messages = []
if self.system_prompt:
messages.append({"role": "system", "content": self.system_prompt})
messages.append({"role": "user", "content": user_message})
for round_num in range(max_rounds):
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.tools if self.tools else None,
tool_choice="auto" if self.tools else None
)
assistant_message = response.choices[0].message
if not assistant_message.tool_calls:
return assistant_message.content
messages.append(assistant_message)
for tool_call in assistant_message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
func = self.functions.get(func_name)
try:
if func:
result = func(**func_args)
else:
result = {"error": f"未注册的函数: {func_name}"}
except Exception as e:
result = {"error": str(e)}
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False,
default=str)
})
return "达到最大推理轮数，未能生成最终回复。"
# 使用示例
if __name__ == "__main__":
agent = GrokAgent(api_key="your-xai-api-key")
agent.set_system_prompt("你是一个多功能智能助手。")
# 注册函数
agent.register_function(
name="get_weather",
description="获取天气",
parameters={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名"}
},
"required": ["city"]
},
func=lambda city: {"city": city, "temp": 25, "condition": "晴"}
)
result = agent.run("查一下上海的天气")
print(result)
第七章：Grok Studio 创意模式
7.1 什么是 Grok Studio
Grok Studio 是 xAI 推出的创意工作模式，专注于生成可交互的内容，包括：
- 网页应用：生成完整的 HTML/CSS/JS 单页应用
- 代码项目：创建可运行的代码项目
- 文档：生成结构化的 Markdown 文档
- 数据可视化：创建图表和仪表板
7.2 通过 API 生成交互式网页
from openai import OpenAI
def generate_web_app(api_key: str, description: str) -> str:
"""使用 Grok 生成交互式网页应用。"""
client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
prompt = f"""请根据以下描述生成一个完整的、可直接运行的单页HTML应用。
要求：
1. 所有代码写在一个HTML文件中（内联CSS和JavaScript）
2. 使用现代UI设计，视觉效果美观
3. 功能完整，可以直接在浏览器中运行
4. 响应式设计，适配不同屏幕尺寸
描述：{description}
请只输出HTML代码，不要包含其他解释。"""
response = client.chat.completions.create(
model="grok-3-beta",
messages=[
{"role": "system", "content": "你是一个专业的前端开发工程师，擅长创建精美的交互式网页应用。"},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=8192
)
return response.choices[0].message.content
# 生成一个计算器应用
html_code = generate_web_app(
api_key="your-xai-api-key",
description="一个科学计算器，支持基本运算、三角函数、对数运算，有深色主题"
)
# 保存为HTML文件
with open("calculator.html", "w", encoding="utf-8") as f:
f.write(html_code)
print("已生成 calculator.html")
7.3 生成数据可视化仪表板
def generate_dashboard(api_key: str, data_description: str) -> str:
"""生成数据可视化仪表板。"""
client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
prompt = f"""请创建一个数据可视化仪表板（单页HTML应用）。
数据描述：{data_description}
要求：
1. 使用 Chart.js 或 D3.js 进行数据可视化
2. 包含至少3种不同类型的图表（折线图、柱状图、饼图等）
3. 使用模拟数据，但数据要合理真实
4. 界面美观，使用卡片式布局
5. 支持响应式设计
6. 包含数据筛选和交互功能
请输出完整的HTML代码。"""
response = client.chat.completions.create(
model="grok-3-beta",
messages=[
{"role": "system", "content": "你是数据可视化专家，擅长创建美观实用的仪表板。"},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=8192
)
return response.choices[0].message.content
7.4 代码项目生成
def generate_project(api_key: str, project_description: str) -> dict:
"""生成完整的代码项目。"""
client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
prompt = f"""请根据以下描述生成一个完整的项目代码。
项目描述：{project_description}
请以JSON格式返回项目结构：
{{
"project_name": "项目名称",
"files": [
{{
"path": "文件路径",
"content": "文件内容"
}}
],
"setup_instructions": "安装和运行说明"
}}"""
response = client.chat.completions.create(
model="grok-3-beta",
messages=[
{"role": "system", "content": "你是一个全栈开发专家。请以JSON格式返回项目代码。"},
{"role": "user", "content": prompt}
],
temperature=0.5,
max_tokens=8192
)
result = response.choices[0].message.content
try:
json_start = result.find('{')
json_end = result.rfind('}') + 1
return json.loads(result[json_start:json_end])
except json.JSONDecodeError:
return {"raw_text": result}
第八章：DeepSearch 深度搜索
8.1 DeepSearch 概述
DeepSearch 是 xAI 基于 Grok-3 推出的智能搜索引擎，它能够：
- 从 X 平台和互联网抓取实时信息
- 进行多步推理和信息综合
- 生成带来源引用的深度研究报告
- 提供推理过程的透明展示
8.2 通过 API 使用 DeepSearch
from openai import OpenAI
def deep_search(api_key: str, query: str) -> str:
"""使用 DeepSearch 进行深度搜索。"""
client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
response = client.chat.completions.create(
model="grok-3-beta",
messages=[
{
"role": "system",
"content": """你是一个深度研究助手。在回答问题时：
1. 搜索多个来源获取最新信息
2. 交叉验证信息的准确性
3. 引用具体的信息来源
4. 提供全面而深入的分析
5. 标注信息的时效性"""
},
{"role": "user", "content": query}
],
temperature=0.3,
max_tokens=4096
)
return response.choices[0].message.content
# 使用示例
result = deep_search(
api_key="your-xai-api-key",
query="2025年全球半导体行业的最新发展趋势和主要厂商动态"
)
print(result)
8.3 构建研究助手
from openai import OpenAI
from dataclasses import dataclass
from typing import List
import json
@dataclass
class ResearchReport:
"""研究报告结构。"""
topic: str
summary: str
key_findings: List[str]
sources: List[str]
analysis: str
outlook: str
class DeepResearchAssistant:
"""深度研究助手。"""
def __init__(self, api_key: str):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
def research(self, topic: str, depth: str = "standard") -> ResearchReport:
"""执行深度研究。"""
depth_instructions = {
"quick": "提供简洁的概要，重点列出3-5个关键发现。",
"standard": "提供全面的分析，包括背景、现状、关键发现和趋势预测。",
"deep": "提供极其详尽的研究报告，包括多角度分析、数据支撑、风险评估和战略建议。"
}
prompt = f"""请对以下主题进行深度研究：
主题：{topic}
研究深度：{depth_instructions.get(depth, depth_instructions['standard'])}
请以JSON格式返回研究报告：
{{
"topic": "研究主题",
"summary": "200字以内的摘要",
"key_findings": ["发现1", "发现2", ...],
"sources": ["来源1", "来源2", ...],
"analysis": "详细分析（500-1000字）",
"outlook": "未来展望（200-300字）"
}}"""
response = self.client.chat.completions.create(
model="grok-3-beta",
messages=[
{
"role": "system",
"content": "你是一个专业的行业研究分析师，擅长深度研究和趋势分析。请基于最新信息进行分析，以JSON格式返回。"
},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=4096
)
result_text = response.choices[0].message.content
try:
json_start = result_text.find('{')
json_end = result_text.rfind('}') + 1
data = json.loads(result_text[json_start:json_end])
return ResearchReport(**data)
except (json.JSONDecodeError, TypeError):
return ResearchReport(
topic=topic,
summary=result_text[:200],
key_findings=[],
sources=[],
analysis=result_text,
outlook=""
)
def quick_qa(self, question: str) -> str:
"""快速问答。"""
response = self.client.chat.completions.create(
model="grok-3-mini-beta",
messages=[
{"role": "system", "content": "你是一个知识渊博的助手，请简洁准确地回答问题。"},
{"role": "user", "content": question}
],
temperature=0.3,
max_tokens=1024
)
return response.choices[0].message.content
# 使用示例
if __name__ == "__main__":
assistant = DeepResearchAssistant(api_key="your-xai-api-key")
# 深度研究
report = assistant.research("2025年大语言模型技术发展趋势", depth="standard")
print(f"主题: {report.topic}")
print(f"摘要: {report.summary}")
print(f"关键发现:")
for finding in report.key_findings:
print(f" - {finding}")
# 快速问答
answer = assistant.quick_qa("什么是 MoE 架构？")
print(answer)
8.4 DeepSearch 与传统搜索的对比
| 特性 | DeepSearch | 传统搜索引擎 | Perplexity |
|---|---|---|---|
| 实时数据 | ✅ X 平台 + 互联网 | ✅ 互联网 | ✅ 互联网 |
| 推理透明 | ✅ 展示推理链 | ❌ | 部分支持 |
| 多步推理 | ✅ 自动深入 | ❌ | ✅ |
| 来源引用 | ✅ | ✅ | ✅ |
| 情感分析 | ✅ 社交媒体情绪 | ❌ | ❌ |
| API 接入 | ✅ | 有限 | ✅ |
第九章：Grok 与 OpenAI / Anthropic 对比分析
9.1 模型能力对比
| 维度 | Grok-3 | GPT-4o | Claude 3.5 Sonnet |
|---|---|---|---|
| 数学推理 (AIME) | 93.3% | ~79% | - |
| 编码 (LiveCodeBench) | 79.4% | 72.9% | - |
| 上下文窗口 | 131K | 128K | 200K |
| 实时数据 | ✅ X 平台集成 | ❌ | ❌ |
| 图像理解 | ✅ | ✅ | ✅ |
| Function Calling | ✅ | ✅ | ✅ |
| 推理透明 (Think) | ✅ | 部分 | 部分 |
| API 起步价 | $0.40/M tokens | $2.50/M tokens | $3.00/M tokens |
9.2 API 兼容性对比
# 统一接口适配器示例
from openai import OpenAI
from typing import Optional
class UnifiedLLMClient:
"""统一的 LLM 客户端，支持多个提供商。"""
PROVIDERS = {
"grok": {
"base_url": "https://api.x.ai/v1",
"default_model": "grok-3-mini-beta"
},
"openai": {
"base_url": "https://api.openai.com/v1",
"default_model": "gpt-4o"
},
"anthropic": {
"base_url": "https://api.anthropic.com/v1",
"default_model": "claude-3-5-sonnet-20241022"
}
}
def __init__(self, provider: str, api_key: str,
model: Optional[str] = None):
config = self.PROVIDERS.get(provider)
if not config:
raise ValueError(f"不支持的提供商: {provider}")
self.client = OpenAI(
api_key=api_key,
base_url=config["base_url"]
)
self.model = model or config["default_model"]
def chat(self, messages: list, **kwargs) -> str:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
**kwargs
)
return response.choices[0].message.content
# 使用示例
grok_client = UnifiedLLMClient("grok", "your-xai-key")
openai_client = UnifiedLLMClient("openai", "your-openai-key")
question = [{"role": "user", "content": "解释量子计算的基本原理"}]
grok_answer = grok_client.chat(question)
openai_answer = openai_client.chat(question)
9.3 场景选择建议
选择 Grok 的场景：
- 需要实时社交媒体数据分析
- 预算敏感的大规模应用
- 需要推理透明度的科研场景
- 长文档处理和分析
选择 GPT-4o 的场景：
- 需要最广泛的插件生态
- 企业级合规要求较高
- 多语言支持要求极高
选择 Claude 的场景：
- 超长文档处理（200K 上下文）
- 代码生成和分析
- 严格的安全和对齐要求
第十章：企业级集成方案
10.1 架构设计
企业级 Grok 应用通常采用以下架构：
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 前端应用 │────▶│ API 网关 │────▶│ Grok API │
└─────────────┘ └──────┬──────┘ └─────────────┘
│
┌──────┴──────┐
│ 业务服务 │
├─────────────┤
│ 缓存层 │
├─────────────┤
│ 数据库 │
└─────────────┘
10.2 企业级 API 网关
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from openai import OpenAI
from typing import List, Optional
import hashlib
import json
import time
from functools import lru_cache
app = FastAPI(title="Grok Enterprise Gateway")
# 配置
class Settings:
GROK_API_KEY = "your-xai-api-key"
GROK_BASE_URL = "https://api.x.ai/v1"
DEFAULT_MODEL = "grok-3-mini-beta"
MAX_TOKENS = 4096
RATE_LIMIT_PER_MINUTE = 60
settings = Settings()
# 请求/响应模型
class ChatRequest(BaseModel):
message: str
model: str = settings.DEFAULT_MODEL
temperature: float = 0.7
max_tokens: int = settings.MAX_TOKENS
system_prompt: Optional[str] = None
use_cache: bool = True
class ChatResponse(BaseModel):
reply: str
model: str
tokens_used: int
cached: bool
latency_ms: float
# 简单缓存实现
class ResponseCache:
def __init__(self, ttl_seconds: int = 3600):
self.cache = {}
self.ttl = ttl_seconds
def _make_key(self, request: ChatRequest) -> str:
data = f"{request.message}:{request.model}:{request.temperature}"
return hashlib.md5(data.encode()).hexdigest()
def get(self, request: ChatRequest) -> Optional[str]:
key = self._make_key(request)
if key in self.cache:
entry = self.cache[key]
if time.time() - entry["time"] < self.ttl:
return entry["value"]
del self.cache[key]
return None
def set(self, request: ChatRequest, response: str):
key = self._make_key(request)
self.cache[key] = {"value": response, "time": time.time()}
cache = ResponseCache()
# Grok 客户端
@lru_cache()
def get_grok_client() -> OpenAI:
return OpenAI(
api_key=settings.GROK_API_KEY,
base_url=settings.GROK_BASE_URL
)
# API 端点
@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
start_time = time.time()
# 检查缓存
if request.use_cache:
cached = cache.get(request)
if cached:
return ChatResponse(
reply=cached,
model=request.model,
tokens_used=0,
cached=True,
latency_ms=(time.time() - start_time) * 1000
)
# 调用 Grok API
client = get_grok_client()
messages = []
if request.system_prompt:
messages.append({"role": "system", "content": request.system_prompt})
messages.append({"role": "user", "content": request.message})
try:
response = client.chat.completions.create(
model=request.model,
messages=messages,
temperature=request.temperature,
max_tokens=request.max_tokens
)
reply = response.choices[0].message.content
tokens = response.usage.total_tokens if response.usage else 0
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# 更新缓存
if request.use_cache:
cache.set(request, reply)
return ChatResponse(
reply=reply,
model=request.model,
tokens_used=tokens,
cached=False,
latency_ms=(time.time() - start_time) * 1000
)
@app.get("/api/models")
async def list_models():
return {
"models": [
{"id": "grok-3-beta", "description": "旗舰模型"},
{"id": "grok-3-fast-beta", "description": "快速旗舰模型"},
{"id": "grok-3-mini-beta", "description": "轻量模型"},
{"id": "grok-3-mini-fast-beta", "description": "超快轻量模型"},
]
}
# 启动: uvicorn main:app --host 0.0.0.0 --port 8000
10.3 批量处理管道
from openai import OpenAI
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict, Any
import json
import time
class BatchProcessor:
"""Grok API 批量处理器。"""
def __init__(self, api_key: str, model: str = "grok-3-mini-beta",
max_workers: int = 5, requests_per_minute: int = 60):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
self.model = model
self.max_workers = max_workers
self.rpm = requests_per_minute
self.request_interval = 60.0 / requests_per_minute
def process_single(self, task: Dict[str, Any]) -> Dict[str, Any]:
"""处理单个任务。"""
start = time.time()
try:
messages = task.get("messages", [])
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=task.get("temperature", 0.7),
max_tokens=task.get("max_tokens", 1024)
)
return {
"task_id": task.get("id"),
"status": "success",
"result": response.choices[0].message.content,
"tokens": response.usage.total_tokens if response.usage else 0,
"latency_ms": (time.time() - start) * 1000
}
except Exception as e:
return {
"task_id": task.get("id"),
"status": "error",
"error": str(e),
"latency_ms": (time.time() - start) * 1000
}
def process_batch(self, tasks: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""批量处理任务。"""
results = []
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {}
for i, task in enumerate(tasks):
future = executor.submit(self.process_single, task)
futures[future] = i
# 控制提交速率
if (i + 1) % self.max_workers == 0:
time.sleep(self.request_interval)
for future in as_completed(futures):
result = future.result()
results.append(result)
print(f"完成任务 {result['task_id']}: {result['status']}")
return results
# 使用示例
if __name__ == "__main__":
processor = BatchProcessor(api_key="your-xai-api-key")
tasks = [
{
"id": f"task_{i}",
"messages": [
{"role": "user", "content": f"用一句话总结以下概念：{topic}"}
]
}
for i, topic in enumerate([
"量子计算", "区块链", "边缘计算", "联邦学习", "知识图谱"
])
]
results = processor.process_batch(tasks)
for r in results:
print(f"[{r['task_id']}] {r.get('result', r.get('error'))}")
10.4 监控与日志
import logging
import time
from functools import wraps
from typing import Callable
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class APIMetrics:
"""API 使用指标。"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_tokens: int = 0
total_latency_ms: float = 0
errors: list = field(default_factory=list)
@property
def success_rate(self) -> float:
if self.total_requests == 0:
return 0
return self.successful_requests / self.total_requests
@property
def avg_latency_ms(self) -> float:
if self.successful_requests == 0:
return 0
return self.total_latency_ms / self.successful_requests
class GrokMonitor:
"""Grok API 监控器。"""
def __init__(self):
self.metrics = APIMetrics()
self.logger = logging.getLogger("grok_monitor")
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
def track(self, func: Callable) -> Callable:
"""装饰器：追踪 API 调用指标。"""
@wraps(func)
def wrapper(*args, **kwargs):
self.metrics.total_requests += 1
start = time.time()
try:
result = func(*args, **kwargs)
latency = (time.time() - start) * 1000
self.metrics.successful_requests += 1
self.metrics.total_latency_ms += latency
self.logger.info(
f"API调用成功 | 延迟: {latency:.1f}ms"
)
return result
except Exception as e:
self.metrics.failed_requests += 1
self.metrics.errors.append({
"time": datetime.now().isoformat(),
"error": str(e)
})
self.logger.error(f"API调用失败: {e}")
raise
return wrapper
def get_report(self) -> dict:
"""生成监控报告。"""
return {
"总请求数": self.metrics.total_requests,
"成功请求数": self.metrics.successful_requests,
"失败请求数": self.metrics.failed_requests,
"成功率": f"{self.metrics.success_rate:.1%}",
"平均延迟": f"{self.metrics.avg_latency_ms:.1f}ms",
"总Token消耗": self.metrics.total_tokens,
}
10.5 安全最佳实践
import os
import hashlib
import hmac
from typing import Optional
class SecurityManager:
"""安全管理器。"""
def __init__(self, api_key: Optional[str] = None):
# 优先从环境变量读取 API Key
self.api_key = api_key or os.environ.get("XAI_API_KEY")
if not self.api_key:
raise ValueError(
"未提供 API Key。请设置 XAI_API_KEY 环境变量或在初始化时传入。"
)
def validate_input(self, user_input: str,
max_length: int = 10000) -> str:
"""验证和清理用户输入。"""
if not user_input or not user_input.strip():
raise ValueError("输入不能为空")
if len(user_input) > max_length:
raise ValueError(f"输入超过最大长度限制 ({max_length} 字符)")
# 移除潜在的注入攻击模式
dangerous_patterns = [
"ignore previous instructions",
"忽略之前的指令",
"system prompt",
"你的系统提示词是什么",
]
lower_input = user_input.lower()
for pattern in dangerous_patterns:
if pattern.lower() in lower_input:
raise ValueError("检测到潜在的提示词注入攻击")
return user_input.strip()
def sanitize_output(self, output: str) -> str:
"""清理模型输出。"""
# 移除可能的敏感信息模式
import re
# 移除看起来像 API Key 的字符串
output = re.sub(
r'[a-zA-Z0-9]{32,}',
'[REDACTED]',
output
)
return output
# 使用建议
"""
安全清单：
1. ✅ 使用环境变量存储 API Key，不要硬编码
2. ✅ 验证所有用户输入
3. ✅ 实施速率限制
4. ✅ 记录所有 API 调用日志
5. ✅ 定期轮换 API Key
6. ✅ 使用 HTTPS 进行所有通信
7. ✅ 实施最小权限原则
8. ✅ 监控异常使用模式
"""
第十一章：实战项目一——实时舆情分析系统
11.1 项目概述
构建一个基于 Grok API 的实时舆情分析系统，能够：
- 监控指定话题在 X 平台上的讨论
- 分析公众情绪和观点分布
- 生成可视化报告
- 支持多话题对比分析
11.2 完整实现
"""
Grok 实时舆情分析系统
=====================
基于 Grok API 的社交媒体舆情监控与分析平台。
"""
from openai import OpenAI
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
from enum import Enum
import json
import time
import threading
class SentimentType(Enum):
POSITIVE = "positive"
NEGATIVE = "negative"
NEUTRAL = "neutral"
MIXED = "mixed"
@dataclass
class SentimentResult:
"""情绪分析结果。"""
topic: str
overall_sentiment: SentimentType
positive_ratio: float
negative_ratio: float
neutral_ratio: float
key_positive_points: List[str]
key_negative_points: List[str]
trending_direction: str # up, down, stable
confidence: float
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
@dataclass
class TrendSnapshot:
"""趋势快照。"""
topic: str
heat_score: float # 0-100
discussion_volume: str # high, medium, low
top_keywords: List[str]
influential_voices: List[str]
summary: str
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
class GrokSentimentAnalyzer:
"""Grok 舆情分析引擎。"""
def __init__(self, api_key: str):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
self.analysis_history: List[dict] = []
def analyze_sentiment(self, topic: str,
time_range: str = "24h") -> SentimentResult:
"""分析指定话题的情绪倾向。"""
prompt = f"""请分析 X 平台上关于"{topic}"的最新讨论情绪（最近{time_range}）。
请以JSON格式返回分析结果：
{{
"overall_sentiment": "positive/negative/neutral/mixed",
"positive_ratio": 0.0到1.0之间的正面情绪比例,
"negative_ratio": 0.0到1.0之间的负面情绪比例,
"neutral_ratio": 0.0到1.0之间的中性情绪比例,
"key_positive_points": ["正面观点1", "正面观点2", ...],
"key_negative_points": ["负面观点1", "负面观点2", ...],
"trending_direction": "up/down/stable",
"confidence": 0.0到1.0之间的置信度
}}"""
response = self.client.chat.completions.create(
model="grok-3-beta",
messages=[
{
"role": "system",
"content": "你是一个专业的社交媒体情绪分析师。基于X平台的最新数据进行分析，以JSON格式返回结果。"
},
{"role": "user", "content": prompt}
],
temperature=0.2,
max_tokens=1024
)
result_text = response.choices[0].message.content
try:
json_start = result_text.find('{')
json_end = result_text.rfind('}') + 1
data = json.loads(result_text[json_start:json_end])
result = SentimentResult(topic=topic, **data)
except (json.JSONDecodeError, TypeError):
result = SentimentResult(
topic=topic,
overall_sentiment=SentimentType.MIXED,
positive_ratio=0.33,
negative_ratio=0.33,
neutral_ratio=0.34,
key_positive_points=[],
key_negative_points=[],
trending_direction="stable",
confidence=0.5
)
self.analysis_history.append({
"type": "sentiment",
"result": result.__dict__,
"timestamp": datetime.now().isoformat()
})
return result
def get_trend_snapshot(self, topic: str) -> TrendSnapshot:
"""获取话题趋势快照。"""
prompt = f"""请分析"{topic}"在X平台上的当前讨论趋势。
请以JSON格式返回：
{{
"heat_score": 0到100的热度评分,
"discussion_volume": "high/medium/low",
"top_keywords": ["关键词1", "关键词2", ...],
"influential_voices": ["账号1", "账号2", ...],
"summary": "100字以内的趋势摘要"
}}"""
response = self.client.chat.completions.create(
model="grok-3-mini-beta",
messages=[
{
"role": "system",
"content": "你是社交媒体趋势分析师。以JSON格式返回分析结果。"
},
{"role": "user", "content": prompt}
],
temperature=0.2
)
result_text = response.choices[0].message.content
try:
json_start = result_text.find('{')
json_end = result_text.rfind('}') + 1
data = json.loads(result_text[json_start:json_end])
return TrendSnapshot(topic=topic, **data)
except (json.JSONDecodeError, TypeError):
return TrendSnapshot(
topic=topic,
heat_score=50,
discussion_volume="medium",
top_keywords=[],
influential_voices=[],
summary="无法获取趋势数据"
)
def compare_topics(self, topics: List[str]) -> dict:
"""对比多个话题的舆情。"""
comparisons = {}
for topic in topics:
sentiment = self.analyze_sentiment(topic)
trend = self.get_trend_snapshot(topic)
comparisons[topic] = {
"sentiment": sentiment.__dict__,
"trend": trend.__dict__
}
time.sleep(1) # 避免速率限制
return comparisons
def generate_report(self, topic: str) -> str:
"""生成完整的舆情分析报告。"""
sentiment = self.analyze_sentiment(topic)
trend = self.get_trend_snapshot(topic)
report = f"""
# 舆情分析报告：{topic}
生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
## 一、情绪分析
- **整体情绪**: {sentiment.overall_sentiment.value}
- **正面比例**: {sentiment.positive_ratio:.1%}
- **负面比例**: {sentiment.negative_ratio:.1%}
- **中性比例**: {sentiment.neutral_ratio:.1%}
- **趋势方向**: {sentiment.trending_direction}
- **分析置信度**: {sentiment.confidence:.1%}
### 主要正面观点
"""
for point in sentiment.key_positive_points:
report += f"- {point}\n"
report += "\n### 主要负面观点\n"
for point in sentiment.key_negative_points:
report += f"- {point}\n"
report += f"""
## 二、趋势分析
- **热度评分**: {trend.heat_score}/100
- **讨论量级**: {trend.discussion_volume}
- **关键词**: {', '.join(trend.top_keywords)}
- **活跃账号**: {', '.join(trend.influential_voices)}
### 趋势摘要
{trend.summary}
"""
return report
# 主程序
if __name__ == "__main__":
analyzer = GrokSentimentAnalyzer(api_key="your-xai-api-key")
# 分析单个话题
result = analyzer.animate_sentiment("人工智能", time_range="7d")
print(f"情绪: {result.overall_sentiment.value}")
print(f"正面: {result.positive_ratio:.1%}")
print(f"负面: {result.negative_ratio:.1%}")
# 生成完整报告
report = analyzer.generate_report("Grok-3 发布")
print(report)
# 多话题对比
comparison = analyzer.compare_topics(["Grok", "GPT-4", "Claude"])
for topic, data in comparison.items():
print(f"\n{topic}:")
print(f" 情绪: {data['sentiment']['overall_sentiment']}")
print(f" 热度: {data['trend']['heat_score']}")
11.3 数据持久化与可视化
import json
import csv
from datetime import datetime
from typing import List
class SentimentDataStore:
"""舆情数据持久化存储。"""
def __init__(self, data_dir: str = "./sentiment_data"):
self.data_dir = data_dir
import os
os.makedirs(data_dir, exist_ok=True)
def save_to_json(self, results: List[dict], filename: str = None):
"""保存分析结果到 JSON 文件。"""
if not filename:
filename = f"sentiment_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
filepath = f"{self.data_dir}/{filename}"
with open(filepath, "w", encoding="utf-8") as f:
json.dump(results, f, ensure_ascii=False, indent=2)
return filepath
def save_to_csv(self, results: List[dict], filename: str = None):
"""保存分析结果到 CSV 文件。"""
if not filename:
filename = f"sentiment_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
filepath = f"{self.data_dir}/{filename}"
if not results:
return filepath
keys = results[0].keys()
with open(filepath, "w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=keys)
writer.writeheader()
writer.writerows(results)
return filepath
def generate_html_chart(self, data: dict, output_path: str = "report.html"):
"""生成 HTML 可视化报告。"""
html = f"""<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="UTF-8">
<title>舆情分析报告</title>
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
<style>
body {{ font-family: 'Microsoft YaHei', sans-serif; margin: 20px; background: #f5f5f5; }}
.card {{ background: white; border-radius: 12px; padding: 24px; margin: 16px 0; box-shadow: 0 2px 8px rgba(0,0,0,0.1); }}
h1 {{ color: #1a1a2e; }}
h2 {{ color: #16213e; border-bottom: 2px solid #0f3460; padding-bottom: 8px; }}
.metric {{ display: inline-block; margin: 10px 20px; text-align: center; }}
.metric-value {{ font-size: 2em; font-weight: bold; color: #0f3460; }}
.metric-label {{ color: #666; }}
.chart-container {{ max-width: 600px; margin: 20px auto; }}
</style>
</head>
<body>
<h1>📊 舆情分析报告</h1>
<p>生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
<div class="card">
<h2>情绪分布</h2>
<div class="chart-container">
<canvas id="sentimentChart"></canvas>
</div>
</div>
<script>
const ctx = document.getElementById('sentimentChart').getContext('2d');
new Chart(ctx, {{
type: 'doughnut',
data: {{
labels: ['正面', '负面', '中性'],
datasets: [{{
data: [{data.get('positive', 33)}, {data.get('negative', 33)}, {data.get('neutral', 34)}],
backgroundColor: ['#4CAF50', '#f44336', '#9E9E9E']
}}]
}}
}});
</script>
</body>
</html>"""
with open(output_path, "w", encoding="utf-8") as f:
f.write(html)
return output_path
第十二章：实战项目二——智能 AI 助手
12.1 项目概述
构建一个功能完整的智能 AI 助手，集成以下能力：
- 多轮对话管理
- Function Calling 工具调用
- 实时信息查询
- 文件分析处理
- 任务规划与执行
12.2 完整实现
"""
Grok 智能 AI 助手
================
基于 Grok API 的多功能智能助手，支持工具调用、实时数据和多模态分析。
"""
from openai import OpenAI
from typing import List, Dict, Callable, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import json
import os
import base64
class MessageRole(Enum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"
TOOL = "tool"
@dataclass
class ConversationMessage:
"""对话消息。"""
role: MessageRole
content: str
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
metadata: dict = field(default_factory=dict)
class ToolRegistry:
"""工具注册表。"""
def __init__(self):
self.tools: List[dict] = []
self.functions: Dict[str, Callable] = {}
def register(self, name: str, description: str,
parameters: dict, func: Callable):
"""注册一个工具。"""
self.tools.append({
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters
}
})
self.functions[name] = func
def execute(self, name: str, arguments: dict) -> Any:
"""执行一个工具。"""
func = self.functions.get(name)
if not func:
return {"error": f"未注册的工具: {name}"}
try:
return func(**arguments)
except Exception as e:
return {"error": str(e)}
class GrokAssistant:
"""Grok 智能助手。"""
def __init__(self, api_key: str, model: str = "grok-3-beta"):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.x.ai/v1"
)
self.model = model
self.tool_registry = ToolRegistry()
self.conversation: List[Dict[str, str]] = []
self.max_history = 50
self._register_default_tools()
def _register_default_tools(self):
"""注册默认工具集。"""
# 获取当前时间
self.tool_registry.register(
name="get_current_time",
description="获取当前日期和时间",
parameters={"type": "object", "properties": {}, "required": []},
func=lambda: {"datetime": datetime.now().isoformat(),
"formatted": datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")}
)
# 数学计算
self.tool_registry.register(
name="calculate",
description="执行数学计算",
parameters={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如 '2 + 3 * 4'"
}
},
"required": ["expression"]
},
func=lambda expression: {"result": eval(expression)}
)
# 文本统计
self.tool_registry.register(
name="text_stats",
description="统计文本的字数、行数等信息",
parameters={
"type": "object",
"properties": {
"text": {"type": "string", "description": "要统计的文本"}
},
"required": ["text"]
},
func=lambda text: {
"characters": len(text),
"words": len(text.split()),
"lines": text.count('\n') + 1,
"chinese_chars": sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
}
)
def set_system_prompt(self, prompt: str):
"""设置系统提示词。"""
self.conversation = [{"role": "system", "content": prompt}]
def chat(self, user_message: str) -> str:
"""与助手对话。"""
self.conversation.append({"role": "user", "content": user_message})
self._trim_history()
# 第一次调用
response = self.client.chat.completions.create(
model=self.model,
messages=self.conversation,
tools=self.tool_registry.tools if self.tool_registry.tools else None,
tool_choice="auto" if self.tool_registry.tools else None,
temperature=0.7,
max_tokens=4096
)
assistant_msg = response.choices[0].message
# 处理工具调用
if assistant_msg.tool_calls:
self.conversation.append(assistant_msg.model_dump())
for tool_call in assistant_msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f" 🔧 调用工具: {func_name}({func_args})")
result = self.tool_registry.execute(func_name, func_args)
self.conversation.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False, default=str)
})
# 第二次调用生成最终回复
final_response = self.client.chat.completions.create(
model=self.model,
messages=self.conversation,
temperature=0.7,
max_tokens=4096
)
reply = final_response.choices[0].message.content
else:
reply = assistant_msg.content
self.conversation.append({"role": "assistant", "content": reply})
return reply
def chat_stream(self, user_message: str):
"""流式对话。"""
self.conversation.append({"role": "user", "content": user_message})
self._trim_history()
stream = self.client.chat.completions.create(
model=self.model,
messages=self.conversation,
stream=True,
temperature=0.7,
max_tokens=4096
)
full_response = ""
for chunk in stream:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_response += content
yield content
self.conversation.append({"role": "assistant", "content": full_response})
def analyze_image(self, image_path: str, question: str) -> str:
"""分析图像。"""
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode("utf-8")
ext = image_path.lower().split(".")[-1]
mime = f"image/{'jpeg' if ext in ['jpg', 'jpeg'] else ext}"
self.conversation.append({
"role": "user",
"content": [
{"type": "text", "text": question},
{"type": "image_url",
"image_url": {"url": f"data:{mime};base64,{image_data}"}}
]
})
response = self.client.chat.completions.create(
model=self.model,
messages=self.conversation,
max_tokens=2048
)
reply = response.choices[0].message.content
self.conversation.append({"role": "assistant", "content": reply})
return reply
def clear_conversation(self):
"""清空对话历史。"""
system_msg = self.conversation[0] if self.conversation and self.conversation[0]["role"] == "system" else None
self.conversation = [system_msg] if system_msg else []
def get_conversation_summary(self) -> dict:
"""获取对话摘要。"""
return {
"total_messages": len(self.conversation),
"user_messages": sum(1 for m in self.conversation if m.get("role") == "user"),
"assistant_messages": sum(1 for m in self.conversation if m.get("role") == "assistant"),
"tool_calls": sum(1 for m in self.conversation if m.get("role") == "tool"),
}
def _trim_history(self):
"""裁剪对话历史以控制 token 消耗。"""
if len(self.conversation) > self.max_history:
system = self.conversation[0] if self.conversation[0]["role"] == "system" else None
recent = self.conversation[-(self.max_history - 1):]
self.conversation = ([system] if system else []) + recent
def export_conversation(self, filepath: str):
"""导出对话记录。"""
with open(filepath, "w", encoding="utf-8") as f:
json.dump(self.conversation, f, ensure_ascii=False, indent=2, default=str)
# 使用示例
if __name__ == "__main__":
assistant = GrokAssistant(api_key="your-xai-api-key")
assistant.set_system_prompt("""你是一个智能AI助手，名叫Grok小助手。
你的能力包括：
1. 回答各种知识问题
2. 查询当前时间
3. 执行数学计算
4. 分析文本信息
5. 分析图像内容
请用友好、专业的语气回答问题。遇到不确定的信息时，请诚实说明。""")
# 交互式对话
print("Grok 智能助手已启动！输入 'quit' 退出。\n")
while True:
user_input = input("你: ").strip()
if user_input.lower() in ["quit", "exit", "退出"]:
print("再见！")
break
if not user_input:
continue
reply = assistant.chat(user_input)
print(f"助手: {reply}\n")
12.3 添加自定义工具
import requests
def setup_advanced_tools(assistant: GrokAssistant):
"""为助手添加高级工具。"""
# 文件读取工具
assistant.tool_registry.register(
name="read_file",
description="读取本地文件内容",
parameters={
"type": "object",
"properties": {
"filepath": {"type": "string", "description": "文件路径"}
},
"required": ["filepath"]
},
func=lambda filepath: {
"content": open(filepath, "r", encoding="utf-8").read()
if os.path.exists(filepath) else "文件不存在"
}
)
# JSON 分析工具
assistant.tool_registry.register(
name="analyze_json",
description="分析JSON数据的结构和内容",
parameters={
"type": "object",
"properties": {
"json_str": {"type": "string", "description": "JSON字符串"}
},
"required": ["json_str"]
},
func=lambda json_str: {
"parsed": json.loads(json_str),
"keys": list(json.loads(json_str).keys()) if isinstance(json.loads(json_str), dict) else "非对象类型",
"depth": _json_depth(json.loads(json_str))
}
)
# Markdown 生成工具
assistant.tool_registry.register(
name="generate_markdown",
description="生成格式化的Markdown文档",
parameters={
"type": "object",
"properties": {
"title": {"type": "string", "description": "文档标题"},
"sections": {
"type": "array",
"items": {
"type": "object",
"properties": {
"heading": {"type": "string"},
"content": {"type": "string"}
}
},
"description": "章节列表"
}
},
"required": ["title", "sections"]
},
func=lambda title, sections: {
"markdown": f"# {title}\n\n" + "\n\n".join(
f"## {s['heading']}\n\n{s['content']}" for s in sections
)
}
)
def _json_depth(obj, depth=0):
"""计算 JSON 对象的深度。"""
if isinstance(obj, dict):
return max((_json_depth(v, depth + 1) for v in obj.values()), default=depth)
elif isinstance(obj, list):
return max((_json_depth(item, depth + 1) for item in obj), default=depth)
return depth
第十三章：常见问题与排错指南
13.1 API 连接问题
问题：连接超时或无法访问 API
# 解决方案：添加超时和重试配置
from openai import OpenAI
client = OpenAI(
api_key="your-xai-api-key",
base_url="https://api.x.ai/v1",
timeout=30.0, # 30秒超时
max_retries=3 # 最多重试3次
)
问题：SSL 证书错误
# 解决方案：检查系统时间或更新证书
import ssl
import openai
# 如果在企业网络中，可能需要配置代理
client = OpenAI(
api_key="your-xai-api-key",
base_url="https://api.x.ai/v1",
http_client=httpx.Client(verify=True) # 或设置为 False（不推荐）
)
13.2 认证与权限问题
问题：401 Unauthorized
- 检查 API Key 是否正确
- 确认 API Key 是否已过期
- 验证账户是否有足够额度
import os
# 使用环境变量管理 API Key
api_key = os.environ.get("XAI_API_KEY")
if not api_key:
raise ValueError("请设置 XAI_API_KEY 环境变量")
问题：403 Forbidden
- 检查是否有访问特定模型的权限
- 确认是否需要 X Premium+ 或 SuperGrok 订阅
13.3 速率限制问题
问题：429 Too Many Requests
import time
from openai import OpenAI, RateLimitError
client = OpenAI(api_key="your-xai-api-key", base_url="https://api.x.ai/v1")
def call_with_backoff(func, *args, max_retries=5, **kwargs):
"""带指数退避的重试机制。"""
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except RateLimitError:
if attempt == max_retries - 1:
raise
wait_time = 2 ** attempt
print(f"速率限制，等待 {wait_time} 秒...")
time.sleep(wait_time)
# 使用
response = call_with_backoff(
client.chat.completions.create,
model="grok-3-mini-beta",
messages=[{"role": "user", "content": "你好"}]
)
13.4 Token 与上下文问题
问题：超出上下文窗口限制
def estimate_tokens(text: str) -> int:
"""粗略估算 token 数量。"""
# 中文约1.5-2字符/token，英文约4字符/token
chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
other_chars = len(text) - chinese_chars
return int(chinese_chars / 1.5 + other_chars / 4)
def trim_messages(messages: list, max_tokens: int = 100000) -> list:
"""裁剪消息以适应上下文窗口。"""
total_tokens = sum(estimate_tokens(m.get("content", "")) for m in messages)
if total_tokens <= max_tokens:
return messages
# 保留系统消息和最近的消息
system_msgs = [m for m in messages if m.get("role") == "system"]
other_msgs = [m for m in messages if m.get("role") != "system"]
while other_msgs and total_tokens > max_tokens:
removed = other_msgs.pop(0)
total_tokens -= estimate_tokens(removed.get("content", ""))
return system_msgs + other_msgs
13.5 输出质量问题
问题：模型输出不准确或不稳定
# 降低温度以获得更稳定的输出
response = client.chat.completions.create(
model="grok-3-beta",
messages=[...],
temperature=0.1, # 接近0时输出最确定
top_p=0.9
)
# 使用更强的提示词约束
system_prompt = """你是一个专业的数据分析师。
规则：
1. 只基于提供的数据进行分析，不要编造数据
2. 所有数字必须精确到小数点后两位
3. 必须引用具体的数据来源
4. 如果信息不足，请明确说明"""
问题：JSON 输出格式错误
def get_structured_output(client, prompt: str, schema: dict) -> dict:
"""获取结构化的 JSON 输出。"""
full_prompt = f"""{prompt}
请严格按照以下JSON格式返回，不要包含任何其他文字：
{json.dumps(schema, ensure_ascii=False, indent=2)}"""
response = client.chat.completions.create(
model="grok-3-beta",
messages=[{"role": "user", "content": full_prompt}],
temperature=0.1
)
result = response.choices[0].message.content
# 尝试提取 JSON
try:
# 尝试直接解析
return json.loads(result)
except json.JSONDecodeError:
# 尝试提取被包裹的 JSON
import re
json_match = re.search(r'```(?:json)?\s*([\s\S]*?)```', result)
if json_match:
return json.loads(json_match.group(1))
# 尝试找到 { } 或 [ ]
for start_char, end_char in [('{', '}'), ('[', ']')]:
start = result.find(start_char)
end = result.rfind(end_char)
if start != -1 and end != -1:
try:
return json.loads(result[start:end + 1])
except json.JSONDecodeError:
continue
raise ValueError(f"无法从输出中提取JSON: {result[:200]}")
13.6 多模态问题
问题：图像分析返回错误
# 确保图像格式和大小正确
def prepare_image(image_path: str, max_size_mb: int = 10) -> str:
"""准备图像用于 API 调用。"""
import os
file_size = os.path.getsize(image_path) / (1024 * 1024)
if file_size > max_size_mb:
# 压缩图像
from PIL import Image
img = Image.open(image_path)
# 调整大小
ratio = (max_size_mb / file_size) ** 0.5
new_size = (int(img.width * ratio), int(img.height * ratio))
img = img.resize(new_size, Image.LANCZOS)
# 保存为 JPEG
compressed_path = image_path.rsplit('.', 1)[0] + '_compressed.jpg'
img.save(compressed_path, 'JPEG', quality=85)
image_path = compressed_path
with open(image_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
13.7 性能优化建议
| 优化方向 | 具体措施 | 预期效果 |
|---|---|---|
| 模型选择 | 简单任务用 mini 版本 | 成本降低 5-10 倍 |
| 缓存 | 相同查询使用缓存结果 | 延迟降低 90%+ |
| 批量处理 | 合并多个请求 | 吞吐量提升 2-5 倍 |
| 流式输出 | 使用 stream=True | 首字延迟降低 80% |
| 提示词优化 | 精简系统提示词 | Token 消耗降低 30% |
| 异步调用 | 使用 asyncio + httpx | 并发性能提升 5-10 倍 |
# 异步调用示例
import asyncio
from openai import AsyncOpenAI
async def async_chat(prompt: str) -> str:
client = AsyncOpenAI(
api_key="your-xai-api-key",
base_url="https://api.x.ai/v1"
)
response = await client.chat.completions.create(
model="grok-3-mini-beta",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
async def batch_async(prompts: list) -> list:
"""并发执行多个请求。"""
tasks = [async_chat(p) for p in prompts]
return await asyncio.gather(*tasks)
# 运行
results = asyncio.run(batch_async(["问题1", "问题2", "问题3"]))
附录：资源汇总
官方资源
| 资源 | 链接 |
|---|---|
| xAI 官网 | https://x.ai |
| API 控制台 | https://console.x.ai |
| API 文档 | https://docs.x.ai |
| X 平台 | https://x.com |
SDK 与工具
| 工具 | 说明 |
|---|---|
| OpenAI Python SDK | 通过兼容接口调用 Grok API |
| xAI SDK | xAI 官方 Python SDK |
| LangChain | 支持 Grok 的 AI 应用框架 |
| LlamaIndex | 支持 Grok 的数据框架 |
定价速查表
| 模型 | 输入 (\(/M tokens) | 输出 (\)/M tokens) | 适用场景 | |
|---|---|---|---|
| grok-3-beta | $2.00 | $8.00 | 复杂推理、研究 |
| grok-3-fast-beta | $2.50 | $10.00 | 低延迟交互 |
| grok-3-mini-beta | $0.40 | $1.60 | 通用任务、批量处理 |
| grok-3-mini-fast-beta | $0.50 | $2.00 | 高吞吐量任务 |
速率限制参考
| 账户类型 | grok-3-beta | grok-3-mini-beta |
|---|---|---|
| 免费试用 | 3 RPM | 10 RPM |
| 付费账户 | 100 RPM | 100 RPM |
| X Premium+ | 优先响应 | 优先响应 |
RPM = Requests Per Minute（每分钟请求数）
学习路径建议
入门阶段（第 1-2 天）
- 阅读本教程第 1-3 章
- 完成第一个 API 调用
- 熟悉基本参数和错误处理
进阶阶段（第 3-5 天）
- 学习 Function Calling（第 6 章）
- 掌握多模态能力（第 5 章）
- 实现简单的 AI 助手
实战阶段（第 6-10 天）
- 完成舆情分析系统（第 11 章）
- 构建智能助手（第 12 章）
- 探索 DeepSearch 和实时数据
企业级阶段（第 11-15 天）
- 设计企业级架构（第 10 章）
- 实现监控和安全机制
- 优化性能和成本
本教程编写时间：2025 年
适用 API 版本：xAI API v1
字数统计：约 12,000+ 字
版权声明：本教程为原创技术文档，可自由用于学习和参考。
