1. 项目概述：当命令行遇上多模态大模型

如果你和我一样，每天大部分时间都泡在终端里，那么“效率”这个词对你来说，可能就意味着如何减少在GUI和CLI之间来回切换的次数。我们习惯了用 curl 查询API，用 jq 处理JSON，用管道（ | ）串联起一系列工具，构建出高效的工作流。但当我们需要处理一张图片、分析一份PDF，或者只是想和AI模型进行一场多轮、有上下文的对话时，往往不得不离开熟悉的终端，打开浏览器，登录某个平台的网页界面。这种割裂感，总让人觉得不够“极客”，不够流畅。

oh-my-gemini-cli 这个项目，就是为了弥合这道鸿沟而生的。它本质上是一个命令行接口工具，让你能够直接在终端里，调用类似Gemini这样的多模态大模型的能力。想象一下，你正在分析服务器日志，突然需要理解日志中某段错误信息的截图；或者你在写脚本，需要从一份技术文档的扫描件里提取关键参数。传统做法是：截图/保存文档 -> 打开浏览器 -> 上传文件 -> 等待结果 -> 复制粘贴回终端。而有了 oh-my-gemini-cli ，整个过程可以简化为一行命令： og process screenshot.png ，结果直接输出到终端，或者通过管道传递给下一个命令。

这个工具的核心价值，在于它将强大的多模态AI能力无缝集成到了Unix哲学的工作流中。它不再是一个孤立的AI聊天应用，而是一个可以与其他命令行工具（如 grep , awk , sed , ffmpeg 等）协同工作的“过滤器”或“处理器”。它适合所有以终端为主要工作环境的开发者、运维工程师、技术写作者，以及任何希望用脚本自动化处理图像、文档、音频内容的用户。你不需要改变现有的工作习惯，只需要在工具链中加入这个新的“瑞士军刀”，就能极大地扩展终端的能力边界。

2. 核心设计思路：打造一个“Unix友好”的AI客户端

2.1 为什么是命令行？

在图形界面应用泛滥的今天，为什么还要选择命令行？答案在于 可脚本化 和 可集成性 。一个CLI工具可以被轻易地嵌入到Shell脚本、Python脚本或其他自动化流程中。它的输入是标准输入或文件，输出是标准输出或文件，错误信息是标准错误。这种简单的接口，使得它能够完美地融入现有的CI/CD流水线、监控告警系统、数据处理管道。例如，你可以写一个监控脚本，定期截图，然后用 oh-my-gemini-cli 分析截图内容，判断服务状态是否异常。

2.2 核心功能定位

从项目名称和其设计来看， oh-my-gemini-cli 的核心功能定位非常清晰：

1. 多模态交互 ：支持文本、图像、PDF、音频等多种格式文件的输入。这不是一个简单的文本问答机器人，而是一个真正的多模态终端。
2. 对话上下文管理 ：能够在单次会话中保持多轮对话的上下文，这对于进行复杂的、分步骤的推理或调试至关重要。CLI需要能记住你之前问过什么，模型回答过什么。
3. 流式输出 ：模仿类似 curl 的体验，让模型生成的内容能够像下载文件一样，实时地、一块一块地显示在终端上，而不是等待全部生成完毕再一次性输出。这提升了交互的即时感，对于生成长文本时尤其有用。
4. 配置驱动 ：所有关键参数，如API密钥、默认模型、代理设置、输出格式等，都应通过配置文件或环境变量进行管理，便于在不同环境（开发、测试、生产）间切换和进行版本控制。

2.3 技术栈选型考量

要构建这样一个工具，技术栈的选择直接决定了开发体验和最终用户体验。

• 开发语言 ：通常选择Go、Rust或Python。Go和Rust能编译成单一可执行文件，分发和部署极其简单，性能也好。Python则拥有丰富的生态库，开发速度快。从项目名称和常见实践推断，选择Python的可能性较大，因为它能快速集成各种处理不同文件格式的库（如 Pillow 处理图片， PyPDF2 处理PDF），并且调用HTTP API也更方便。
• 命令行框架 ：Python中， click 或 typer 是构建美观、功能强大CLI的首选。它们能轻松处理子命令、参数、选项、帮助文本生成和彩色输出，大大减轻了开发负担。
• 网络与并发 ：需要处理流式HTTP响应（Server-Sent Events）。 aiohttp 或 httpx （支持异步）库是处理这类请求的良好选择，特别是当需要实现流畅的、不间断的流式输出时，异步编程模型能避免阻塞主线程。
• 配置管理 ：使用 pydantic 配合 pyyaml 或 toml 来定义和验证配置文件是一个稳健的方案。它能确保配置项的类型安全，并提供清晰的错误提示。

