第 6 章：AI Agent 基础

# 注意力机制的直观理解（简化版）
# 想象你在翻译 "The cat sat on the mat, it was warm."
# 处理 "it" 时，注意力权重会集中在 "mat" 上

sentence = ["The", "cat", "sat", "on", "the", "mat", "it", "was", "warm"]

# 处理 "it" 时，各词的注意力权重（示意）
attention_weights_for_it = {
    "The": 0.01,
    "cat": 0.05,
    "sat": 0.02,
    "on": 0.02,
    "the": 0.05,
    "mat": 0.75,   # "it" 最可能指代 "mat"
    "it": 0.00,
    "was": 0.05,
    "warm": 0.05
}

import tiktoken

# 获取 GPT-4 的编码器
encoder = tiktoken.encoding_for_model("gpt-4")

text = "Hello world! 你好世界！"
tokens = encoder.encode(text)

print(f"文本: {text}")
print(f"Token 数量: {len(tokens)}")
print(f"Token ID 列表: {tokens}")

# 解码查看每个 Token
for i, token_id in enumerate(tokens):
    token_text = encoder.decode([token_id])
    print(f"  Token {i}: ID={token_id}, 文本='{token_text}'")

# 输出示例：
# Token 数量: 10
# Token ID 列表: [9906, 1917, 0, 57668, 53901, 12131, 10401, 245, 11319, 63823]

import requests

# OpenAI API 调用示例（原生 HTTP）
API_KEY = "sk-your-api-key"

response = requests.post(
    "https://api.openai.com/v1/chat/completions",
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    },
    json={
        "model": "gpt-4o-mini",
        "messages": [
            {"role": "system", "content": "你是一个 helpful 的助手"},
            {"role": "user", "content": "用 Python 写一个快速排序"}
        ],
        "temperature": 0.7,  # 创造性程度：0=保守，2=奔放
        "max_tokens": 1000   # 最大输出 Token 数
    }
)

data = response.json()
print(data["choices"][0]["message"]["content"])

# DeepSeek API 调用（兼容 OpenAI 格式）
import requests

response = requests.post(
    "https://api.deepseek.com/v1/chat/completions",
    headers={
        "Authorization": "Bearer sk-your-deepseek-key",
        "Content-Type": "application/json"
    },
    json={
        "model": "deepseek-chat",
        "messages": [
            {"role": "user", "content": "解释什么是闭包"}
        ],
        "stream": False  # 非流式输出
    }
)

print(response.json()["choices"][0]["message"]["content"])

pip install openai python-dotenv

import os
from openai import OpenAI, APIError, RateLimitError, APIConnectionError
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 初始化客户端（可指向任意兼容 OpenAI 的 API）
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)

def chat_with_llm(messages, model="gpt-4o-mini", temperature=0.7):
    """
    与 LLM 对话，包含完整的错误处理
    """
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature,
            max_tokens=2000
        )
        return {
            "success": True,
            "content": response.choices[0].message.content,
            "usage": {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens
            }
        }
    except RateLimitError:
        return {"success": False, "error": "请求过于频繁，请稍后再试"}
    except APIConnectionError:
        return {"success": False, "error": "网络连接失败，请检查网络"}
    except APIError as e:
        return {"success": False, "error": f"API 错误: {e.message}"}
    except Exception as e:
        return {"success": False, "error": f"未知错误: {str(e)}"}

# 使用示例
result = chat_with_llm([
    {"role": "user", "content": "用一句话解释 REST API"}
])

if result["success"]:
    print(f"回复: {result['content']}")
    print(f"消耗 Token: {result['usage']['total_tokens']}")
else:
    print(f"错误: {result['error']}")

def stream_chat(messages, model="gpt-4o-mini"):
    """
    流式对话，实时输出每个 Token
    """
    try:
        stream = client.chat.completions.create(
            model=model,
            messages=messages,
            stream=True,  # 启用流式输出
            max_tokens=2000
        )

        full_content = ""
        for chunk in stream:
            # 每个 chunk 包含一个增量 Token
            delta = chunk.choices[0].delta.content
            if delta:
                full_content += delta
                print(delta, end="", flush=True)  # 实时打印

        print()  # 换行
        return full_content

    except Exception as e:
        print(f"\n错误: {e}")
        return None

# 使用示例：你会看到文字像打字一样逐字出现
stream_chat([
    {"role": "user", "content": "讲一个关于程序员的短笑话"}
])

