AI 应用开发实战

// ========== 文件：chat_http.mjs ==========
// 运行：OPENAI_API_KEY=sk-xxx node chat_http.mjs
const KEY = process.env.OPENAI_API_KEY;
const URL = 'https://api.openai.com/v1/chat/completions';
// 换成 DashScope: 'https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions'
// 换成硅基流动: 'https://api.siliconflow.cn/v1/chat/completions'

async function chat(messages, { model = 'gpt-4o-mini', temperature = 0.7, stream = false } = {}) {
  const resp = await fetch(URL, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${KEY}` },
    body: JSON.stringify({ model, messages, temperature, stream }),
  });
  if (!resp.ok) throw new Error(`HTTP ${resp.status}: ${await resp.text()}`);
  return (await resp.json()).choices[0].message.content;
}
console.log(await chat([{ role: 'user', content: '一句话介绍 Node.js fetch' }]));

# ========== 文件：chat_openai.py ==========
# 依赖：uv add openai python-dotenv
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()

client = OpenAI(api_key=os.environ['OPENAI_API_KEY'])

# ========== 方式 A：chat.completions（传统，老项目大多这个）==========
resp = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[
        {'role': 'system', 'content': '你是简洁的助手'},
        {'role': 'user',   'content': '一句话介绍 Python'},
    ],
    temperature=0.7,
    max_tokens=200,
)
print('chat.completions:', resp.choices[0].message.content)
print('tokens:', resp.usage.total_tokens)

# ========== 方式 B：responses（OpenAI 2025+ 主推，新 API）==========
# 优势：内置 conversation 状态、内置 file/web_search 工具
resp2 = client.responses.create(
    model='gpt-4.1-mini',
    input='用一句话介绍 Python',
)
print('responses:', resp2.output_text)

# ========== 文件：chat_claude.py ==========
# 依赖：uv add anthropic python-dotenv
import os
import anthropic
from dotenv import load_dotenv
load_dotenv()

client = anthropic.Anthropic(api_key=os.environ['ANTHROPIC_API_KEY'])

# ========== 同步调用 ==========
resp = client.messages.create(
    model='claude-sonnet-5-*',    # 或 claude-opus-4-6 / claude-3-5-haiku-latest
    max_tokens=1024,
    system='你是简洁的助手，回答控制在 50 字内。',
    messages=[{'role': 'user', 'content': '一句话介绍 Python async/await'}],
)
print('Claude:', resp.content[0].text)
print('usage:', resp.usage.input_tokens, resp.usage.output_tokens)

# ========== 流式 ===========
print('\n--- 流式 ---')
with client.messages.stream(
    model='claude-sonnet-5-*',
    max_tokens=512,
    messages=[{'role': 'user', 'content': '写一首关于 Python 的俳句'}],
) as stream:
    for text in stream.text_stream:
        print(text, end='', flush=True)
print()

# ========== 多轮 ===========
history = []
def ask(q):
    history.append({'role': 'user', 'content': q})
    r = client.messages.create(
        model='claude-sonnet-5-*', max_tokens=1024,
        system='你是 Python 老师', messages=history,
    )
    history.append({'role': 'assistant', 'content': r.content[0].text})
    return r.content[0].text

print('\nQ1:', ask('闭包是什么？'))
print('Q2:', ask('那和装饰器啥关系？'))

# ========== 文件：chat_gemini.py ==========
# 依赖：uv add google-genai python-dotenv
import os
from google import genai
from google.genai import types
from dotenv import load_dotenv
load_dotenv()

client = genai.Client(api_key=os.environ['GOOGLE_API_KEY'])

# ========== 同步 ==========
resp = client.models.generate_content(
    model='gemini-2.5-flash',     # 便宜 + 快；强用 'gemini-2.5-pro'
    contents='用一句话介绍 Python',
    config=types.GenerateContentConfig(
        max_output_tokens=300,
        temperature=0.7,
        system_instruction='你是简洁的中文助手',
    ),
)
print('Gemini:', resp.text)
print('usage:', resp.usage_metadata)

# ========== 流式 ===========
print('\n--- 流式 ---')
for chunk in client.models.generate_content_stream(
    model='gemini-2.5-flash', contents='讲个冷笑话',
):
    print(chunk.text, end='', flush=True)
print()

# ========== 多轮（用 chats.create 自动维护 history）==========
chat = client.chats.create(
    model='gemini-2.5-flash',
    config=types.GenerateContentConfig(system_instruction='Python 老师'),
)
print('\nA1:', chat.send_message('闭包是啥？').text)
print('A2:', chat.send_message('那和装饰器的关系？').text)

# ========== 文件：chat_dashscope.py ==========
# 依赖：uv add openai dashscope python-dotenv
import os
from openai import OpenAI
from dashscope import Generation
from dotenv import load_dotenv
load_dotenv()

# ========== 方式 A：OpenAI 兼容模式（最简）==========
# 换 base_url 就走国内直连
ds = OpenAI(
    api_key=os.environ['DASHSCOPE_API_KEY'],
    base_url='https://dashscope.aliyuncs.com/compatible-mode/v1',
)
resp = ds.chat.completions.create(
    model='qwen-plus',        # 或 'qwen-max' / 'qwen3-max' / 'qwen-coder-plus'
    messages=[{'role': 'user', 'content': '一句话介绍 Python async'}],
)
print('DashScope (OpenAI 模式):', resp.choices[0].message.content)

# ========== 方式 B：DashScope 原生 SDK（用到 qwen-vl/qwen-audio/qwen-long 时）==========
r = Generation.call(
    model='qwen-plus',
    messages=[{'role': 'user', 'content': '一句话介绍 Rust'}],
    result_format='message',   # 一定要加，默认是 text
)
print('DashScope (native):', r.output.choices[0].message.content)
print('usage:', r.usage)

# ========== 文件：llm_client.py ==========
# 依赖：uv add openai anthropic google-genai python-dotenv pydantic tenacity
import os
from typing import Iterator
from openai import OpenAI
from anthropic import Anthropic
from google import genai
from google.genai import types as gtypes
from pydantic import BaseModel
from dotenv import load_dotenv
load_dotenv()


class LLMConfig(BaseModel):
    """统一配置"""
    provider: str            # 'openai' | 'anthropic' | 'gemini' | 'dashscope' | 'siliconflow' | 'deepseek'
    model: str               # 模型 id
    api_key: str             # api key（从环境变量读）
    base_url: str | None = None
    system: str | None = None
    temperature: float = 0.7
    max_tokens: int = 2048

    @classmethod
    def from_env(cls, provider: str, model: str, **kw):
        """从 .env 自动读 key + base_url"""
        env_map = {
            'openai':      ('OPENAI_API_KEY',     'https://api.openai.com/v1'),
            'dashscope':   ('DASHSCOPE_API_KEY',  'https://dashscope.aliyuncs.com/compatible-mode/v1'),
            'siliconflow': ('SILICONFLOW_API_KEY','https://api.siliconflow.cn/v1'),
            'deepseek':    ('DEEPSEEK_API_KEY',   'https://api.deepseek.com/v1'),
        }
        env_key, default_url = env_map.get(provider, (None, None))
        return cls(
            provider=provider, model=model,
            api_key=os.environ[env_key] if env_key else os.environ.get('API_KEY', ''),
            base_url=default_url, **kw,
        )


class UnifiedLLM:
    """统一调用入口，业务代码只认这一个类"""

    def __init__(self, cfg: LLMConfig):
        self.cfg = cfg
        self.provider = cfg.provider
        if cfg.provider in ('openai', 'dashscope', 'siliconflow', 'deepseek'):
            # OpenAI 兼容协议：都用 openai SDK
            self._openai = OpenAI(api_key=cfg.api_key, base_url=cfg.base_url)
        elif cfg.provider == 'anthropic':
            self._anthropic = Anthropic(api_key=cfg.api_key)
        elif cfg.provider == 'gemini':
            self._gemini = genai.Client(api_key=cfg.api_key)
        else:
            raise ValueError(f'unknown provider {cfg.provider}')

    def chat(self, messages: list[dict], **override) -> str:
        """同步调用，返回文本"""
        cfg = self.cfg.model_copy(update=override) if override else self.cfg
        if self.provider in ('openai', 'dashscope', 'siliconflow', 'deepseek'):
            kwargs = self._kwargs(cfg)
            kwargs['messages'] = [{'role': m['role'], 'content': m['content']} for m in messages]
            return self._openai.chat.completions.create(**kwargs).choices[0].message.content
        if self.provider == 'anthropic':
            kwargs = self._kwargs(cfg)
            sys = kwargs.pop('system', None)
            return self._anthropic.messages.create(
                model=kwargs['model'], system=sys or ' ',
                messages=kwargs['messages'], **kwargs,
            ).content[0].text
        if self.provider == 'gemini':
            prompt = self._gemini_messages_to_prompt(messages)
            return self._gemini.models.generate_content(
                model=cfg.model, contents=prompt,
                config=gtypes.GenerateContentConfig(
                    system_instruction=cfg.system, temperature=cfg.temperature,
                    max_output_tokens=cfg.max_tokens,
                ),
            ).text
        raise RuntimeError('unreachable')

    def stream(self, messages: list[dict], **override) -> Iterator[str]:
        """流式，返回增量文本"""
        cfg = self.cfg.model_copy(update=override) if override else self.cfg
        if self.provider in ('openai', 'dashscope', 'siliconflow', 'deepseek'):
            kwargs = self._kwargs(cfg)
            kwargs['messages'] = [{'role': m['role'], 'content': m['content']} for m in messages]
            kwargs['stream'] = True
            for chunk in self._openai.chat.completions.create(**kwargs):
                d = chunk.choices[0].delta.content
                if d: yield d
        elif self.provider == 'anthropic':
            kwargs = self._kwargs(cfg)
            sys = kwargs.pop('system', None)
            with self._anthropic.messages.stream(
                model=kwargs['model'], system=sys or ' ',
                messages=kwargs['messages'], **kwargs,
            ) as s:
                for t in s.text_stream: yield t
        elif self.provider == 'gemini':
            prompt = self._gemini_messages_to_prompt(messages)
            for chunk in self._gemini.models.generate_content_stream(
                model=cfg.model, contents=prompt,
                config=gtypes.GenerateContentConfig(
                    system_instruction=cfg.system, temperature=cfg.temperature,
                    max_output_tokens=cfg.max_tokens,
                ),
            ):
                if chunk.text: yield chunk.text

    # ========== 内部辅助 ==========
    def _kwargs(self, cfg):
        return {
            'model': cfg.model,
            'temperature': cfg.temperature,
            'max_tokens': cfg.max_tokens,
            'system': cfg.system,
        }

    @staticmethod
    def _gemini_messages_to_prompt(messages):
        # 业务侧 messages 都被自动转成一段文本
        parts = []
        for m in messages:
            if m['role'] == 'system':  continue  # 放 system_instruction
            parts.append(f"{m['role'].upper()}: {m['content']}")
        return '\n\n'.join(parts)


# ========== 使用示例 ===========
if __name__ == '__main__':
    # 业务侧完全无感知
    llm = UnifiedLLM(LLMConfig.from_env('dashscope', 'qwen-plus'))
    print('Q:', '一句话介绍 Python async/await')
    print('A:', llm.chat([
        {'role': 'user', 'content': '一句话介绍 Python async/await'}
    ]))

    print('\n--- 流式 ---')
    for chunk in llm.stream([{'role': 'user', 'content': '用 3 行讲清楚闭包'}]):
        print(chunk, end='', flush=True)
    print()

# ========== 文件：llm_router.py ==========
# 在 llm_client.py 基础上加：路由 + 降级
import logging
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import APIError, RateLimitError, APITimeoutError
import anthropic

log = logging.getLogger(__name__)

# ========== 路由规则 ==========
# 业务场景 -> 主模型/备模型/降级模型（按优先级）
ROUTES = {
    'simple_qa':        ('dashscope',   'qwen-turbo',    'siliconflow', 'Qwen/Qwen2.5-7B-Instruct', 'gemini', 'gemini-2.5-flash'),
    'coding':           ('anthropic',   'claude-sonnet-5', 'openai',   'gpt-4.1',            'dashscope', 'qwen-coder-plus'),
    'long_document':    ('gemini',      'gemini-3.5-pro', 'openai',    'gpt-4.1',            'dashscope', 'qwen-long'),
    'reasoning':        ('openai',      'o3',            'anthropic', 'claude-opus-4-6',    'dashscope', 'qwq-plus'),
}


class LLMRouter:
    def __init__(self, route_name: str):
        self.chain = ROUTES[route_name]   # 平铺 provider/model 三元组
        self._clients = {}

    def _get(self, provider: str, model: str) -> UnifiedLLM:
        key = (provider, model)
        if key not in self._clients:
            cfg = LLMConfig.from_env(provider, model)
            self._clients[key] = UnifiedLLM(cfg)
        return self._clients[key]

    @retry(
        stop=stop_after_attempt(3),                       # 最多 3 次
        wait=wait_exponential(multiplier=1, min=1, max=10), # 1s/2s/4s 退避
        retry=retry_if_exception_type((RateLimitError, APITimeoutError)),
        reraise=True,
    )
    def chat(self, messages: list[dict], **override) -> str:
        # 依次尝试主/备/降级模型
        last_err = None
        for i in range(0, len(self.chain), 2):
            provider, model = self.chain[i], self.chain[i+1]
            llm = self._get(provider, model)
            try:
                return llm.chat(messages, **override)
            except (APIError, anthropic.APIError) as e:
                log.warning('model %s/%s failed: %s', provider, model, e)
                last_err = e
                continue
        raise RuntimeError(f'all models failed: {last_err}')

    def stream(self, messages: list[dict], **override):
        # 流式不重试：直接用主模型
        provider, model = self.chain[0], self.chain[1]
        yield from self._get(provider, model).stream(messages, **override)


# ========== 使用 ===========
if __name__ == '__main__':
    r = LLMRouter('coding')
    print(r.chat([{'role': 'user', 'content': '写个 Python 装饰器，做 retry with backoff'}]))

# ========== 文件：usage_tracker.py ==========
# 业务逻辑：每次调用记录 token，然后按你最新一次手动填的价格表算钱。
# 价格表存到 config/pricing.yaml 而不是硬编码，每次定时从厂商定价页拉。
import json, time
from pathlib import Path

# 这只是模板。当前真实价格请按下面链接去查：
PRICING_URLS = {
    'openai':     'https://platform.openai.com/docs/pricing',
    'anthropic':  'https://docs.anthropic.com/en/docs/about-claude/pricing',
    'gemini':     'https://ai.google.dev/gemini-api/docs/pricing',
    'dashscope':  'https://help.aliyun.com/zh/model-studio/getting-started/models',  # 国内
    'deepseek':   'https://api-docs.deepseek.com/quick_start/pricing',
    'siliconflow':'https://cloud.siliconflow.cn/siliconCloud/model-square',
}

def calc_cost(provider: str, model: str, in_tok: int, out_tok: int, prices: dict) -> float:
    p = prices.get((provider, model), {'in': 1.0, 'out': 3.0})  # 兜底不报销
    return in_tok * p['in'] / 1_000_000 + out_tok * p['out'] / 1_000_000

class UsageTracker:
    def __init__(self, log_path='usage.jsonl'):
        self.log_path = Path(log_path)
        self.today = {'in': 0, 'out': 0, 'cost': 0.0, 'calls': 0}

    def record(self, provider: str, model: str, in_tok: int, out_tok: int,
               prices: dict, user: str = 'anon'):
        cost = calc_cost(provider, model, in_tok, out_tok, prices)
        rec = {
            'ts': time.time(), 'provider': provider, 'model': model,
            'in': in_tok, 'out': out_tok, 'cost': round(cost, 6), 'user': user,
        }
        self.log_path.open('a').write(json.dumps(rec, ensure_ascii=False) + '\n')
        self.today['in'] += in_tok; self.today['out'] += out_tok
        self.today['cost'] += cost; self.today['calls'] += 1
        return rec

# ========== 用法（注意 prices 是外部注入，不是硬编码）==========
# 业务方拉取最新一份 yaml（CI 每日更新）→ 传给 record
# tracker.record('openai', 'gpt-4o-mini', 100, 200, prices=load_pricing())

# ========== 文件：structured_output.py ==========
# 依赖：uv add openai pydantic
import os
from openai import OpenAI
from pydantic import BaseModel
from dotenv import load_dotenv
load_dotenv()

client = OpenAI(api_key=os.environ['OPENAI_API_KEY'])

# ========== OpenAI：用 Pydantic 自动生成 Schema ==========
class Person(BaseModel):
    name: str
    age: int
    skills: list[str]
    is_student: bool

resp = client.responses.parse(
    model='gpt-4.1-mini',
    input='解析：张三，25 岁，会 Python 和 Rust，不是学生',
    text_format=Person,           # 自动转成 JSON Schema + 校验
)
person: Person = resp.output_parsed
print(person.model_dump_json(indent=2))
# 必然按 schema 输出，类型不对就报错

# ========== 老接口 chat.completions 也支持 ==========
import json
schema = Person.model_json_schema()
resp2 = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': '解析：张三，25 岁...'}],
    response_format={
        'type': 'json_schema',
        'json_schema': {
            'name': 'person',
            'schema': schema,
            'strict': True,   # 强制符合 schema
        },
    },
)
data = json.loads(resp2.choices[0].message.content)
Person.model_validate(data)   # 再次校验

# ========== 文件：function_calling.py ==========
# 依赖：uv add openai python-dotenv
import os, json
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()

client = OpenAI(api_key=os.environ['OPENAI_API_KEY'])

# ========== 1. 定义工具 schema ==========
tools = [{
    'type': 'function',
    'function': {
        'name': 'get_weather',
        'description': '查询某城市当前天气',
        'parameters': {
            'type': 'object',
            'properties': {
                'city': {'type': 'string', 'description': '城市名，如"北京"'},
                'unit': {'type': 'string', 'enum': ['celsius', 'fahrenheit']},
            },
            'required': ['city'],
        },
    },
}]

def get_weather(city: str, unit: str = 'celsius') -> dict:
    return {'city': city, 'temp': 22, 'unit': unit, 'cond': '晴'}

# ========== 2. 第一轮：让模型决定 ==========
messages = [{'role': 'user', 'content': '北京今天多少度？要华氏度'}]
resp = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=messages,
    tools=tools,
)
assist_msg = resp.choices[0].message

# ========== 3. 如果决定调用，就执行后回传 ==========
if assist_msg.tool_calls:
    messages.append(assist_msg)   # 把 assistant 的 tool_calls 一起回传
    for tc in assist_msg.tool_calls:
        args = json.loads(tc.function.arguments)
        result = get_weather(**args)
        messages.append({
            'role': 'tool',
            'tool_call_id': tc.id,
            'content': json.dumps(result),
        })

    # ========== 4. 第二轮：拿工具结果生成最终回答 ==========
    final = client.chat.completions.create(model='gpt-4o-mini', messages=messages)
    print(final.choices[0].message.content)
else:
    print(assist_msg.content)

# ========== Anthropic Prompt Cache ==========
import anthropic
client = anthropic.Anthropic()

# 在 system 里声明 cache breakpoint；messages 里也可以加 cache_control: {'type': 'ephemeral'}
system_block = {
    'type': 'text',
    'text': '你是一个大型项目的代码助手...（很长很长的大文档）' * 200,
    'cache_control': {'type': 'ephemeral'},   # 标记缓存点
}

resp = client.messages.create(
    model='claude-sonnet-5-*',
    max_tokens=1024,
    system=[system_block],   # 必须传 list 才能用 cache_control
    messages=[{'role': 'user', 'content': '解释其中的 xx 模块'}],
)

# 用 usage 看是否命中
print('input_tokens:', resp.usage.input_tokens)
print('cache_creation_input_tokens:', resp.usage.cache_creation_input_tokens or 0)
print('cache_read_input_tokens:', resp.usage.cache_read_input_tokens or 0)
# 第一次会 cache_creation，下次就 cache_read（便宜 90%）

# ========== OpenAI 自动 cache（不需要特殊声明）==========
# OpenAI 自动检测 1024 token 以上的相同前缀，超过的部分自动按 cache 价格算
# 不用写代码，确保 system + messages 前缀稳定即可

# ========== 文件：vision.py ==========
# 依赖：uv add openai anthropic google-genai python-dotenv
import os, base64
from openai import OpenAI
from anthropic import Anthropic
from google import genai
from google.genai import types as gtypes
from dotenv import load_dotenv
load_dotenv()

# ========== 准备图片（base64）==========
with open('photo.jpg', 'rb') as f:
    img_b64 = base64.b64encode(f.read()).decode()

# ========== OpenAI ==========
oa = OpenAI(api_key=os.environ['OPENAI_API_KEY'])
r1 = oa.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{
        'role': 'user',
        'content': [
            {'type': 'text', 'text': '这张图里有什么？'},
            {'type': 'image_url', 'image_url': {'url': f'data:image/jpeg;base64,{img_b64}'}},
        ],
    }],
)
print('OpenAI:', r1.choices[0].message.content)

# ========== Anthropic Claude ==========
ac = Anthropic(api_key=os.environ['ANTHROPIC_API_KEY'])
r2 = ac.messages.create(
    model='claude-sonnet-5-*',
    max_tokens=1024,
    messages=[{
        'role': 'user',
        'content': [
            {'type': 'image', 'source': {'type': 'base64', 'media_type': 'image/jpeg', 'data': img_b64}},
            {'type': 'text', 'text': '用中文描述这张图'},
        ],
    }],
)
print('Claude:', r2.content[0].text)

# ========== Gemini ==========
gc = genai.Client(api_key=os.environ['GOOGLE_API_KEY'])
with open('photo.jpg', 'rb') as f:
    img_bytes = f.read()
r3 = gc.models.generate_content(
    model='gemini-2.5-flash',
    contents=[
        gtypes.Part.from_bytes(data=img_bytes, mime_type='image/jpeg'),
        'describe this image in one sentence',
    ],
)
print('Gemini:', r3.text)

# ========== 模式 1：角色 + 任务 + 格式 + 约束 ==========
你是一个[角色]。
请帮我[完成任务]。
输出格式：[JSON / Markdown / 表格]
约束条件：[字数 / 语言 / 不要 X]

# ========== 模式 2：Few-shot（给例子学习）==========
任务：将中文翻译成英文（适合业务邮件风格）

示例 1：
输入：您好，请问贵司产品报价？ → 输出：Hello, could you share your product quote?

示例 2：
输入：感谢配合，期待合作 → 输出：Thanks for your cooperation. Looking forward to working with you.

现在翻译：{{user_input}}

# ========== 模式 3：思维链 CoT ==========
[问题]
请分步骤思考：先 XXX，再 XXX，最后 XXX。
若涉及计算，请把每一步算式列出。
最后给出最终答案。

# ========== 模式 4：分隔符 + 反注入 ==========
文档内容用 --- 包裹：
---
{{user_document}}
---

请基于以上文档回答用户问题。如果文档不包含答案，直接说"超出文档范围"，不要编造。
用户问题：{{user_query}}

# ========== 模式 5：Anthropic 推荐的 XML 风格 ==========
（在 system prompt 里用 XML 标签，对 Claude 特别有效）

你是资深 Python 后端

{{background}}


- 回答简洁，控制在 100 字内
- 用中文
- 不要给出未经文档支持的猜测


{{user_query}}

# ========== 文件：server.py（FastAPI 后端）==========
# 依赖：uv add fastapi uvicorn openai pydantic
import os
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from openai import OpenAI
from pydantic import BaseModel
from dotenv import load_dotenv
load_dotenv()

app = FastAPI()
app.add_middleware(CORSMiddleware, allow_origins=['*'], allow_methods=['*'], allow_headers=['*'])

oa = OpenAI(api_key=os.environ['OPENAI_API_KEY'])

class ChatReq(BaseModel):
    messages: list   # [{role, content}]
    model: str = 'gpt-4o-mini'
    stream: bool = True

@app.post('/api/chat')
async def chat(req: ChatReq):
    if not req.stream:
        r = oa.chat.completions.create(model=req.model, messages=req.messages)
        return {'content': r.choices[0].message.content}
    # 流式
    def gen():
        s = oa.chat.completions.create(model=req.model, messages=req.messages, stream=True)
        for c in s:
            d = c.choices[0].delta.content
            if d: yield f'data: {repr(d)}\n\n'
        yield 'data: [DONE]\n\n'
    return StreamingResponse(gen(), media_type='text/event-stream')

# 运行：uv run uvicorn server:app --reload --port 8000

# ========== 文件：rag.py（LlamaIndex 2026 写法）==========
# 依赖：uv add llama-index llama-index-llms-dashscope llama-index-embeddings-dashscope
#       chromadb  （或 qdrant-client / lancedb）
import os
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.dashscope import DashScope
from llama_index.embeddings.dashscope import DashScopeEmbedding
from dotenv import load_dotenv
load_dotenv()

# 设置 LLM 和 embedding（用 DashScope 国内直连）
Settings.llm = DashScope(model='qwen-plus', api_key=os.environ['DASHSCOPE_API_KEY'])
Settings.embed_model = DashScopeEmbedding(model='text-embedding-v3', api_key=os.environ['DASHSCOPE_API_KEY'])

# 加载 ./docs 下所有 txt/md/pdf
docs = SimpleDirectoryReader('./docs', recursive=True).load_data()
print(f'加载 {len(docs)} 个文档')

# 建立索引（自动切块 + embedding + 存入内存）
index = VectorStoreIndex.from_documents(docs)

# 问答
qe = index.as_query_engine(similarity_top_k=5)
resp = qe.query('总结这些文档关于 XX 的要点')
print(resp)
for sn in resp.source_nodes:
    print(f'  - 引用: {sn.node.metadata.get("file_name", "?")} (score={sn.score:.3f})')

# ========== 文件：simple_agent.py ==========
# 依赖：uv add openai python-dotenv tenacity
import os, json
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()

oa = OpenAI(api_key=os.environ['OPENAI_API_KEY'])

# ========== 工具集（手写 ReAct，不依赖框架最快理解原理）==========
def search(query: str) -> str:
    """模拟搜索"""
    return f"[搜索结果] 关于 {query} 的信息：...（真实项目接 Serper / Bocha / Tavily）"

def calc(expr: str) -> str:
    """算术计算"""
    try: return str(eval(expr))
    except: return '计算失败'

TOOLS = {
    'search': {'fn': search, 'desc': '搜索实时信息'},
    'calc':   {'fn': calc,   'desc': '执行数学计算'},
}

def ask_agent(question: str, max_steps=5) -> str:
    history = [
        {'role': 'system', 'content': '你是智能体。要回答用户问题，必要时调用工具。\n工具：\n' +
            '\n'.join(f'- {name}: {cfg["desc"]}' for name, cfg in TOOLS.items()) +
            '\n调用格式：{"action": "name", "input": "..."}\n否则直接给答案。'},
        {'role': 'user', 'content': question},
    ]
    for step in range(max_steps):
        r = oa.chat.completions.create(model='gpt-4o-mini', messages=history, temperature=0)
        ans = r.choices[0].message.content.strip()
        history.append({'role': 'assistant', 'content': ans})

        if ans.startswith('{') and '"action"' in ans:
            try:
                call = json.loads(ans)
                tool = TOOLS[call['action']]
                result = tool['fn'](call['input'])
                history.append({'role': 'user', 'content': f'Observation: {result}'})
            except Exception as e:
                history.append({'role': 'user', 'content': f'调用失败：{e}'})
        else:
            return ans
    return history[-1]['content']

print(ask_agent('搜索 AI 应用开发的最新趋势，然后用一句话回答'))

# ========== 文件：mcp_server.py（自己写一个 MCP 服务）==========
# 依赖：uv add mcp
# 这个服务会被 Claude Desktop / Claude Code 拉起，给模型暴露工具

import asyncio
from mcp.server.fastmcp import FastMCP

mcp = FastMCP('my-tools')

@mcp.tool()
def get_weather(city: str) -> str:
    """查询某城市天气（给 LLM 的工具）"""
    return f'{city} 当前 22 度，晴'

@mcp.tool()
def add(a: int, b: int) -> int:
    """两个整数相加"""
    return a + b

@mcp.resource('config://app')
def app_config() -> str:
    """应用配置（给 LLM 的资源）"""
    return 'version: 1.0, status: ok'

if __name__ == '__main__':
    mcp.run()   # 默认 stdio 传输，Claude Desktop 用 stdio 拉起
# 运行：uv run mcp_server.py

# ========== Claude Desktop 集成（~/Library/Application Support/Claude/claude_desktop_config.json）==========
{
  "mcpServers": {
    "my-tools": {
      "command": "uv",
      "args": ["--directory", "/path/to/project", "run", "mcp_server.py"]
    }
  }
}

# 1. 用 curl 先排除环境问题
curl -sS https://api.openai.com/v1/chat/completions \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}],"max_tokens":10}' \
  | jq .

# 国内走 DashScope（无代理）
curl -sS https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions \
  -H "Authorization: Bearer $DASHSCOPE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"qwen-plus","messages":[{"role":"user","content":"hi"}]}' \
  | jq .

# 国内走 Gemini
curl -sS "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key=$GOOGLE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"contents":[{"parts":[{"text":"hi"}]}]}' | jq .

# ========== 安装 ==========
# 1. Node.js ≥ 18
node -v

# 2. 装 Claude Code
npm install -g @anthropic-ai/claude-code

# 3. 配置 API Key
export ANTHROPIC_API_KEY=sk-ant-...

# ========== 使用 ==========
cd your-project/
claude                     # 启动交互模式
> 给 readme.md 加一个「快速开始」章节

# 自动读代码 → 自动编辑 → 自动跑测试

# ========== 推荐实践 ==========
# 1. 把 CLAUDE.md 放项目根，描述项目约定
cat > CLAUDE.md << 'EOF'
# Project Rules
- 用 uv 管理依赖
- Python 3.12
- 测试用 pytest
- 代码风格 ruff
EOF

# 2. 用 .claude/settings.json 配置权限
mkdir -p .claude
cat > .claude/settings.json << 'EOF'
{
  "permissions": {
    "allow": ["Bash(npm test:*)", "Read"],
    "deny": ["Bash(rm -rf:*)", "Bash(curl:*)"]
  }
}
EOF

# 3. 在 CI 里调用
echo 'auto-merge simple test fixes if tests pass' | claude --ci

# ========== 1. 安装 ==========
# 推荐用 conda 建一个新环境
conda create -n unsloth python=3.11 -y && conda activate unsloth
pip install unsloth vllm trl datasets

# 检查 GPU
nvidia-smi | head -12
# 至少要 16G 显存（Qwen3-8B） / 24G（Qwen3-32B-AWQ）

# ========== 文件：dpo_train.py ==========
# 依赖：uv add unsloth trl
from unsloth import FastLanguageModel, PatchDPOTrainer
PatchDPOTrainer()
from trl import DPOTrainer, DPOConfig

model, tokenizer = FastLanguageModel.from_pretrained('Qwen/Qwen3-8B', max_seq_length=2048, load_in_4bit=True)
model = FastLanguageModel.get_peft_model(model, r=16, lora_alpha=32)

# 数据格式：chosen / rejected 一对对的回答
dataset = load_dataset('json', data_files='dpo_data.jsonl', split='train')
# dpo_data.jsonl: {"prompt":"...", "chosen":"好回答", "rejected":"差回答"}

DPOConfig_kwargs = dict(
    output_dir='./dpo_outputs',
    beta=0.1,                       # KL 系数，越大越保守
    per_device_train_batch_size=2,
    gradient_accumulation_steps=4,
    learning_rate=5e-5,
    num_train_epochs=1,
    bf16=True,
    logging_steps=10,
)
trainer = DPOTrainer(model=model, tokenizer=tokenizer, train_dataset=dataset,
                     args=DPOConfig(**DPOConfig_kwargs))
trainer.train()
model.save_pretrained_merged('dpo_merged', tokenizer, save_method='merged_16bit')

# ========== 安装 ==========
# macOS / Linux
curl -fsSL https://ollama.com/install.sh | sh

# Windows：去 https://ollama.com 下载安装

# ========== 拉模型 ==========
ollama pull qwen3:8b                 # 通用对话
ollama pull qwen3:32b                # 需要 ≥ 24G 显存
ollama pull deepseek-r1:8b           # 推理专精
ollama pull llama4:scout             # 超长上下文
ollama pull nomic-embed-text         # embedding 模型

# ========== 命令行玩 ==========
ollama run qwen3:8b "用 3 句话介绍 Python async"

# ========== OpenAI 兼容 HTTP API（自动监听 11434）==========
# 任何 OpenAI SDK 都能直接调 Ollama
curl http://localhost:11434/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model":"qwen3:8b","messages":[{"role":"user","content":"hi"}]}' | jq .

# 用 Python 调用
python -c "
from openai import OpenAI
c = OpenAI(base_url='http://localhost:11434/v1', api_key='ollama')
print(c.chat.completions.create(model='qwen3:8b', messages=[{'role':'user','content':'hi'}]).choices[0].message.content)
"

# ========== 自定义模型（用 GGUF + Modelfile）==========
# 把第 8 章合并出来的模型导出 GGUF，然后用 ollama create
cat > Modelfile << 'EOF'
FROM ./qwen3-8b-finetuned-q4_k_m.gguf
PARAMETER temperature 0.7
PARAMETER num_ctx 8192
SYSTEM "你是 helpful 中文助手"
EOF
ollama create my-model -f Modelfile
ollama run my-model

# ========== 安装 ==========
# 官方推荐用 venv 隔离 + 装匹配 CUDA 版本
python3.12 -m venv vllm-env
source vllm-env/bin/activate
pip install --upgrade pip
pip install vllm                   # 自动选当前机器 CUDA 对应版本

# ========== 启动 vLLM 服务（新版命令）==========
vllm serve Qwen/Qwen3-8B-Instruct \
  --host 0.0.0.0 --port 8000 \
  --gpu-memory-utilization 0.9 \
  --max-model-len 8192 \
  --served-model-name qwen3-8b      # 客户端用的别名

# 多卡 Tensor Parallel
vllm serve meta-llama/Llama-3.3-70B-Instruct \
  --host 0.0.0.0 --port 8000 \
  --tensor-parallel-size 4 \
  --max-model-len 16384

# ========== 测试（OpenAI 兼容）==========
curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model":"qwen3-8b","messages":[{"role":"user","content":"hi"}],"max_tokens":50}' | jq .

# ========== 性能参数（按需调）==========
vllm serve Qwen/Qwen3-8B-Instruct \
  --gpu-memory-utilization 0.92 \
  --max-num-seqs 256           # 最大并发请求数
  --max-num-batched-tokens 4096 \
  --enable-prefix-caching       # 启用 prompt cache（多次相同前缀的 prompt 省算力）
  --max-model-len 8192

# ========== Docker 一行启动 ==========
docker run --gpus all -p 8000:8000 \
  -v ~/.cache/huggingface:/root/.cache/huggingface \
  vllm/vllm-openai:latest \
  --model Qwen/Qwen3-8B-Instruct \
  --served-model-name qwen3-8b