注意 ：在工具设计初期，就必须明确区分“用户配置”（如API密钥、默认模型）和“运行时参数”（如本次查询的提示词、温度值）。前者应持久化在本地配置文件中，后者则通过命令行参数传入。这符合CLI工具的设计惯例。

3. 环境配置与核心依赖解析

要让 oh-my-gemini-cli 跑起来，第一步是搭建一个稳定、可复现的Python环境，并理清其核心依赖。

3.1 Python环境隔离：虚拟环境的必要性

强烈建议使用虚拟环境。这能避免项目依赖污染系统级的Python包，也便于在不同项目间切换。这里以 venv 为例：

# 在项目根目录下
python3 -m venv .venv

# 激活虚拟环境 (Linux/macOS)
source .venv/bin/activate

# 激活虚拟环境 (Windows PowerShell)
.venv\Scripts\Activate.ps1
# 如果遇到执行策略限制，可以先执行：Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

激活后，你的命令行提示符前通常会显示 (.venv) ，表示已进入虚拟环境。

3.2 依赖管理与 requirements.txt

一个标准的Python项目会通过 requirements.txt 或 pyproject.toml 来声明依赖。对于 oh-my-gemini-cli ，其核心依赖可能包括：

# requirements.txt
click>=8.0.0  # 命令行界面框架
httpx>=0.24.0  # 现代HTTP客户端，支持异步和流式响应
pydantic>=2.0.0  # 数据验证与配置管理
python-dotenv>=1.0.0  # 从.env文件加载环境变量
pillow>=10.0.0  # 图像处理（用于图片上传前的验证或简单处理）
pypdf2>=3.0.0  # PDF文件处理（用于提取文本或元数据）
rich>=13.0.0  # 终端富文本渲染，让输出更美观
typer[all]>=0.9.0  # 另一个优秀的CLI框架，基于Click，类型提示更友好

使用以下命令安装：

pip install -r requirements.txt

如果项目使用 pyproject.toml （现代Python项目的趋势），那么依赖会定义在 [project] 或 [tool.poetry] 部分。安装命令可能是 pip install -e . 或 poetry install 。

3.3 核心依赖深度解读

• httpx vs requests ：为什么选择 httpx ？除了原生的异步支持， httpx 对HTTP/2的支持更好，连接池管理更高效。对于需要频繁调用外部API的CLI工具，这些特性有助于提升性能和响应速度。其流式响应API也非常直观。
• pydantic 的作用 ：它不仅仅是数据验证。我们将用它来定义“配置模型”和“请求/响应模型”。例如，可以定义一个 GeminiConfig 类，包含 api_key , base_url , model 等字段，并自动从环境变量或配置文件中加载。这比手动解析字典要安全、清晰得多。
• rich 的妙用 ：这个库能让你的CLI输出脱胎换骨。你可以用它来渲染Markdown格式的AI回复、高亮代码块、创建进度条（用于文件上传）、甚至输出漂亮的表格。这极大地提升了工具的专业感和用户体验。

4. 核心功能实现与代码拆解

接下来，我们深入到具体功能的实现层面。假设我们使用 typer 作为CLI框架，因为它能利用Python的类型提示，让代码更简洁，自动生成高质量的帮助文档。

4.1 配置系统的构建

配置是CLI工具的基石。我们设计一个分层配置系统：默认值 < 环境变量 < 配置文件 < 命令行参数。

首先，定义配置模型：

# config.py
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings
from typing import Optional
import os

class GeminiSettings(BaseSettings):
    """Gemini API相关配置"""
    api_key: str = Field(default="", description="Gemini API密钥")
    model: str = Field(default="gemini-pro", description="默认使用的模型")
    base_url: str = Field(default="https://generativelanguage.googleapis.com/v1beta", description="API基础地址")
    temperature: float = Field(default=0.7, ge=0.0, le=1.0, description="生成温度，控制随机性")
    max_output_tokens: Optional[int] = Field(default=None, description="最大输出token数")

    # 可以从.env文件加载
    class Config:
        env_file = ".env"
        env_prefix = "GEMINI_"  # 环境变量前缀，如 GEMINI_API_KEY

    @field_validator('api_key')
    def api_key_must_be_set(cls, v):
        if not v:
            raise ValueError('API密钥未设置。请通过环境变量GEMINI_API_KEY或配置文件设置。')
        return v