# 好的 System Prompt 应该具体、明确
messages = [
    {
        "role": "system",
        "content": """你是一位资深前端工程师，正在指导后端开发者学习 React。
        你的回答特点：
        1. 用后端开发者熟悉的概念做类比
        2. 每个概念都附带可运行的代码示例
        3. 解释"为什么"而不仅仅是"怎么做"
        4. 语气友好但专业"""
    },
    {"role": "user", "content": "什么是 React 的 useEffect？"}
]

# 差的 System Prompt：过于笼统
# "你是一个 helpful 的助手"  -- 没有提供任何行为指导

# Few-shot：让模型按固定格式提取信息
messages = [
    {"role": "system", "content": "从用户输入中提取姓名、年龄和城市，按 JSON 格式输出。"},
    {"role": "user", "content": "我叫张三，今年 25 岁，住在上海。"},
    {"role": "assistant", "content": '{"name": "张三", "age": 25, "city": "上海"}'},
    {"role": "user", "content": "李四，30 岁，北京人。"},
    {"role": "assistant", "content": '{"name": "李四", "age": 30, "city": "北京"}'},
    # 真正的输入
    {"role": "user", "content": "王五今年 28，来自深圳。"}
]

# 不加思维链：模型可能直接猜答案，容易出错
# 用户：15 个工人 8 天修路 2400 米，30 个工人 12 天能修多少米？
# 模型可能直接输出一个错误数字

# 加思维链：引导模型逐步计算
messages = [
    {"role": "user", "content": """15 个工人 8 天修路 2400 米，30 个工人 12 天能修多少米？

请按以下步骤思考：
1. 先计算每个工人每天修多少米
2. 再计算 30 个工人 12 天能修多少米
3. 给出最终答案"""}
]

# 模型会输出类似：
# 1. 每个工人每天修：2400 / 15 / 8 = 20 米
# 2. 30 个工人 12 天修：20 * 30 * 12 = 7200 米
# 3. 最终答案：7200 米

import json

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "你是一个分类器，将用户评论分类为正面/负面/中性。"},
        {"role": "user", "content": "这个手机电池太差了，一天要充三次电！"}
    ],
    response_format={"type": "json_object"},  # 强制 JSON 输出
    max_tokens=500
)

# 解析 JSON 结果
result = json.loads(response.choices[0].message.content)
print(result)
# 输出: {"sentiment": "负面", "confidence": 0.95, "keywords": ["电池差", "充电频繁"]}

# 定义可供模型调用的工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如'北京'、'上海'"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学计算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如'2 + 3 * 4'"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]

import json

# 模拟工具函数实现
def get_weather(city):
    """模拟天气查询"""
    weather_db = {
        "北京": "晴天，25°C",
        "上海": "多云，28°C",
        "深圳": "小雨，30°C"
    }
    return weather_db.get(city, "未知城市")

def calculate(expression):
    """安全计算数学表达式"""
    try:
        # 使用 eval 前做安全检查（仅允许数字和运算符）
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return "表达式包含非法字符"
        return str(eval(expression))
    except Exception as e:
        return f"计算错误: {e}"

# 工具映射表
tool_map = {
    "get_weather": get_weather,
    "calculate": calculate
}

def chat_with_tools(user_message):
    """
    支持 Function Calling 的对话
    """
    messages = [
        {"role": "system", "content": "你是一个 helpful 助手，可以使用工具帮助用户。"},
        {"role": "user", "content": user_message}
    ]

    # 第一次调用：让模型决定是否需要调用工具
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools,
        tool_choice="auto"  # 让模型自动决定
    )

    message = response.choices[0].message

    # 检查模型是否要求调用工具
    if message.tool_calls:
        # 将模型的工具调用请求加入对话历史
        # 将模型的工具调用请求加入对话历史（使用 model_dump 兼容新版 SDK）
        messages.append(message.model_dump(exclude_none=True))

        # 执行每个工具调用
        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            print(f"模型决定调用: {func_name}({func_args})")

            # 执行函数
            result = tool_map[func_name](**func_args)

            # 将结果返回给模型
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(result)
            })

        # 第二次调用：让模型基于工具结果生成回复
        final_response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages
        )
        return final_response.choices[0].message.content

    else:
        # 模型直接回复，不需要调用工具
        return message.content

# 测试：模型会自动识别需要调用哪个工具
print(chat_with_tools("北京今天天气怎么样？"))
print(chat_with_tools("123 乘以 456 等于多少？"))