# 全局配置实例
settings = GeminiSettings()

然后，我们需要一个命令来初始化配置文件，比如 og config init ：

# cli.py
import typer
import json
from pathlib import Path
from config import GeminiSettings

app = typer.Typer(help="🤖 你的终端AI助手")

config_app = typer.Typer(help="管理配置")
app.add_typer(config_app, name="config")

@config_app.command("init")
def init_config(
    api_key: str = typer.Option(..., prompt=True, hide_input=True, help="你的Gemini API密钥"),
    model: str = typer.Option("gemini-pro", help="默认模型"),
    config_dir: Path = typer.Option(Path.home() / ".config" / "oh-my-gemini", help="配置文件目录")
):
    """初始化配置文件"""
    config_dir.mkdir(parents=True, exist_ok=True)
    config_file = config_dir / "config.json"
    
    config_data = {
        "api_key": api_key,
        "model": model,
        "base_url": "https://generativelanguage.googleapis.com/v1beta",
        "temperature": 0.7
    }
    
    config_file.write_text(json.dumps(config_data, indent=2))
    typer.echo(f"✅ 配置已保存至: {config_file}")
    typer.echo("你可以通过 `og config set <key> <value>` 修改配置。")

4.2 多模态请求的组装与发送

这是工具最核心的部分。我们需要根据不同的输入类型（纯文本、本地图片、PDF、网络图片URL等），构造符合Gemini API要求的请求体。

# gemini_client.py
import httpx
import base64
from pathlib import Path
from typing import List, Dict, Any, Optional, Union
import mimetypes
from config import settings

class GeminiClient:
    def __init__(self):
        self.api_key = settings.api_key
        self.model = settings.model
        self.base_url = settings.base_url
        self.client = httpx.AsyncClient(timeout=30.0)
        
    async def _prepare_image_part(self, image_path: Path) -> Dict[str, Any]:
        """将本地图片文件转换为API接受的格式"""
        if not image_path.exists():
            raise FileNotFoundError(f"图片文件不存在: {image_path}")
        
        mime_type, _ = mimetypes.guess_type(image_path)
        if mime_type is None:
            mime_type = "image/jpeg"  # 默认类型
        
        with open(image_path, "rb") as f:
            image_data = base64.b64encode(f.read()).decode("utf-8")
        
        return {
            "inline_data": {
                "mime_type": mime_type,
                "data": image_data
            }
        }
    
    async def _prepare_file_part(self, file_path: Path) -> Dict[str, Any]:
        """处理通用文件（如PDF），Gemini可能支持直接上传文件或需要先上传到云存储"""
        # 注意：Gemini API对文件处理有特定要求，可能需要先调用files.upload接口
        # 这里简化处理，假设API支持base64内联（实际需查阅最新API文档）
        mime_type, _ = mimetypes.guess_type(file_path)
        if mime_type is None:
            mime_type = "application/octet-stream"
        
        with open(file_path, "rb") as f:
            file_data = base64.b64encode(f.read()).decode("utf-8")
        
        return {
            "inline_data": {
                "mime_type": mime_type,
                "data": file_data
            }
        }
    
    async def generate_content(
        self,
        prompt: str,
        images: Optional[List[Path]] = None,
        files: Optional[List[Path]] = None,
        stream: bool = False
    ) -> Union[str, httpx.Response]:
        """生成内容核心方法"""
        contents = []
        
        # 构建parts列表
        parts = [{"text": prompt}]
        
        # 添加图片部分
        if images:
            for img_path in images:
                image_part = await self._prepare_image_part(img_path)
                parts.append(image_part)
        
        # 添加文件部分
        if files:
            for file_path in files:
                file_part = await self._prepare_file_part(file_path)
                parts.append(file_part)
        
        # 构建请求体
        request_body = {
            "contents": [{"parts": parts}],
            "generationConfig": {
                "temperature": settings.temperature,
                "maxOutputTokens": settings.max_output_tokens,
            }
        }
        
        # 移除None值
        request_body["generationConfig"] = {k: v for k, v in request_body["generationConfig"].items() if v is not None}
        
        url = f"{self.base_url}/models/{self.model}:generateContent"
        if stream:
            url += "?alt=sse"  # 流式端点
        
        params = {"key": self.api_key}
        
        async with self.client as client:
            response = await client.post(
                url,
                params=params,
                json=request_body,
                headers={"Content-Type": "application/json"}
            )
            response.raise_for_status()
            
            if stream:
                return response  # 返回流式响应对象，供后续迭代
            else:
                result = response.json()
                # 解析响应，提取文本
                try:
                    return result["candidates"][0]["content"]["parts"][0]["text"]
                except (KeyError, IndexError) as e:
                    raise ValueError(f"无法解析API响应: {result}") from e
    
    async def close(self):
        await self.client.aclose()

4.3 流式输出的终端渲染

流式输出能带来更好的交互体验。我们需要从Server-Sent Events (SSE) 响应中实时读取数据块并打印。

# stream_handler.py
import asyncio
import json
import re
from typing import AsyncGenerator

async def handle_stream_response(response: httpx.Response) -> AsyncGenerator[str, None]:
    """处理SSE流式响应，逐块生成文本"""
    buffer = ""
    async for chunk in response.aiter_bytes():
        buffer += chunk.decode('utf-8')
        # SSE数据以\n\n分隔每个事件
        while "\n\n" in buffer:
            event, buffer = buffer.split("\n\n", 1)
            for line in event.splitlines():
                if line.startswith("data: "):
                    data = line[6:]  # 去掉"data: "前缀
                    if data == "[DONE]":
                        return
                    try:
                        data_json = json.loads(data)
                        # 提取文本内容，这里路径需要根据实际API响应结构调整
                        text_piece = data_json.get("candidates", [{}])[0].get("content", {}).get("parts", [{}])[0].get("text", "")
                        if text_piece:
                            yield text_piece
                    except json.JSONDecodeError:
                        # 忽略非JSON数据
                        pass

# 在CLI命令中使用
async def stream_generation(prompt: str, image_paths: List[Path]):
    client = GeminiClient()
    try:
        response = await client.generate_content(prompt, images=image_paths, stream=True)
        async for text_chunk in handle_stream_response(response):
            typer.echo(text_chunk, nl=False)  # nl=False避免自动换行
            await asyncio.sleep(0)  # 让出控制权，使输出更流畅
        typer.echo()  # 最后换行
    finally:
        await client.close()

4.4 实现对话上下文管理

要实现多轮对话，我们需要在本地维护一个会话历史。一个简单的方法是使用一个全局的列表来存储 contents （即每轮对话的请求结构）。

# conversation.py
from typing import List, Dict, Any
from pathlib import Path
import json

class Conversation:
    def __init__(self, session_id: str = "default"):
        self.session_id = session_id
        self.history: List[Dict[str, Any]] = []  # 存储完整的contents记录
        self.history_file = Path.home() / ".cache" / "oh-my-gemini" / f"session_{session_id}.json"
        self.history_file.parent.mkdir(parents=True, exist_ok=True)
        self._load()
    
    def _load(self):
        """从文件加载历史会话"""
        if self.history_file.exists():
            try:
                with open(self.history_file, 'r') as f:
                    self.history = json.load(f)
            except json.JSONDecodeError:
                self.history = []
    
    def _save(self):
        """保存历史会话到文件"""
        with open(self.history_file, 'w') as f:
            json.dump(self.history, f, indent=2)
    
    def add_turn(self, role: str, parts: List[Dict]):
        """添加一轮对话（用户或模型）"""
        self.history.append({
            "role": role,  # "user" 或 "model"
            "parts": parts
        })
        # 为了控制上下文长度，可以设置一个最大轮数限制
        max_turns = 20  # 例如，只保留最近20轮对话
        if len(self.history) > max_turns * 2:  # 每轮包含user和model
            self.history = self.history[-(max_turns * 2):]
        self._save()
    
    def get_context_for_api(self) -> List[Dict[str, Any]]:
        """将历史转换为API需要的contents格式"""
        # API需要的格式是 [{"parts": [...]}, {"parts": [...]}, ...]
        # 我们的历史是 [{"role": "...", "parts": [...]}, ...]
        # 需要转换为连续的parts列表，但Gemini API可能通过role区分
        # 根据实际API文档调整
        # 这里假设API接受带role的格式
        return [{"role": turn["role"], "parts": turn["parts"]} for turn in self.history]
    
    def clear(self):
        """清空当前会话历史"""
        self.history.clear()
        if self.history_file.exists():
            self.history_file.unlink()
    
    def get_last_n_turns(self, n: int) -> List[Dict[str, Any]]:
        """获取最近n轮对话（每轮包含用户输入和模型回复）"""
        return self.history[-(n*2):] if n > 0 else []