class ConversationManager:
    """
    多轮对话管理器
    """
    def __init__(self, system_prompt="你是一个 helpful 助手", max_history=10):
        self.messages = [
            {"role": "system", "content": system_prompt}
        ]
        self.max_history = max_history  # 保留最近 N 轮对话

    def add_user_message(self, content):
        """添加用户消息"""
        self.messages.append({"role": "user", "content": content})

    def add_assistant_message(self, content):
        """添加助手回复"""
        self.messages.append({"role": "assistant", "content": content})

    def get_messages(self):
        """获取当前对话历史（包含上下文截断）"""
        # 保留 system 消息 + 最近 max_history 轮对话
        if len(self.messages) <= self.max_history * 2 + 1:
            return self.messages

        # system 消息 + 最近的对话
        return [self.messages[0]] + self.messages[-(self.max_history * 2):]

    def clear(self):
        """清空对话（保留 system）"""
        system = self.messages[0]
        self.messages = [system]

    def estimate_tokens(self):
        """
        估算当前对话的 Token 数量（粗略估算）
        """
        total_chars = sum(len(m["content"]) for m in self.messages)
        # 中文字符约 1.5 Token/字，英文约 0.25 Token/字符
        return int(total_chars * 0.6)

# 使用示例
conv = ConversationManager("你是一个 Python 导师", max_history=5)

conv.add_user_message("什么是列表推导式？")
# -> 调用 API，获取回复
conv.add_assistant_message("列表推导式是 Python 中一种简洁的创建列表的方式...")

conv.add_user_message("和 map 有什么区别？")
# -> 调用 API 时，模型能看到前面的对话，理解"它"指代"列表推导式"
print(conv.get_messages())

def summarize_history(messages, client, model="gpt-4o-mini"):
    """
    将早期对话压缩为摘要
    """
    summary_prompt = """请将以下对话历史总结为一段简洁的摘要，保留所有关键信息和决策。
    摘要应该让没有看过原始对话的人也能理解上下文。"""

    history_text = "\n".join([
        f"{'用户' if m['role'] == 'user' else '助手'}: {m['content'][:200]}"
        for m in messages[1:-4]  # 排除 system 和最近 2 轮
    ])

    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": summary_prompt},
            {"role": "user", "content": history_text}
        ],
        max_tokens=300
    )

    summary = response.choices[0].message.content
    return [
        messages[0],  # system
        {"role": "user", "content": f"【历史摘要】{summary}"},
        *messages[-4:]  # 最近 2 轮完整对话
    ]

#!/usr/bin/env python3
"""
cli_ai_assistant.py
命令行 AI 助手 - 支持多轮对话和流式输出
"""

import os
import sys
from openai import OpenAI
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class CLIAssistant:
    def __init__(self):
        self.client = OpenAI(
            api_key=os.getenv("OPENAI_API_KEY"),
            base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
        )
        self.model = os.getenv("LLM_MODEL", "gpt-4o-mini")
        self.messages = [
            {
                "role": "system",
                "content": "你是一个 helpful 的编程助手。回答简洁，代码示例要完整可运行。"
            }
        ]
        self.total_tokens = 0

    def stream_chat(self, user_input):
        """流式对话"""
        self.messages.append({"role": "user", "content": user_input})

        try:
            stream = self.client.chat.completions.create(
                model=self.model,
                messages=self.messages,
                stream=True,
                max_tokens=2000
            )

            print("助手: ", end="", flush=True)
            full_response = ""

            for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    print(delta, end="", flush=True)
                    full_response += delta

            print()  # 换行
            self.messages.append({"role": "assistant", "content": full_response})

            # 简单的 Token 估算
            self.total_tokens += len(user_input) + len(full_response)

        except Exception as e:
            print(f"\n错误: {e}")

    def show_stats(self):
        """显示对话统计"""
        print(f"\n--- 统计 ---")
        print(f"对话轮数: {(len(self.messages) - 1) // 2}")
        print(f"估算 Token: {self.total_tokens}")
        print(f"------------\n")

    def clear_history(self):
        """清空对话历史"""
        system = self.messages[0]
        self.messages = [system]
        self.total_tokens = 0
        print("对话历史已清空。\n")

    def run(self):
        """主循环"""
        print("=" * 50)
        print("  命令行 AI 助手")
        print("  输入 /quit 退出，/clear 清空历史，/stats 查看统计")
        print("=" * 50)
        print()

        while True:
            try:
                user_input = input("你: ").strip()
            except (KeyboardInterrupt, EOFError):
                print("\n再见！")
                break

            if not user_input:
                continue

            # 处理命令
            if user_input == "/quit":
                print("再见！")
                break
            elif user_input == "/clear":
                self.clear_history()
                continue
            elif user_input == "/stats":
                self.show_stats()
                continue

            self.stream_chat(user_input)
            print()