在CLI命令中，我们需要维护一个全局的 Conversation 实例，并在每次生成内容后，将用户的请求和模型的回复分别添加到历史中。

5. CLI命令设计与用户体验优化

有了核心的客户端和会话管理，我们就可以设计直观易用的命令行接口了。

5.1 主命令结构

# cli.py (续)
import asyncio
from typing import List, Optional
from pathlib import Path
import typer
from rich.console import Console
from rich.markdown import Markdown
from rich.progress import Progress, SpinnerColumn, TextColumn

from gemini_client import GeminiClient
from conversation import Conversation
from stream_handler import handle_stream_response

console = Console()
app = typer.Typer(help="🤖 你的终端AI助手 - 多模态对话与文件处理")
conversation = Conversation()  # 全局会话实例

# 子命令：chat - 进行交互式对话
@app.command()
def chat(
    prompt: Optional[str] = typer.Argument(None, help="直接输入提示词，不提供则进入交互模式"),
    image: Optional[List[Path]] = typer.Option(None, "--image", "-i", help="图片文件路径，可多个", exists=True),
    file: Optional[List[Path]] = typer.Option(None, "--file", "-f", help="文件路径（如PDF），可多个", exists=True),
    stream: bool = typer.Option(True, "--stream/--no-stream", help="是否启用流式输出"),
    clear: bool = typer.Option(False, "--clear", "-c", help="清空当前会话历史"),
):
    """
    与Gemini进行对话。支持文本、图片和文件。
    若不提供prompt参数，则进入交互式对话模式。
    """
    if clear:
        conversation.clear()
        typer.echo("🗑️  会话历史已清空。")
        return
    
    if prompt is None:
        # 交互模式
        typer.echo("💬 进入交互模式。输入消息，或使用命令：/clear 清空历史，/exit 退出，/save <file> 保存历史。")
        while True:
            try:
                user_input = typer.prompt("\n[你]")
                if user_input.strip() == "/exit":
                    break
                elif user_input.strip() == "/clear":
                    conversation.clear()
                    typer.echo("历史已清空。")
                    continue
                elif user_input.strip().startswith("/save"):
                    # 处理保存命令
                    parts = user_input.split()
                    if len(parts) > 1:
                        save_path = Path(parts[1])
                        # 保存逻辑...
                        typer.echo(f"历史已保存至 {save_path}")
                    continue
                
                # 处理用户输入
                await _process_user_input(user_input, image, file, stream, is_interactive=True)
                # 交互模式下，每次循环后重置文件参数
                image = None
                file = None
            except KeyboardInterrupt:
                typer.echo("\n👋 退出交互模式。")
                break
            except EOFError:
                break
    else:
        # 单次命令模式
        await _process_user_input(prompt, image, file, stream, is_interactive=False)

async def _process_user_input(
    prompt: str,
    image: Optional[List[Path]],
    file: Optional[List[Path]],
    stream: bool,
    is_interactive: bool
):
    """处理用户输入的核心逻辑"""
    # 1. 将用户输入添加到会话历史
    user_parts = [{"text": prompt}]
    # ... 这里可以添加处理image/file转换为parts的逻辑，并追加到user_parts
    conversation.add_turn("user", user_parts)
    
    # 2. 调用API
    client = GeminiClient()
    try:
        # 构建完整的上下文（历史 + 当前输入）
        full_context = conversation.get_context_for_api()
        # 注意：实际API调用需要根据上下文调整，这里简化演示
        
        if stream:
            with Progress(
                SpinnerColumn(),
                TextColumn("[progress.description]{task.description}"),
                transient=True,
            ) as progress:
                progress.add_task(description="思考中...", total=None)
                response = await client.generate_content(prompt, images=image, files=file, stream=True)
                full_response_text = ""
                async for chunk in handle_stream_response(response):
                    console.print(chunk, end="", highlight=False)
                    full_response_text += chunk
                console.print()  # 换行
        else:
            with Progress(
                SpinnerColumn(),
                TextColumn("[progress.description]{task.description}"),
                transient=True,
            ) as progress:
                progress.add_task(description="思考中...", total=None)
                response_text = await client.generate_content(prompt, images=image, files=file, stream=False)
                # 使用Rich渲染Markdown格式的输出
                md = Markdown(response_text)
                console.print(md)
                full_response_text = response_text
    except Exception as e:
        typer.echo(f"❌ 请求出错: {e}", err=True)
        # 从历史中移除失败的这一轮用户输入
        if conversation.history and conversation.history[-1]["role"] == "user":
            conversation.history.pop()
        return
    finally:
        await client.close()
    
    # 3. 将模型回复添加到会话历史
    conversation.add_turn("model", [{"text": full_response_text}])
    
    if not is_interactive:
        # 单次模式，可以选择是否打印历史信息
        typer.echo(f"\n📝 本轮对话已添加到会话历史（共{len(conversation.history)//2}轮）。")

# 子命令：config - 管理配置
@app.command()
def config(
    show: bool = typer.Option(False, "--show", "-s", help="显示当前配置"),
    set_key: Optional[str] = typer.Option(None, "--set", help="设置配置项，格式：key=value"),
):
    """管理工具配置"""
    from config import settings
    if show:
        console.print("[bold]当前配置:[/bold]")
        for key, value in settings.model_dump().items():
            console.print(f"  {key}: {value}")
    elif set_key:
        if "=" not in set_key:
            typer.echo("❌ 格式错误，请使用 key=value 格式。")
            raise typer.Exit(1)
        key, value = set_key.split("=", 1)
        # 这里需要实现将配置保存到文件或环境变量的逻辑
        typer.echo(f"✅ 已设置 {key}={value} (需重启工具或重新加载配置生效)")
    else:
        typer.echo("请使用 --show 查看配置，或 --set key=value 修改配置。")

if __name__ == "__main__":
    app()

5.2 实用命令示例

安装后，用户可以通过以下命令体验：

# 设置API密钥（首次使用）
og config init
# 或直接设置环境变量
export GEMINI_API_KEY="your_api_key_here"

# 简单文本问答
og chat "用Python写一个快速排序函数"

# 分析图片
og chat "描述这张图片的内容" -i photo.jpg

# 分析图片并回答特定问题
og chat "图片里的设备是什么型号？预计市场价格是多少？" -i device.jpg

# 处理PDF文件
og chat "总结这份PDF的核心观点" -f document.pdf

# 多文件输入
og chat "对比这两张设计图的异同" -i design1.png -i design2.png

# 进入交互式多轮对话模式
og chat
# 然后直接输入问题，模型会记住上下文

# 在交互模式中使用命令
/clear  # 清空对话历史
/exit   # 退出交互模式

# 流式输出（默认开启）
og chat "写一篇关于量子计算的科普文章" --stream
# 使用 --no-stream 一次性输出

# 清空当前会话
og chat --clear

6. 高级功能与扩展思路

一个基础的CLI工具已经成型，但要使其更强大、更实用，还需要考虑以下高级功能和扩展方向。

6.1 文件上传优化与预处理

对于大文件（如高清图片、长PDF），直接Base64编码放入请求体会导致请求体积巨大，可能超出API限制。更优的方案是：

1. 使用Gemini的文件上传接口 ：许多AI服务提供专用的文件上传端点，返回一个文件ID，然后在生成请求中引用该ID。
2. 本地预处理 ：对于图片，可以先进行压缩或缩放；对于PDF，可以提取关键页面或转换为图片后再处理。这需要集成更多库，如 pdf2image （将PDF转为图片）、 opencv-python （图像处理）。

# 示例：图片预处理函数
from PIL import Image
import io

def preprocess_image(image_path: Path, max_size: tuple = (1024, 1024)) -> bytes:
    """压缩图片到指定最大尺寸，并转换为字节流"""
    with Image.open(image_path) as img:
        img.thumbnail(max_size, Image.Resampling.LANCZOS)
        # 转换为RGB模式（避免RGBA问题）
        if img.mode in ('RGBA', 'LA'):
            background = Image.new('RGB', img.size, (255, 255, 255))
            background.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else img.getchannel('A'))
            img = background
        elif img.mode != 'RGB':
            img = img.convert('RGB')
        
        # 保存到字节流
        img_byte_arr = io.BytesIO()
        img.save(img_byte_arr, format='JPEG', quality=85, optimize=True)
        return img_byte_arr.getvalue()