if __name__ == "__main__":
    assistant = CLIAssistant()
    assistant.run()

# 1. 创建 .env 文件
cat > .env << 'EOF'
OPENAI_API_KEY=sk-your-api-key
OPENAI_BASE_URL=https://api.openai.com/v1
LLM_MODEL=gpt-4o-mini
EOF

# 2. 安装依赖
pip install openai python-dotenv

# 3. 运行
python cli_ai_assistant.py

import base64
from openai import OpenAI

client = OpenAI()

def encode_image(image_path: str) -> str:
    """将本地图片编码为 base64"""
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

def analyze_image(image_path: str, question: str) -> str:
    """分析图片内容"""
    base64_image = encode_image(image_path)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": question},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/png;base64,{base64_image}",
                        "detail": "auto"  # auto / low / high
                    }
                }
            ]
        }],
        max_tokens=1000
    )

    return response.choices[0].message.content

# 使用示例
# result = analyze_image("screenshot.png", "这个界面有哪些设计问题？")
# result = analyze_image("chart.png", "从这张图表中提取数据")
# result = analyze_image("error.png", "这个错误是什么意思？如何修复？")

"""Token 计数与成本估算 (pip install tiktoken)"""
import tiktoken

# 获取编码器
encoder = tiktoken.encoding_for_model("gpt-4o")

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    """计算文本的 Token 数量"""
    encoder = tiktoken.encoding_for_model(model)
    return len(encoder.encode(text))

def count_messages_tokens(messages: list[dict], model: str = "gpt-4o") -> int:
    """计算完整消息列表的 Token 数量

    这包括消息内容 + 每条消息的固定开销（约 3-4 tokens/message）
    """
    encoder = tiktoken.encoding_for_model(model)
    total = 0
    for msg in messages:
        total += 3  # 每条消息的基本开销
        for key, value in msg.items():
            total += len(encoder.encode(str(value)))
        if "name" in msg:
            total += 1  # name 字段的额外开销
    total += 3  # 每次请求的开销
    return total

def estimate_cost(prompt_tokens: int, completion_tokens: int, model: str = "gpt-4o-mini") -> dict:
    """估算 API 调用成本"""
    pricing = {
        "gpt-4o": (0.0025, 0.01),      # (输入, 输出) 每 1K tokens 的美元价格
        "gpt-4o-mini": (0.00015, 0.0006),
        "gpt-4": (0.03, 0.06),
    }

    input_price, output_price = pricing.get(model, (0, 0))

    cost = (prompt_tokens * input_price + completion_tokens * output_price) / 1000

    return {
        "model": model,
        "prompt_tokens": prompt_tokens,
        "completion_tokens": completion_tokens,
        "estimated_cost_usd": round(cost, 6),
        "estimated_cost_cny": round(cost * 7.2, 4),  # 换算人民币
    }

# 使用示例
messages = [
    {"role": "system", "content": "你是一个 Python 助手"},
    {"role": "user", "content": "解释装饰器的用法"}
]

tokens = count_messages_tokens(messages, "gpt-4o-mini")
print(f"输入 Token: {tokens}")
print(estimate_cost(tokens, 500, "gpt-4o-mini"))

# 使用 JSON Schema 强制结构化输出（OpenAI SDK v1.40+）
# 这比 JSON Mode + 自然语言描述更可靠

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": "分析这段代码的问题：for i in range(len(items)): print(items[i])"
    }],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "code_review",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "issues": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "severity": {
                                    "type": "string",
                                    "enum": ["error", "warning", "info"]
                                },
                                "line": {"type": "integer"},
                                "description": {"type": "string"},
                                "suggestion": {"type": "string"}
                            },
                            "required": ["severity", "description", "suggestion"],
                            "additionalProperties": False
                        }
                    },
                    "summary": {"type": "string"}
                },
                "required": ["issues", "summary"],
                "additionalProperties": False
            }
        }
    }
)

result = json.loads(response.choices[0].message.content)
print(json.dumps(result, indent=2, ensure_ascii=False))