6.2 支持多种输出格式

默认的终端输出很好，但有时我们需要将结果用于其他程序。可以增加输出格式选项：

# JSON格式输出，便于用jq等工具处理
og chat "分析这张图片" -i chart.png --output-format json

# 纯文本输出，去除所有Markdown格式
og chat "写一份报告摘要" --output-format plain

# 直接保存到文件
og chat "生成周报模板" --output-file weekly_report.md

实现时，可以在 _process_user_input 函数中，根据格式参数对 full_response_text 进行后处理（如解析JSON响应、剥离Markdown标记），然后输出或保存。

6.3 上下文长度管理与智能摘要

长时间对话后，上下文会越来越长，可能导致API调用成本增加或超出token限制。需要实现上下文窗口管理和智能摘要。

• 固定长度滑动窗口 ：只保留最近N轮对话。
• 基于Token数的截断 ：估算历史对话的token数，超过阈值时从最旧的消息开始删除。
• 智能摘要 ：当历史过长时，可以调用模型本身，对之前的对话历史进行总结，然后用总结摘要替代旧历史，从而保留核心信息的同时大幅缩短上下文。这是一个递归过程，需要谨慎设计提示词。

def summarize_conversation(history_text: str, client: GeminiClient) -> str:
    """调用模型对长历史进行摘要"""
    summary_prompt = f"""
请将以下对话历史浓缩成一个简洁的摘要，保留所有关键决策、事实和结论。
摘要将用于后续对话的上下文，因此请确保包含所有对未来对话必要的信息。

对话历史：
{history_text}

摘要：
"""
    # 调用client.generate_content，注意避免循环调用
    # 这是一个简化示例，实际需处理异步和错误
    return "摘要内容..."

6.4 插件系统与工具调用

未来的方向是支持“工具调用”（Function Calling）。让CLI不仅可以问答，还能执行动作。例如，用户说“查看当前目录下最大的5个文件”，CLI可以调用一个本地函数来执行 du -sh * | sort -rh | head -5 并返回结果。

这需要：

1. 定义一套工具描述（名称、描述、参数模式）。
2. 在调用模型时，将这些工具描述作为参数传入。
3. 解析模型的响应，如果它要求调用某个工具，则执行对应的本地函数，并将结果再次发送给模型，形成多步推理。

这能将CLI从一个问答机升级为一个真正的AI辅助终端代理。

7. 常见问题排查与实战技巧

在实际使用和开发过程中，你肯定会遇到各种问题。这里记录了一些典型场景和解决方案。

7.1 网络与API相关问题

问题1：请求超时或连接错误

• 可能原因 ：网络不稳定、代理设置不正确、API服务暂时不可用。
• 排查步骤 ：
  1. 先用 curl 或 ping 测试到API域名的连通性： curl -v https://generativelanguage.googleapis.com 。
  2. 检查代理设置。如果你的网络需要代理，需要在代码中为 httpx.AsyncClient 配置 proxies 参数，或者设置环境变量 HTTP_PROXY / HTTPS_PROXY 。
  3. 增加超时时间：在初始化 httpx.AsyncClient(timeout=30.0) 时，可以适当增加超时限制，特别是处理大文件时。
  4. 查看API状态页面（如果服务商提供），确认是否为服务端问题。

问题2：API返回认证错误（如403 Invalid API Key）

• 可能原因 ：API密钥错误、过期、或未启用相应服务。
• 排查步骤 ：
  1. 确认API密钥是否正确复制，前后有无多余空格。可以通过 og config show 检查配置的密钥。
  2. 登录API提供商的控制台，确认该密钥是否被启用，以及是否对目标模型有访问权限。
  3. 检查密钥是否有使用量限制或已过期。
  4. 尝试在控制台用同一个密钥进行简单测试，以排除本地环境问题。

问题3：流式输出中断或不完整

• 可能原因 ：网络连接在流式传输过程中断开、SSE解析逻辑有缺陷、缓冲区处理不当。
• 排查步骤 ：
  1. 启用调试日志，查看原始SSE数据流。可以在 handle_stream_response 函数中加入打印语句，查看接收到的原始块。
  2. 检查 aiter_bytes() 的缓冲区大小。有时网络包可能被拆分，需要更健壮的SSE消息边界检测逻辑。
  3. 考虑加入重试机制。对于非致命网络错误，可以尝试从断点恢复（但这需要API支持断点续传，通常不支持）。

7.2 文件处理相关问题

问题4：上传图片或PDF时提示“Invalid content”或“Unsupported MIME type”

• 可能原因 ：文件格式不受支持、文件损坏、或Base64编码出错。
• 排查步骤 ：
  1. 确认API文档支持的文件格式列表。例如，某些模型可能只支持 image/jpeg , image/png , image/webp 。
  2. 使用 file 命令（Linux/macOS）或在线工具检查文件的真实MIME类型。 mimetypes.guess_type 并不总是100%准确。
  3. 尝试用PIL/Pillow打开图片，用PyPDF2打开PDF，确认文件未被损坏。
  4. 对于PDF，API可能要求先将文件上传到云存储。仔细阅读API文档中关于文件处理的部分。

问题5：处理大文件时内存占用过高或程序崩溃

• 可能原因 ：将整个大文件读入内存进行Base64编码。
• 解决方案 ：
  1. 实现分块读取和编码。虽然Base64编码需要完整数据，但可以分块读取文件，编码后拼接。
  2. 如前所述，优先使用API的文件上传服务，它通常支持流式上传。
  3. 在本地进行预处理，压缩或裁剪文件到合理大小。

7.3 配置与会话问题

问题6：修改配置文件后不生效

• 可能原因 ：配置未重载、环境变量优先级更高、配置文件路径错误。
• 排查步骤 ：
  1. 确认你修改的是正确的配置文件。使用 og config show 查看当前加载的配置路径和值。
  2. 检查是否有同名的环境变量设置（如 GEMINI_API_KEY ），环境变量的优先级通常高于配置文件。
  3. 尝试重启CLI工具，或者实现一个 og config reload 命令来强制重载配置。
  4. 检查配置文件语法（JSON/YAML/TOML）是否正确。

问题7：对话历史丢失或混乱

• 可能原因 ：会话文件被误删、多进程同时写入导致损坏、会话ID冲突。
• 解决方案 ：
  1. 为会话文件加入更完善的错误处理。在 _load 和 _save 方法中捕获 json.JSONDecodeError 和 IOError ，并提供默认值或备份恢复。
  2. 如果支持多会话（通过 --session 参数），确保每个会话ID对应独立的文件。
  3. 考虑使用更可靠的数据存储，如SQLite，但会引入额外依赖。对于CLI工具，简单的JSON文件在大多数情况下足够，只需做好错误处理。

7.4 性能优化技巧

1. 连接复用 ：确保 GeminiClient 中的 httpx.AsyncClient 实例在多次请求间复用，而不是每次调用都创建新的。这可以通过在类初始化时创建，并在整个应用生命周期中使用同一个客户端来实现。
2. 异步并发 ：如果你需要批量处理多个独立的请求（例如，分析一个目录下的所有图片），可以使用 asyncio.gather 来并发执行，而不是顺序执行，这能大幅提升效率。
3. 缓存 ：对于相同的提示词和文件输入，结果在一定时间内可能是不变的。可以考虑实现一个简单的缓存层（例如使用 diskcache 或 joblib.Memory ），将API响应缓存到本地磁盘，下次相同请求时直接返回缓存结果。注意设置合理的过期时间。
4. 进度反馈 ：对于文件上传等耗时操作，使用 rich.progress 给用户清晰的进度提示，提升体验。

开发这样一个工具，最大的收获不是代码本身，而是对“工具思维”的深化。一个好的CLI工具，其价值在于它如何优雅地融入现有的生态系统，如何通过简单的接口解决复杂问题。 oh-my-gemini-cli 的潜力远不止于当前的功能，随着AI模型能力的演进，它可以成为连接终端本地能力与云端智能的超级枢纽。你可以尝试为它添加对本地代码库的检索增强生成（RAG）功能，让它能回答关于你项目的问题；或者集成系统通知，当长时间运行的任务完成时，让AI总结结果并推送给你。关键在于，始终保持它的“Unix哲学”内核：做好一件事，并能与其他工具完美协作。