def execute(self, action: str, **kwargs) -> str:
    """执行记忆操作

    支持的操作：
    - add: 添加记忆（支持4种类型: working/episodic/semantic/perceptual）
    - search: 搜索记忆
    - summary: 获取记忆摘要
    - stats: 获取统计信息
    - update: 更新记忆
    - remove: 删除记忆
    - forget: 遗忘记忆（多种策略）
    - consolidate: 整合记忆（短期→长期）
    - clear_all: 清空所有记忆
    """

    if action == "add":
        return self._add_memory(**kwargs)
    elif action == "search":
        return self._search_memory(**kwargs)
    elif action == "summary":
        return self._get_summary(**kwargs)
    # ... 其他操作

def _add_memory(
    self,
    content: str = "",
    memory_type: str = "working",
    importance: float = 0.5,
    file_path: str = None,
    modality: str = None,
    **metadata
) -> str:
    """添加记忆"""
    try:
        # 确保会话ID存在
        if self.current_session_id is None:
            self.current_session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"

        # 感知记忆文件支持
        if memory_type == "perceptual" and file_path:
            inferred = modality or self._infer_modality(file_path)
            metadata.setdefault("modality", inferred)
            metadata.setdefault("raw_data", file_path)

        # 添加会话信息到元数据
        metadata.update({
            "session_id": self.current_session_id,
            "timestamp": datetime.now().isoformat()
        })

        memory_id = self.memory_manager.add_memory(
            content=content,
            memory_type=memory_type,
            importance=importance,
            metadata=metadata,
            auto_classify=False
        )

        return f"✅ 记忆已添加 (ID: {memory_id[:8]}...)"

    except Exception as e:
        return f"❌ 添加记忆失败: {str(e)}"

# 1. 工作记忆 - 临时信息，容量有限
memory_tool.execute("add",
    content="用户刚才问了关于Python函数的问题",
    memory_type="working",
    importance=0.6
)

# 2. 情景记忆 - 具体事件和经历
memory_tool.execute("add",
    content="2024年3月15日，用户张三完成了第一个Python项目",
    memory_type="episodic",
    importance=0.8,
    event_type="milestone",
    location="在线学习平台"
)

# 3. 语义记忆 - 抽象知识和概念
memory_tool.execute("add",
    content="Python是一种解释型、面向对象的编程语言",
    memory_type="semantic",
    importance=0.9,
    knowledge_type="factual"
)

# 4. 感知记忆 - 多模态信息
memory_tool.execute("add",
    content="用户上传了一张Python代码截图，包含函数定义",
    memory_type="perceptual",
    importance=0.7,
    modality="image",
    file_path="./uploads/code_screenshot.png"
)

def _search_memory(
    self,
    query: str,
    limit: int = 5,
    memory_types: List[str] = None,
    memory_type: str = None,
    min_importance: float = 0.1
) -> str:
    """搜索记忆"""
    try:
        # 参数标准化处理
        if memory_type and not memory_types:
            memory_types = [memory_type]

        results = self.memory_manager.retrieve_memories(
            query=query,
            limit=limit,
            memory_types=memory_types,
            min_importance=min_importance
        )

        if not results:
            return f"🔍 未找到与 '{query}' 相关的记忆"

        # 格式化结果
        formatted_results = []
        formatted_results.append(f"🔍 找到 {len(results)} 条相关记忆:")

        for i, memory in enumerate(results, 1):
            memory_type_label = {
                "working": "工作记忆",
                "episodic": "情景记忆", 
                "semantic": "语义记忆",
                "perceptual": "感知记忆"
            }.get(memory.memory_type, memory.memory_type)
# .get(key, default) 安全取值
# 如果 memory.memory_type 在字典里存在 → 返回对应的中文名；
# 如果 不存在（比如出现了未知类型）→ 返回 memory.memory_type 本身（即原始值）

            content_preview = memory.content[:80] + "..." if len(memory.content) > 80 else memory.content
            formatted_results.append(
                f"{i}. [{memory_type_label}] {content_preview} (重要性: {memory.importance:.2f})"
            )

        return "\n".join(formatted_results)

    except Exception as e:
        return f"❌ 搜索记忆失败: {str(e)}"

# 基础搜索
result = memory_tool.execute("search", query="Python编程", limit=5)

# 指定记忆类型搜索
result = memory_tool.execute("search",
    query="学习进度",
    memory_type="episodic",
    limit=3
)

# 多类型搜索
result = memory_tool.execute("search",
    query="函数定义",
    memory_types=["semantic", "episodic"],
    min_importance=0.5
)

def _forget(self, strategy: str = "importance_based", threshold: float = 0.1, max_age_days: int = 30) -> str:
    """遗忘记忆（支持多种策略）"""
    try:
        count = self.memory_manager.forget_memories(
            strategy=strategy,
            threshold=threshold,
            max_age_days=max_age_days
        )
        return f"🧹 已遗忘 {count} 条记忆（策略: {strategy}）"
    except Exception as e:
        return f"❌ 遗忘记忆失败: {str(e)}"

# 1. 基于重要性的遗忘 - 删除重要性低于阈值的记忆
memory_tool.execute("forget",
    strategy="importance_based",
    threshold=0.2
)

# 2. 基于时间的遗忘 - 删除超过指定天数的记忆
memory_tool.execute("forget",
    strategy="time_based",
    max_age_days=30
)

# 3. 基于容量的遗忘 - 当记忆数量超限时删除最不重要的
memory_tool.execute("forget",
    strategy="capacity_based",
    threshold=0.3
)

def _consolidate(self, from_type: str = "working", to_type: str = "episodic", importance_threshold: float = 0.7) -> str:
    """整合记忆（将重要的短期记忆提升为长期记忆）"""
    try:
        count = self.memory_manager.consolidate_memories(
            from_type=from_type,
            to_type=to_type,
            importance_threshold=importance_threshold,
        )
        return f"🔄 已整合 {count} 条记忆为长期记忆（{from_type} → {to_type}，阈值={importance_threshold}）"
    except Exception as e:
        return f"❌ 整合记忆失败: {str(e)}"

# 将重要的工作记忆转为情景记忆
memory_tool.execute("consolidate",
    from_type="working",
    to_type="episodic",
    importance_threshold=0.7
)

# 将重要的情景记忆转为语义记忆
memory_tool.execute("consolidate",
    from_type="episodic",
    to_type="semantic",
    importance_threshold=0.8
)

class MemoryTool(Tool):
    """记忆工具 - 为Agent提供记忆功能"""
    
    def __init__(
        self,
        user_id: str = "default_user",
        memory_config: MemoryConfig = None,
        memory_types: List[str] = None
    ):
        super().__init__(
            name="memory",
            description="记忆工具 - 可以存储和检索对话历史、知识和经验"
        )
        
        # 初始化记忆管理器
        self.memory_config = memory_config or MemoryConfig()
        self.memory_types = memory_types or ["working", "episodic", "semantic"]
        
        self.memory_manager = MemoryManager(
            config=self.memory_config,
            user_id=user_id,
            enable_working="working" in self.memory_types,
            enable_episodic="episodic" in self.memory_types,
            enable_semantic="semantic" in self.memory_types,
            enable_perceptual="perceptual" in self.memory_types
        )

class MemoryManager:
    """记忆管理器 - 统一的记忆操作接口"""

    def __init__(
        self,
        config: Optional[MemoryConfig] = None,
        user_id: str = "default_user",
        enable_working: bool = True,
        enable_episodic: bool = True,
        enable_semantic: bool = True,
        enable_perceptual: bool = False
    ):
        self.config = config or MemoryConfig()
        self.user_id = user_id

        # 初始化存储和检索组件
        self.store = MemoryStore(self.config)
        self.retriever = MemoryRetriever(self.store, self.config)

        # 初始化各类型记忆
        self.memory_types = {}

        if enable_working:
            self.memory_types['working'] = WorkingMemory(self.config, self.store)

        if enable_episodic:
            self.memory_types['episodic'] = EpisodicMemory(self.config, self.store)

        if enable_semantic:
            self.memory_types['semantic'] = SemanticMemory(self.config, self.store)

        if enable_perceptual:
            self.memory_types['perceptual'] = PerceptualMemory(self.config, self.store)

class WorkingMemory:
    """工作记忆实现
    特点：
    - 容量有限（默认50条）+ TTL自动清理
    - 纯内存存储，访问速度极快
    - 混合检索：TF-IDF向量化 + 关键词匹配
    """
    
    def __init__(self, config: MemoryConfig):
        self.max_capacity = config.working_memory_capacity or 50
        self.max_age_minutes = config.working_memory_ttl or 60
        self.memories = []
    
    def add(self, memory_item: MemoryItem) -> str:
        """添加工作记忆"""
        self._expire_old_memories()  # 过期清理
        
        if len(self.memories) >= self.max_capacity:
            self._remove_lowest_priority_memory()  # 容量管理
        
        self.memories.append(memory_item)
        return memory_item.id
    
    def retrieve(self, query: str, limit: int = 5, **kwargs) -> List[MemoryItem]:
        """混合检索：TF-IDF向量化 + 关键词匹配"""
        self._expire_old_memories()
        
        # 尝试TF-IDF向量检索
        vector_scores = self._try_tfidf_search(query)
        
        # 计算综合分数
        scored_memories = []
        for memory in self.memories:
            vector_score = vector_scores.get(memory.id, 0.0)
            keyword_score = self._calculate_keyword_score(query, memory.content)
            
            # 混合评分
            base_relevance = vector_score * 0.7 + keyword_score * 0.3 if vector_score > 0 else keyword_score
            time_decay = self._calculate_time_decay(memory.timestamp)
            importance_weight = 0.8 + (memory.importance * 0.4)
            
            final_score = base_relevance * time_decay * importance_weight
            if final_score > 0:
                scored_memories.append((final_score, memory))
        
        scored_memories.sort(key=lambda x: x[0], reverse=True)
        return [memory for _, memory in scored_memories[:limit]]

class EpisodicMemory:
    """情景记忆实现
    特点：
    - SQLite+Qdrant混合存储架构
    - 支持时间序列和会话级检索
    - 结构化过滤 + 语义向量检索
    """
    
    def __init__(self, config: MemoryConfig):
        self.doc_store = SQLiteDocumentStore(config.database_path)
        self.vector_store = QdrantVectorStore(config.qdrant_url, config.qdrant_api_key)
        self.embedder = create_embedding_model_with_fallback()
        self.sessions = {}  # 会话索引
    
    def add(self, memory_item: MemoryItem) -> str:
        """添加情景记忆"""
        # 创建情景对象
        episode = Episode(
            episode_id=memory_item.id,
            session_id=memory_item.metadata.get("session_id", "default"),
            timestamp=memory_item.timestamp,
            content=memory_item.content,
            context=memory_item.metadata
        )
        
        # 更新会话索引
        session_id = episode.session_id
        if session_id not in self.sessions:
            self.sessions[session_id] = []
        self.sessions[session_id].append(episode.episode_id)
        
        # 持久化存储（SQLite + Qdrant）
        self._persist_episode(episode)
        return memory_item.id
    
    def retrieve(self, query: str, limit: int = 5, **kwargs) -> List[MemoryItem]:
        """核心所在：混合检索：结构化过滤 + 语义向量检索"""
        # 1. 用SQLite结构化预过滤（时间范围、重要性等）
        candidate_ids = self._structured_filter(**kwargs)
        
        # 2. 向量语义检索（超采样为后续重排序留足空间）
        hits = self._vector_search(query, limit * 5, kwargs.get("user_id"))
        
        # 3. 综合评分与排序
        results = []
        for hit in hits:
            if self._should_include(hit, candidate_ids, kwargs):
                score = self._calculate_episode_score(hit)
                memory_item = self._create_memory_item(hit)
                results.append((score, memory_item))
        
        results.sort(key=lambda x: x[0], reverse=True)
        return [item for _, item in results[:limit]]
    
    def _calculate_episode_score(self, hit) -> float:
        """情景记忆评分算法"""
        # 语义相似度（与查询意图最相关的内容优先）
        vec_score = float(hit.get("score", 0.0)) 
        # 时间近因性（最近的记忆更有参考价值）
        recency_score = self._calculate_recency(hit["metadata"]["timestamp"])
        # 重要性
        importance = hit["metadata"].get("importance", 0.5)
        
        # 评分公式：(向量相似度 × 0.8 + 时间近因性 × 0.2) × 重要性权重（0.8 ~ 1.2 倍，标记为"重要"的记忆获得额外加分）
        base_relevance = vec_score * 0.8 + recency_score * 0.2
        importance_weight = 0.8 + (importance * 0.4)
        
        return base_relevance * importance_weight

class SemanticMemory(BaseMemory):
    """语义记忆实现
    
    特点：
    - 使用HuggingFace中文预训练模型进行文本嵌入
    - 向量检索进行快速相似度匹配
    - 知识图谱存储实体和关系
    - 混合检索策略：向量+图+语义推理
    """
    
    def __init__(self, config: MemoryConfig, storage_backend=None):
        super().__init__(config, storage_backend)
        
        # 嵌入模型（统一提供）
        self.embedding_model = get_text_embedder()
        
        # 专业数据库存储
        self.vector_store = QdrantConnectionManager.get_instance(**qdrant_config)
        self.graph_store = Neo4jGraphStore(**neo4j_config)
        
        # 实体和关系缓存
        self.entities: Dict[str, Entity] = {}
        self.relations: List[Relation] = []
        
        # NLP处理器（支持中英文）
        self.nlp = self._init_nlp()

def add(self, memory_item: MemoryItem) -> str:
    """添加语义记忆"""
    # 1. 生成文本嵌入
    embedding = self.embedding_model.encode(memory_item.content)
    
    # 2. 提取实体和关系
    entities = self._extract_entities(memory_item.content)
    relations = self._extract_relations(memory_item.content, entities)
    
    # 3. 存储到Neo4j图数据库
    for entity in entities:
        self._add_entity_to_graph(entity, memory_item)
    
    for relation in relations:
        self._add_relation_to_graph(relation, memory_item)
    
    # 4. 存储到Qdrant向量数据库
    metadata = {
        "memory_id": memory_item.id,
        "entities": [e.entity_id for e in entities],
        "entity_count": len(entities),
        "relation_count": len(relations)
    }
    
    self.vector_store.add_vectors(
        vectors=[embedding.tolist()],
        metadata=[metadata],
        ids=[memory_item.id]
    )

def retrieve(self, query: str, limit: int = 5, **kwargs) -> List[MemoryItem]:
    """检索语义记忆"""
    # 1. 向量检索
    vector_results = self._vector_search(query, limit * 2, user_id)
    
    # 2. 图检索
    graph_results = self._graph_search(query, limit * 2, user_id)
    
    # 3. 混合排序
    combined_results = self._combine_and_rank_results(
        vector_results, graph_results, query, limit
    )
    
    return combined_results[:limit]

def _combine_and_rank_results(self, vector_results, graph_results, query, limit):
    """混合排序结果"""
    combined = {}
    
    # 合并向量和图检索结果
    for result in vector_results:
        combined[result["memory_id"]] = {
            **result,
            "vector_score": result.get("score", 0.0),
            "graph_score": 0.0
        }
    
    for result in graph_results:
        memory_id = result["memory_id"]
        if memory_id in combined:
            combined[memory_id]["graph_score"] = result.get("similarity", 0.0)
        else:
            combined[memory_id] = {
                **result,
                "vector_score": 0.0,
                "graph_score": result.get("similarity", 0.0)
            }
    
    # 计算混合分数
    for memory_id, result in combined.items():
        vector_score = result["vector_score"]
        graph_score = result["graph_score"]
        importance = result.get("importance", 0.5)
        
        # 基础相似度得分
        base_relevance = vector_score * 0.7 + graph_score * 0.3
        
        # 重要性权重 [0.8, 1.2]
        importance_weight = 0.8 + (importance * 0.4)
        
        # 最终得分：相似度 * 重要性权重
        combined_score = base_relevance * importance_weight
        result["combined_score"] = combined_score
    
    # 排序并返回
    sorted_results = sorted(
        combined.values(),
        key=lambda x: x["combined_score"],
        reverse=True
    )
    
    return sorted_results[:limit]

class PerceptualMemory(BaseMemory):
    """感知记忆实现
    
    特点：
    - 支持多模态数据（文本、图像、音频等）
    - 跨模态相似性搜索
    - 感知数据的语义理解
    - 支持内容生成和检索
    """
    
    def __init__(self, config: MemoryConfig, storage_backend=None):
        super().__init__(config, storage_backend)
        
        # 多模态编码器
        self.text_embedder = get_text_embedder()
        self._clip_model = self._init_clip_model()  # 图像编码
        self._clap_model = self._init_clap_model()  # 音频编码
        
        # 按模态分离的向量存储
        self.vector_stores = {
            "text": QdrantConnectionManager.get_instance(
                collection_name="perceptual_text",
                vector_size=self.vector_dim
            ),
            "image": QdrantConnectionManager.get_instance(
                collection_name="perceptual_image", 
                vector_size=self._image_dim
            ),
            "audio": QdrantConnectionManager.get_instance(
                collection_name="perceptual_audio",
                vector_size=self._audio_dim
            )
        }

def retrieve(self, query: str, limit: int = 5, **kwargs) -> List[MemoryItem]:
    """检索感知记忆（可筛模态；同模态向量检索+时间/重要性融合）"""
    user_id = kwargs.get("user_id")
    target_modality = kwargs.get("target_modality")
    query_modality = kwargs.get("query_modality", target_modality or "text")
    
    # 同模态向量检索
    try:
        query_vector = self._encode_data(query, query_modality)
        store = self._get_vector_store_for_modality(target_modality or query_modality)
        
        where = {"memory_type": "perceptual"}
        if user_id:
            where["user_id"] = user_id
        if target_modality:
            where["modality"] = target_modality
        
        hits = store.search_similar(
            query_vector=query_vector,
            limit=max(limit * 5, 20),
            where=where
        )
    except Exception:
        hits = []
    
    # 融合排序（向量相似度 + 时间近因性 + 重要性权重）
    results = []
    for hit in hits:
        vector_score = float(hit.get("score", 0.0))
        recency_score = self._calculate_recency_score(hit["metadata"]["timestamp"])
        importance = hit["metadata"].get("importance", 0.5)
        
        # 评分算法
        base_relevance = vector_score * 0.8 + recency_score * 0.2
        importance_weight = 0.8 + (importance * 0.4)
        combined_score = base_relevance * importance_weight
        
        results.append((combined_score, self._create_memory_item(hit)))
    
    results.sort(key=lambda x: x[0], reverse=True)
    return [item for _, item in results[:limit]]

def _calculate_recency_score(self, timestamp: str) -> float:
    """计算时间近因性得分"""
    try:
        memory_time = datetime.fromisoformat(timestamp)
        current_time = datetime.now()
        age_hours = (current_time - memory_time).total_seconds() / 3600
        
        # 指数衰减：24小时内保持高分，之后逐渐衰减
        decay_factor = 0.1  # 衰减系数
        recency_score = math.exp(-decay_factor * age_hours / 24)
        
        return max(0.1, recency_score)  # 最低保持0.1的基础分数
    except Exception:
        return 0.5  # 默认中等分数

用户层：RAGTool统一接口
  ↓
应用层：智能问答、搜索、管理
  ↓  
处理层：文档解析、分块、向量化
  ↓
存储层：向量数据库、文档存储
  ↓
基础层：嵌入模型、LLM、数据库

class RAGTool(Tool):
    """RAG工具
    
    提供完整的 RAG 能力：
    - 添加多格式文档（PDF、Office、图片、音频等）
    - 智能检索与召回
    - LLM 增强问答
    - 知识库管理
    """
    
    def __init__(
        self,
        knowledge_base_path: str = "./knowledge_base",
        qdrant_url: str = None,
        qdrant_api_key: str = None,
        collection_name: str = "rag_knowledge_base",
        rag_namespace: str = "default"
    ):
        # 初始化RAG管道
        self._pipelines: Dict[str, Dict[str, Any]] = {}
        self.llm = HelloAgentsLLM()
        
        # 创建默认管道
        default_pipeline = create_rag_pipeline(
            qdrant_url=self.qdrant_url,
            qdrant_api_key=self.qdrant_api_key,
            collection_name=self.collection_name,
            rag_namespace=self.rag_namespace
        )
        self._pipelines[self.rag_namespace] = default_pipeline

任意格式文档 → MarkItDown转换 → Markdown文本 → 智能分块 → 向量化 → 存储检索

def _convert_to_markdown(path: str) -> str:
    """
    Universal document reader using MarkItDown with enhanced PDF processing.
    核心功能：将任意格式文档转换为Markdown文本
    
    支持格式：
    - 文档：PDF、Word、Excel、PowerPoint
    - 图像：JPG、PNG、GIF（通过OCR）
    - 音频：MP3、WAV、M4A（通过转录）
    - 文本：TXT、CSV、JSON、XML、HTML
    - 代码：Python、JavaScript、Java等
    """
    if not os.path.exists(path):
        return ""
    
    # 对PDF文件使用增强处理
    ext = (os.path.splitext(path)[1] or '').lower()
    if ext == '.pdf':
        return _enhanced_pdf_processing(path)
    
    # 其他格式使用MarkItDown统一转换
    md_instance = _get_markitdown_instance()
    if md_instance is None:
        return _fallback_text_reader(path)
    
    try:
        result = md_instance.convert(path)
        markdown_text = getattr(result, "text_content", None)
        if isinstance(markdown_text, str) and markdown_text.strip():
            print(f"[RAG] MarkItDown转换成功: {path} -> {len(markdown_text)} chars Markdown")
            return markdown_text
        return ""
    except Exception as e:
        print(f"[WARNING] MarkItDown转换失败 {path}: {e}")
        return _fallback_text_reader(path)

标准Markdown文本 → 标题层次解析 → 段落语义分割 → Token计算分块 → 重叠策略优化 → 向量化准备
       ↓                ↓              ↓            ↓           ↓            ↓
   统一格式          #/##/###        语义边界      大小控制     信息连续性    嵌入向量
   结构清晰          层次识别        完整性保证    检索优化     上下文保持    相似度匹配

def _split_paragraphs_with_headings(text: str) -> List[Dict]:
    """根据标题层次分割段落，保持语义完整性"""
    lines = text.splitlines()
    heading_stack: List[str] = []
    paragraphs: List[Dict] = []
    buf: List[str] = []
    char_pos = 0
    
    def flush_buf(end_pos: int):
        if not buf:
            return
        content = "\n".join(buf).strip()
        if not content:
            return
        paragraphs.append({
            "content": content,
            "heading_path": " > ".join(heading_stack) if heading_stack else None, # 标题层级路径，如 "章节1 > 子章节1.1"
            "start": max(0, end_pos - len(content)),
            "end": end_pos,
        })
    
    for ln in lines:
        raw = ln
        if raw.strip().startswith("#"):
            # 处理标题行
            flush_buf(char_pos) # 先保存当前段落
            level = len(raw) - len(raw.lstrip('#')) # 计算标题级别
            title = raw.lstrip('#').strip()
            
            # 调整标题栈
            if level <= 0:
                level = 1
            if level <= len(heading_stack):
                heading_stack = heading_stack[:level-1] # 截断到当前层级
            heading_stack.append(title)
            
            char_pos += len(raw) + 1
            continue
        
        # 段落内容累积
        if raw.strip() == "":
            flush_buf(char_pos) # 空行触发段落结束
            buf = []
        else:
            buf.append(raw) # 非空行累积到缓冲区
        char_pos += len(raw) + 1
    
    flush_buf(char_pos)
    
    if not paragraphs:
        paragraphs = [{"content": text, "heading_path": None, "start": 0, "end": len(text)}]
    
    return paragraphs

# 核心策略是以段落为最小单位进行累积，当超出token限制时触发分块并构建重叠部分。
# 重叠机制：通过保留尾部段落实现上下文连贯（保留已累积的段落作为下一个块的起始）

def _chunk_paragraphs(paragraphs: List[Dict], chunk_tokens: int, overlap_tokens: int) -> List[Dict]:
    """基于Token数量的智能分块"""
    chunks: List[Dict] = []
    cur: List[Dict] = []
    cur_tokens = 0
    i = 0
    
    while i < len(paragraphs):
        p = paragraphs[i]
        p_tokens = _approx_token_len(p["content"]) or 1
        
        if cur_tokens + p_tokens <= chunk_tokens or not cur:
            cur.append(p)
            cur_tokens += p_tokens
            i += 1
        else:
            # 生成当前分块
            content = "\n\n".join(x["content"] for x in cur)
            start = cur[0]["start"]
            end = cur[-1]["end"]
            heading_path = next((x["heading_path"] for x in reversed(cur) if x.get("heading_path")), None)
            
            chunks.append({
                "content": content,
                "start": start,
                "end": end,
                "heading_path": heading_path,
            })
            
            # 构建重叠部分
            if overlap_tokens > 0 and cur:
                kept: List[Dict] = []
                kept_tokens = 0
                for x in reversed(cur):
                    t = _approx_token_len(x["content"]) or 1
                    if kept_tokens + t > overlap_tokens:
                        break
                    kept.append(x)
                    kept_tokens += t
                cur = list(reversed(kept))
                cur_tokens = kept_tokens
            else:
                cur = []
                cur_tokens = 0
    
    # 处理最后一个分块
    if cur:
        content = "\n\n".join(x["content"] for x in cur)
        start = cur[0]["start"]
        end = cur[-1]["end"]
        heading_path = next((x["heading_path"] for x in reversed(cur) if x.get("heading_path")), None)
        
        chunks.append({
            "content": content,
            "start": start,
            "end": end,
            "heading_path": heading_path,
        })
    
    return chunks

def _approx_token_len(text: str) -> int:
    """近似估计Token长度，支持中英文混合"""
    # CJK字符按1 token计算
    cjk = sum(1 for ch in text if _is_cjk(ch))
    # 其他字符按空白分词计算
    non_cjk_tokens = len([t for t in text.split() if t])
    return cjk + non_cjk_tokens

def _is_cjk(ch: str) -> bool:
    """判断是否为CJK字符"""
    code = ord(ch)
    return (
        0x4E00 <= code <= 0x9FFF or  # CJK统一汉字
        0x3400 <= code <= 0x4DBF or  # CJK扩展A
        0x20000 <= code <= 0x2A6DF or # CJK扩展B
        0x2A700 <= code <= 0x2B73F or # CJK扩展C
        0x2B740 <= code <= 0x2B81F or # CJK扩展D
        0x2B820 <= code <= 0x2CEAF or # CJK扩展E
        0xF900 <= code <= 0xFAFF      # CJK兼容汉字
    )

def index_chunks(
    store = None, 
    chunks: List[Dict] = None, 
    cache_db: Optional[str] = None, 
    batch_size: int = 64,
    rag_namespace: str = "default"
) -> None:
    """
    Index markdown chunks with unified embedding and Qdrant storage.
    Uses百炼 API with fallback to sentence-transformers.
    """
    if not chunks:
        print("[RAG] No chunks to index")
        return
    
    # 使用统一嵌入模型
    embedder = get_text_embedder()
    dimension = get_dimension(384)
    
    # 创建默认Qdrant存储
    if store is None:
        store = _create_default_vector_store(dimension)
        print(f"[RAG] Created default Qdrant store with dimension {dimension}")
    
    # 预处理Markdown文本以获得更好的嵌入质量
    processed_texts = []
    for c in chunks:
        raw_content = c["content"]
        processed_content = _preprocess_markdown_for_embedding(raw_content)
        processed_texts.append(processed_content)
    
    print(f"[RAG] Embedding start: total_texts={len(processed_texts)} batch_size={batch_size}")
    
    # 批量编码
    vecs: List[List[float]] = []
    for i in range(0, len(processed_texts), batch_size):
        part = processed_texts[i:i+batch_size]
        try:
            # 使用统一嵌入器（内部处理缓存）
            part_vecs = embedder.encode(part)
            
            # 标准化为List[List[float]]格式
            if not isinstance(part_vecs, list):
                if hasattr(part_vecs, "tolist"):
                    part_vecs = [part_vecs.tolist()]
                else:
                    part_vecs = [list(part_vecs)]
            
            # 处理向量格式和维度
            for v in part_vecs:
                try:
                    if hasattr(v, "tolist"):
                        v = v.tolist()
                    v_norm = [float(x) for x in v]
                    
                    # 维度检查和调整
                    if len(v_norm) != dimension:
                        print(f"[WARNING] 向量维度异常: 期望{dimension}, 实际{len(v_norm)}")
                        if len(v_norm) < dimension:
                            v_norm.extend([0.0] * (dimension - len(v_norm)))
                        else:
                            v_norm = v_norm[:dimension]
                    
                    vecs.append(v_norm)
                except Exception as e:
                    print(f"[WARNING] 向量转换失败: {e}, 使用零向量")
                    vecs.append([0.0] * dimension)
                    
        except Exception as e:
            print(f"[WARNING] Batch {i} encoding failed: {e}")
            # 实现重试机制
            # ... 重试逻辑 ...
        
        print(f"[RAG] Embedding progress: {min(i+batch_size, len(processed_texts))}/{len(processed_texts)}")

prompt = [
    {"role": "system", "content": "你是检索查询扩展助手。生成语义等价或互补的多样化查询。使用中文，简短，避免标点。"},
    {"role": "user", "content": f"原始查询：{query}\n请给出{n}个不同表述的查询，每行一个。"}
]

prompt = [
    {"role": "system", "content": "根据用户问题，先写一段可能的答案性段落，用于向量检索的查询文档（不要分析过程）。"},
    {"role": "user", "content": f"问题：{query}\n请直接写一段中等长度、客观、包含关键术语的段落。"}
]

---
id: note_20250119_153000_0
title: 项目进展 - 第一阶段
type: task_state
tags: [refactoring, phase1, backend]
created_at: 2025-01-19T15:30:00
updated_at: 2025-01-19T15:30:00
---

# 项目进展 - 第一阶段

## 完成情况

已完成数据模型层的重构,主要改动包括:

1. 统一了实体类的命名规范
2. 引入了类型提示,提升代码可维护性
3. 优化了数据库查询性能

## 测试覆盖

- 单元测试覆盖率: 85%
- 集成测试覆盖率: 70%

## 下一步计划

1. 重构业务逻辑层
2. 解决依赖冲突问题
3. 提升集成测试覆盖率至85%

{
  "note_20250119_153000_0": {
    "id": "note_20250119_153000_0",
    "title": "项目进展 - 第一阶段",
    "type": "task_state",
    "tags": ["refactoring", "phase1", "backend"],
    "created_at": "2025-01-19T15:30:00",
    "updated_at": "2025-01-19T15:30:00",
    "file_path": "./notes/note_20250119_153000_0.md"
  }
}

​

# 传统方式：手动集成每个服务
class GitHubTool(BaseTool):
    """需要手写GitHub API适配器"""
    def run(self, repo_url):
        # 大量的API调用代码...
        pass

class DatabaseTool(BaseTool):
    """需要手写数据库适配器"""
    def run(self, query):
        # 数据库连接和查询代码...
        pass

class WeatherTool(BaseTool):
    """需要手写天气API适配器"""
    def run(self, location):
        # 天气API调用代码...
        pass

# 每个新服务都需要重复这个过程
agent.add_tool(GitHubTool())
agent.add_tool(DatabaseTool())
agent.add_tool(WeatherTool())

from hello_agents.tools import MCPTool

# 连接到MCP服务器，自动获得所有工具
mcp_tool = MCPTool()  # 内置服务器提供基础工具

# 或者连接到专业的MCP服务器
github_mcp = MCPTool(server_command=["npx", "-y", "@modelcontextprotocol/server-github"])
database_mcp = MCPTool(server_command=["python", "database_mcp_server.py"])

# 智能体自动获得所有能力，无需手写适配器
agent.add_tool(mcp_tool)
agent.add_tool(github_mcp)
agent.add_tool(database_mcp)

"""
功能：
    使用 OpenAI Chat Completions 的 function calling / tool calling 实现一个最小 Agent。

场景：
    用户问：“请查询北京天气，并把摄氏温度换算成华氏温度，最后给出穿衣建议。”
    Agent 会让模型自己决定是否调用工具：
        1. get_weather(city)：查询演示天气数据
        2. celsius_to_fahrenheit(celsius)：摄氏温度转华氏温度
"""

# os 用于读取系统环境变量，例如 LLM_MODEL_ID、LLM_API_KEY。
import os

# json 用于把工具入参和工具结果在 Python dict 与 JSON 字符串之间转换。
import json

# typing 中的 Any 表示“任意类型”，Callable 表示“可调用对象”，用于类型标注。
from typing import Any, Callable

# dotenv 用于从项目根目录的 .env 文件读取环境变量。
from dotenv import load_dotenv

# OpenAI 是官方 Python SDK 中的客户端类；很多国产/本地模型服务也兼容这个接口。
from openai import OpenAI


# 加载 .env 文件中的配置；override=False 表示如果系统环境变量已存在，不强行覆盖。
load_dotenv(override=False)


class FunctionCallAgent:
    """
    一个最小 function calling Agent。

    这个类故意不使用 LangChain/LangGraph，目的是明确 function calling 的底层流程：
        1. 把工具 JSON Schema 发给模型；
        2. 模型返回 tool_calls；
        3. Python 代码真正执行工具；
        4. 把工具结果发回模型；
        5. 模型生成最终自然语言回答。
    """

    def __init__(
        self,
        model: str | None = None,
        apiKey: str | None = None,
        baseUrl: str | None = None,
        timeout: int | None = None,
    ) -> None:
        """
        初始化 Agent，并创建 OpenAI 兼容客户端。

        参数说明：
            model: 模型 ID；如果不传，就从环境变量 LLM_MODEL_ID 读取。
            apiKey: API Key；如果不传，就从环境变量 LLM_API_KEY 读取。
            baseUrl: OpenAI 兼容服务地址；如果不传，就从环境变量 LLM_BASE_URL 读取。
            timeout: 超时时间；如果不传，就从环境变量 LLM_TIMEOUT 读取，默认 60 秒。
        """

        # self.model 保存模型名称；or 表示左边为空时使用右边。
        self.model = model or os.getenv("LLM_MODEL_ID")

        # apiKey 保存 API 密钥；这里变量名沿用你给出的写法。
        apiKey = apiKey or os.getenv("LLM_API_KEY")

        # baseUrl 保存 OpenAI 兼容接口的基础地址，例如 https://api.openai.com/v1。
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")

        # timeout 优先使用参数，其次读取环境变量，最后使用 60 秒。
        timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60))

        # all([...]) 会检查列表中每个值是否都为真；任何一个为空就说明配置不完整。
        if not all([self.model, apiKey, baseUrl]):
            # raise ValueError 用于主动抛出参数错误，避免后续请求时才报难懂的错。
            raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。")

        # 创建 OpenAI 兼容客户端；base_url 允许你接入非 OpenAI 官方但兼容 OpenAI 协议的服务。
        self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout)

        # 保存 Python 侧真实可执行的工具函数；key 是工具名，value 是函数对象。
        # 把工具函数注册到一个字典里，是函数本身，并非可执行的函数结果。
        # 因为 function calling 里，模型返回的通常是一个工具名，Agent 需要根据工具名找到真实函数。
        # Callable[..., Any]表示一个可以被调用的对象；参数数量和参数类型不限；返回值类型也不限。
        self.tool_functions: dict[str, Callable[..., Any]] = {
            "get_weather": self.get_weather,
            "celsius_to_fahrenheit": self.celsius_to_fahrenheit,
        }

        # 保存要发给模型看的工具定义；模型只看到 JSON Schema，不会直接看到 Python 函数源码。
        self.tools = self.build_openai_tools_schema()

    def get_weather(self, city: str) -> dict[str, Any]:
        """
        演示用天气查询工具。

        参数：
            city: 城市名，例如“北京”“上海”“深圳”。

        返回：
            一个 dict，里面包含城市、天气、摄氏温度、湿度等信息。

        注意：
            这里不用真实天气 API，是为了让 demo 不依赖外部网络。
            真实项目中可以把这里替换成数据库查询、HTTP API、RAG 检索等。
        """

        # demo_weather_db 是一个假的天气数据库；key 是城市名，value 是天气信息。
        demo_weather_db = {
            "北京": {"condition": "晴", "temperature_c": 30.0, "humidity": "35%"},
            "上海": {"condition": "小雨", "temperature_c": 27.0, "humidity": "78%"},
            "深圳": {"condition": "多云", "temperature_c": 32.0, "humidity": "70%"},
            "杭州": {"condition": "阴", "temperature_c": 28.0, "humidity": "65%"},
        }

        # dict.get(key, default) 表示如果 city 存在就取对应值，否则取 default 默认值。
        weather = demo_weather_db.get(
            city,
            {"condition": "未知", "temperature_c": 26.0, "humidity": "未知"},
        )

        # 返回结构化结果；模型后续会根据这个 JSON 内容组织最终回答。
        return {
            "city": city,
            "condition": weather["condition"],
            "temperature_c": weather["temperature_c"],
            "humidity": weather["humidity"],
            "source": "demo_local_weather_db",
        }

    def celsius_to_fahrenheit(self, celsius: float) -> dict[str, float]:
        """
        摄氏温度转华氏温度工具。

        参数：
            celsius: 摄氏温度，例如 30.0。

        返回：
            一个 dict，包含摄氏温度和换算后的华氏温度。
        """

        # 摄氏转华氏公式：F = C * 9 / 5 + 32。
        fahrenheit = celsius * 9 / 5 + 32

        # round(x, 2) 表示保留两位小数，便于最终回答展示。
        return {"celsius": celsius, "fahrenheit": round(fahrenheit, 2)}

    def build_openai_tools_schema(self) -> list[dict[str, Any]]:
        """
        构造 OpenAI tools 参数需要的 JSON Schema。

        function calling 的关键点：
            - Python 函数不能直接发给模型；
            - 要把函数名、函数说明、参数结构写成 JSON Schema；
            - 模型根据这个 schema 判断何时调用哪个工具，以及生成什么参数。
        """

        # 返回一个列表；每个元素都是一个工具定义。
        return [
            {
                # type=function 表示这是一个传统函数工具。
                "type": "function",
                # function 字段描述具体函数的名字、用途、参数格式。
                "function": {
                    # name 必须与 self.tool_functions 中的 key 一致，否则 Python 侧找不到函数。
                    "name": "get_weather",
                    # description 会被模型读取，用来判断什么时候应该调用这个工具。
                    "description": "查询指定城市的演示天气信息，返回天气、摄氏温度和湿度。",
                    # parameters 是 JSON Schema，描述函数参数。
                    "parameters": {
                        # type=object 表示参数整体是一个 JSON 对象。
                        "type": "object",
                        # properties 描述对象中的字段。
                        "properties": {
                            "city": {
                                # city 字段类型是字符串。
                                "type": "string",
                                # 字段说明会影响模型填参质量。
                                "description": "城市名，例如：北京、上海、深圳、杭州。",
                            }
                        },
                        # required 表示 city 是必填参数。
                        "required": ["city"],
                    },
                },
            },
            {
                "type": "function",
                "function": {
                    "name": "celsius_to_fahrenheit",
                    "description": "把摄氏温度转换为华氏温度。",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "celsius": {
                                # number 表示整数或小数都可以。
                                "type": "number",
                                "description": "摄氏温度数值，例如 30.0。",
                            }
                        },
                        "required": ["celsius"],
                    },
                },
            },
        ]

    def build_messages(self, user_question: str) -> list[dict[str, Any]]:
        """
        构造对话消息。

        参数：
            user_question: 用户输入的问题。

        返回：
            messages 列表，符合 OpenAI Chat Completions 的消息格式。
        """

        # system prompt 定义 Agent 行为边界：需要数据时必须调用工具，不能瞎编。
        system_prompt = (
            "你是一个天气助手 Agent。"
            "当用户需要天气或温度换算时，你必须优先调用可用工具获取数据，"
            "不要自己编造天气和温度。最后用中文给出简洁回答。"
        )

        # messages 是发送给模型的上下文；role=system 是系统指令，role=user 是用户问题。
        return [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_question},
        ]

    def call_llm(self, messages: list[dict[str, Any]]) -> Any:
        """
        调用大模型。

        参数：
            messages: 当前完整对话上下文。

        返回：
            模型返回的 assistant message，可能包含 content，也可能包含 tool_calls。
        """

        # self.client.chat.completions.create 是 Chat Completions API 的创建方法。
        response = self.client.chat.completions.create(
            # model 指定使用哪个模型。
            model=self.model,
            # messages 是当前对话历史。
            messages=messages,
            # tools 把可用工具的 schema 提供给模型。
            tools=self.tools,
            # tool_choice="auto" 表示让模型自己决定是否调用工具。
            tool_choice="auto",
        )

        # choices[0].message 是第一条候选回答；一般 demo 只取第一条。
        return response.choices[0].message

    def run_tool_call(self, tool_call: Any) -> str:
        """
        执行模型请求的某一次工具调用。

        参数：
            tool_call: 模型返回的工具调用对象，里面包含函数名和 JSON 字符串参数。

        返回：
            工具执行结果的 JSON 字符串，用于作为 role=tool 的消息发回模型。
        """

        # tool_call.function.name 是模型想调用的函数名。
        tool_name = tool_call.function.name

        # tool_call.function.arguments 是 JSON 字符串，例如 '{"city":"北京"}'。
        raw_arguments = tool_call.function.arguments

        # json.loads 把 JSON 字符串转成 Python dict。
        arguments = json.loads(raw_arguments)

        # 根据函数名从工具注册表中取出真正的 Python 函数。
        tool_function = self.tool_functions.get(tool_name)

        # 如果模型调用了不存在的工具，返回错误信息，而不是让程序崩溃。
        if tool_function is None:
            return json.dumps({"error": f"未知工具：{tool_name}"}, ensure_ascii=False)

        # **arguments 表示把 dict 展开成关键字参数，例如 {"city":"北京"} -> city="北京"。
        result = tool_function(**arguments)

        # json.dumps 把 Python dict 转成 JSON 字符串；ensure_ascii=False 保证中文不变成 Unicode 编码。
        return json.dumps(result, ensure_ascii=False)

    def run(self, user_question: str) -> str:
        """
        运行完整 Agent 流程。

        参数：
            user_question: 用户问题。

        返回：
            模型最终生成的中文回答。
        """

        # 第一步：构造初始 messages。
        messages = self.build_messages(user_question)

        # 最多循环 5 轮，防止模型反复调用工具导致死循环。
        for _ in range(5):
            # 第二步：把当前 messages 和工具 schema 发给模型。
            assistant_message = self.call_llm(messages)

            # 第三步：把 assistant 消息加入对话历史；model_dump 可把 SDK 对象转成普通 dict。
            messages.append(assistant_message.model_dump(exclude_none=True))

            # 第四步：如果没有 tool_calls，说明模型已经给出最终回答。
            if not assistant_message.tool_calls:
                # assistant_message.content 就是最终自然语言文本。
                return assistant_message.content or ""

            # 第五步：模型可能一次请求多个工具调用，所以这里逐个执行。
            for tool_call in assistant_message.tool_calls:
                # 真正执行 Python 工具函数，得到 JSON 字符串结果。
                tool_result_json = self.run_tool_call(tool_call)

                # 把工具结果作为 role=tool 的消息追加回上下文。
                messages.append(
                    {
                        # role=tool 表示这条消息是工具执行结果，不是用户或模型说的话。
                        "role": "tool",
                        # tool_call_id 用于告诉模型：这个结果对应刚才哪一次工具调用。
                        "tool_call_id": tool_call.id,
                        # content 是工具返回内容，通常写 JSON 字符串。
                        "content": tool_result_json,
                    }
                )

        # 如果循环 5 次仍没有最终回答，返回一个保护性提示。
        return "工具调用轮数过多，Agent 已停止。"


# Python 约定：只有直接运行本文件时，__name__ 才等于 "__main__"。
if __name__ == "__main__":
    # 实例化 Agent；不传参数时，会自动读取 .env。
    agent = FunctionCallAgent()

    # 准备一个 demo 问题；你可以改成 input("请输入问题：") 做成命令行交互。
    question = "请查询北京天气，并把摄氏温度换算成华氏温度，最后给出穿衣建议。"

    # 运行 Agent，拿到最终回答。
    answer = agent.run(question)

    # 打印最终回答。
    print("\n===== Function Calling Agent 最终回答 =====")
    print(answer)

===== Function Calling Agent 最终回答 =====
北京天气：晴，30°C，湿度 35%。  
换算成华氏温度：86°F。  

穿衣建议：天气较热，建议穿短袖、薄款透气衣物，注意防晒，多补水；如果长时间在户外，可以带帽子或太阳镜。

"""
功能：
    使用 LangChain 框架 + @tool 装饰器实现一个最小 Agent。

场景：
    用户问：“请查询北京天气，并把摄氏温度换算成华氏温度，最后给出穿衣建议。”

    Agent 会让模型自己决定是否调用工具：
        1. get_weather(city)：查询演示天气数据
        2. celsius_to_fahrenheit(celsius)：摄氏温度转华氏温度

安装依赖：
    pip install langchain langchain-openai python-dotenv

.env 示例：
    LLM_MODEL_ID=你的模型名
    LLM_API_KEY=你的API密钥
    LLM_BASE_URL=你的OpenAI兼容接口地址
    LLM_TIMEOUT=60
"""

# os 用于读取系统环境变量，例如 LLM_MODEL_ID、LLM_API_KEY、LLM_BASE_URL。
import os

# json 用于把工具调用参数、工具返回结果在 Python 对象和 JSON 字符串之间转换。
import json

# Any 表示任意类型，用于类型注解。
from typing import Any

# dotenv 用于从当前项目目录下的 .env 文件中加载环境变量。
from dotenv import load_dotenv

# ChatOpenAI 是 LangChain 对 OpenAI Chat Model 的封装。
# 它不仅支持 OpenAI 官方接口，也支持很多 OpenAI-compatible 的模型服务。
from langchain_openai import ChatOpenAI

# @tool 是 LangChain 提供的工具注册装饰器。
# 用 @tool 包裹普通 Python 函数后，这个函数就会变成 LangChain Tool。
from langchain_core.tools import tool

# SystemMessage 表示系统提示词。
# HumanMessage 表示用户消息。
# ToolMessage 表示工具执行结果消息。
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage


# load_dotenv 会读取 .env 文件，并把里面的配置加载到 os.environ 中。
# override=False 表示如果系统环境变量中已经存在同名变量，就不覆盖。
load_dotenv(override=False)


@tool
def get_weather(city: str) -> dict[str, Any]:
    """
    查询指定城市的演示天气信息，返回天气、摄氏温度和湿度。

    Args:
        city: 城市名，例如：北京、上海、深圳、杭州。
    """

    # demo_weather_db 是一个假的本地天气数据库。
    # key 是城市名，value 是该城市的天气信息。
    demo_weather_db = {
        "北京": {"condition": "晴", "temperature_c": 30.0, "humidity": "35%"},
        "上海": {"condition": "小雨", "temperature_c": 27.0, "humidity": "78%"},
        "深圳": {"condition": "多云", "temperature_c": 32.0, "humidity": "70%"},
        "杭州": {"condition": "阴", "temperature_c": 28.0, "humidity": "65%"},
    }

    # dict.get(key, default) 表示：
    # 如果 city 在 demo_weather_db 中存在，就返回对应天气；
    # 如果不存在，就返回后面的默认天气。
    weather = demo_weather_db.get(
        city,
        {"condition": "未知", "temperature_c": 26.0, "humidity": "未知"},
    )

    # 返回结构化结果。
    # LangChain 会把这个结果作为工具执行结果，再交还给模型。
    return {
        "city": city,
        "condition": weather["condition"],
        "temperature_c": weather["temperature_c"],
        "humidity": weather["humidity"],
        "source": "demo_local_weather_db",
    }


@tool
def celsius_to_fahrenheit(celsius: float) -> dict[str, float]:
    """
    把摄氏温度转换为华氏温度。

    Args:
        celsius: 摄氏温度数值，例如 30.0。
    """

    # 摄氏转华氏公式：
    # F = C * 9 / 5 + 32
    fahrenheit = celsius * 9 / 5 + 32

    # round(fahrenheit, 2) 表示把结果保留两位小数。
    return {
        "celsius": celsius,
        "fahrenheit": round(fahrenheit, 2),
    }


class LangChainToolAgent:
    """
    一个最小 LangChain Tool Calling Agent。

    和之前的 function calling 版本相比：

        原来的写法：
            1. 手写 tools JSON Schema；
            2. 手写 self.tool_functions 字典；
            3. 手动根据 tool name 找 Python 函数；
            4. 手动 json.loads 解析参数；
            5. 手动 tool_function(**arguments) 调用函数。

        LangChain 写法：
            1. 用 @tool 注册工具；
            2. LangChain 根据函数名、类型注解、docstring 生成工具 schema；
            3. 用 llm.bind_tools(tools) 把工具绑定到模型；
            4. 模型返回 tool_calls；
            5. 根据 tool name 找到对应 LangChain Tool；
            6. 用 selected_tool.invoke(args) 执行工具。
    """

    def __init__(
        self,
        model: str | None = None,
        apiKey: str | None = None,
        baseUrl: str | None = None,
        timeout: int | None = None,
    ) -> None:
        """
        初始化 Agent，并创建 LangChain 的 ChatOpenAI 客户端。

        参数说明：
            model: 模型 ID；如果不传，就从环境变量 LLM_MODEL_ID 读取。
            apiKey: API Key；如果不传，就从环境变量 LLM_API_KEY 读取。
            baseUrl: OpenAI 兼容服务地址；如果不传，就从环境变量 LLM_BASE_URL 读取。
            timeout: 超时时间；如果不传，就从环境变量 LLM_TIMEOUT 读取，默认 60 秒。
        """

        # self.model 保存模型名称。
        # 如果构造函数传入了 model，就用传入值；
        # 否则读取环境变量 LLM_MODEL_ID。
        self.model = model or os.getenv("LLM_MODEL_ID")

        # apiKey 保存 API 密钥。
        # 如果构造函数传入了 apiKey，就用传入值；
        # 否则读取环境变量 LLM_API_KEY。
        apiKey = apiKey or os.getenv("LLM_API_KEY")

        # baseUrl 保存 OpenAI-compatible 接口地址。
        # 例如：
        #   https://api.openai.com/v1
        #   http://localhost:8000/v1
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")

        # timeout 保存请求超时时间。
        # 优先使用构造函数参数；
        # 其次读取环境变量 LLM_TIMEOUT；
        # 如果都没有，就默认 60 秒。
        timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60))

        # all([...]) 会判断列表里的每个元素是否都为真。
        # 如果 self.model、apiKey、baseUrl 任意一个为空，就说明配置不完整。
        if not all([self.model, apiKey, baseUrl]):
            # 主动抛出 ValueError，避免后面请求模型时才出现更难排查的错误。
            raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。")

        # 创建 LangChain ChatOpenAI 模型对象。
        # model 指定模型名称。
        # api_key 指定 API 密钥。
        # base_url 指定 OpenAI-compatible 接口地址。
        # timeout 指定请求超时时间。
        self.llm = ChatOpenAI(
            model=self.model,
            api_key=apiKey,
            base_url=baseUrl,
            timeout=timeout,
        )

        # self.tools 保存所有 LangChain 工具对象。
        # 注意：
        #   get_weather 和 celsius_to_fahrenheit 已经不是普通函数，
        #   因为它们上面加了 @tool；
        #   它们现在是 LangChain Tool 对象。
        self.tools = [
            get_weather,
            celsius_to_fahrenheit,
        ]

        # tool_map 是一个工具名字到工具对象的映射表。
        # key 是工具名，例如 "get_weather"；
        # value 是 LangChain Tool 对象。
        #
        # 这和原来 self.tool_functions 的作用类似，
        # 但这里保存的是 LangChain Tool，而不是原始 Python 函数。
        self.tool_map = {
            single_tool.name: single_tool
            for single_tool in self.tools
        }

        # bind_tools 会把工具绑定到模型上。
        # 绑定后，模型在 invoke 时就能“看到”这些工具的 schema。
        #
        # LangChain 会根据 @tool 函数的：
        #   1. 函数名；
        #   2. 参数类型注解；
        #   3. docstring；
        # 自动生成工具 schema。
        self.llm_with_tools = self.llm.bind_tools(self.tools)

    def build_messages(self, user_question: str) -> list[Any]:
        """
        构造 LangChain 消息列表。

        参数：
            user_question: 用户输入的问题。

        返回：
            一个 LangChain Message 列表。
        """

        # system_prompt 是系统提示词。
        # 它告诉模型这个 Agent 的行为规则。
        system_prompt = (
            "你是一个天气助手 Agent。"
            "当用户需要天气或温度换算时，你必须优先调用可用工具获取数据，"
            "不要自己编造天气和温度。最后用中文给出简洁回答。"
        )

        # 返回消息列表。
        # SystemMessage 对应 role=system。
        # HumanMessage 对应 role=user。
        return [
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_question),
        ]

    def print_tools_info(self) -> None:
        """
        打印当前 Agent 可用的工具信息。

        这一步不是 Agent 必须的，
        只是为了更直观看到 @tool 注册后，LangChain 保存了哪些工具信息。
        """

        # 打印标题。
        print("\n===== LangChain 已注册工具 =====")

        # 遍历 self.tools 中的每一个工具对象。
        for single_tool in self.tools:
            # single_tool.name 是工具名。
            print(f"工具名: {single_tool.name}")

            # single_tool.description 是工具描述。
            # 这个描述通常来自函数 docstring。
            print(f"工具说明: {single_tool.description}")

            # single_tool.args 是工具参数 schema。
            # LangChain 根据函数参数类型注解自动生成。
            print(f"工具参数: {single_tool.args}")

            # 打印分隔线。
            print("-" * 40)

    def run_one_tool_call(self, tool_call: dict[str, Any]) -> ToolMessage:
        """
        执行模型返回的一次工具调用。

        参数：
            tool_call: 模型返回的工具调用信息。
                       LangChain 中通常是一个 dict，
                       里面包含 name、args、id 等字段。

        返回：
            ToolMessage，表示工具执行结果消息。
        """

        # tool_call["name"] 是模型想要调用的工具名。
        tool_name = tool_call["name"]

        # tool_call["args"] 是模型为工具生成的参数。
        # 例如：
        #   {"city": "北京"}
        #   {"celsius": 30.0}
        tool_args = tool_call["args"]

        # tool_call["id"] 是这次工具调用的唯一 ID。
        # ToolMessage 需要带上这个 ID，
        # 这样模型才能知道工具结果对应刚才哪一次 tool call。
        # tool_call["id"] 是这一次工具调用的 ID，不是工具函数本身的 ID。
        tool_call_id = tool_call["id"]

        # 根据工具名，从 tool_map 中找到对应的 LangChain Tool。
        selected_tool = self.tool_map.get(tool_name)

        # 如果模型调用了不存在的工具，就构造一个错误结果。
        if selected_tool is None:
            # 把错误信息转成 JSON 字符串。
            error_content = json.dumps(
                {"error": f"未知工具：{tool_name}"},
                ensure_ascii=False,
            )

            # 返回 ToolMessage。
            return ToolMessage(
                content=error_content,
                tool_call_id=tool_call_id,
            )

        # selected_tool.invoke(tool_args) 会真正执行工具。
        # 对于 get_weather，相当于执行：
        #   get_weather.invoke({"city": "北京"})
        #
        # 对于 celsius_to_fahrenheit，相当于执行：
        #   celsius_to_fahrenheit.invoke({"celsius": 30.0})
        result = selected_tool.invoke(tool_args)

        # 工具结果可能是 dict，也可能是 str、int、float 等。
        # 为了统一传回模型，这里转成 JSON 字符串。
        result_json = json.dumps(result, ensure_ascii=False)

        # 构造 ToolMessage。
        # content 是工具执行结果。
        # tool_call_id 用于对应模型刚才的工具调用 ID。
        return ToolMessage(
            content=result_json,
            tool_call_id=tool_call_id,
        )

    def run(self, user_question: str) -> str:
        """
        运行完整 Agent 流程。

        参数：
            user_question: 用户问题。

        返回：
            模型最终生成的中文回答。
        """

        # 第一步：构造初始消息。
        messages = self.build_messages(user_question)

        # 最多循环 5 轮。
        # 这是为了防止模型一直调用工具，导致死循环。
        for _ in range(5):
            # 第二步：调用绑定了工具的模型。
            # 如果模型认为需要工具，它会返回 tool_calls。
            # 如果模型认为不需要工具，它会直接返回 content。
            assistant_message = self.llm_with_tools.invoke(messages)

            # 第三步：把模型消息加入历史。
            # 这一步非常重要：
            # 后续工具结果必须接在这个 assistant_message 后面，
            # 模型才能理解上下文。
            messages.append(assistant_message)

            # 第四步：检查模型是否请求了工具调用。
            # 在 LangChain 中，assistant_message.tool_calls 通常是一个 list。
            # 如果为空，说明模型已经给出最终回答。
            if not assistant_message.tool_calls:
                # assistant_message.content 就是模型最终回答。
                return str(assistant_message.content or "")

            # 第五步：模型可能一次性请求多个工具调用，所以这里逐个执行。
            for tool_call in assistant_message.tool_calls:
                # 执行单个工具调用，并得到 ToolMessage。
                tool_message = self.run_one_tool_call(tool_call)

                # 把工具结果加入消息历史。
                # 下一轮模型调用时，模型会看到工具结果，并决定继续调用工具或生成最终回答。
                messages.append(tool_message)

        # 如果 5 轮后仍然没有得到最终回答，就返回保护性提示。
        return "工具调用轮数过多，Agent 已停止。"
    

if __name__ == "__main__":
    # 创建 Agent 实例。
    # 不传参数时，会自动从 .env 文件或系统环境变量读取配置。
    agent = LangChainToolAgent()

    # 打印 @tool 注册后的工具信息。
    # 这一步方便观察 LangChain 根据函数注解和 docstring 生成了什么工具信息。
    agent.print_tools_info()

    # 准备一个 demo 问题。
    question = "请查询北京天气，并把摄氏温度换算成华氏温度，最后给出穿衣建议。"

    # 运行 Agent，得到最终回答。
    answer = agent.run(question)

    # 打印最终回答。
    print("\n===== LangChain Tool Calling Agent 最终回答 =====")
    print(answer)

===== LangChain 已注册工具 =====
工具名: get_weather
工具说明: 查询指定城市的演示天气信息，返回天气、摄氏温度和湿度。

Args:
    city: 城市名，例如：北京、上海、深圳、杭州。
工具参数: {'city': {'title': 'City', 'type': 'string'}}
----------------------------------------
工具名: celsius_to_fahrenheit
工具说明: 把摄氏温度转换为华氏温度。

Args:
    celsius: 摄氏温度数值，例如 30.0。
工具参数: {'celsius': {'title': 'Celsius', 'type': 'number'}}
----------------------------------------

===== LangChain Tool Calling Agent 最终回答 =====
北京天气：晴，30°C，湿度35%  
换算后：86°F

穿衣建议：天气偏热，建议穿轻薄短袖、透气衣物，注意防晒，外出可带水；如果长时间在户外，建议备一件薄外套防空调。

"""
功能：
    使用 MCP（Model Context Protocol）实现与 function_call_agent.py 相同的 Agent 功能。

场景：
    用户问：“请查询北京天气，并把摄氏温度换算成华氏温度，最后给出穿衣建议。”
    Agent 通过 MCP 连接工具服务器，发现工具，再让模型选择并调用工具。

为什么这个文件既是 MCP server 又是 MCP client：
    本文件使用命令行参数区分两种模式：
        python mcp_agent.py --server   启动 MCP 工具服务器
        python mcp_agent.py            启动 MCP Agent 客户端，并自动用子进程拉起 server

注意：
    MCP 负责“工具服务器协议化、工具发现、工具调用”。
    LLM 仍然负责“根据用户问题决定调用哪个工具”。
"""

# asyncio 用于运行异步 MCP client，因为 MCP 的连接和调用都是异步接口。
import asyncio

# os 用于读取 .env 或系统环境变量。
import os

# sys 用于读取命令行参数，也用于拿到当前 Python 解释器路径。
import sys

# json 用于解析模型生成的工具参数，以及序列化工具结果。
import json

# typing.Any 表示任意类型，主要用于函数返回值和消息结构的类型标注。
from typing import Any

# dotenv 用于读取 .env 文件中的 LLM_MODEL_ID、LLM_API_KEY、LLM_BASE_URL 等配置。
from dotenv import load_dotenv

# OpenAI 是 OpenAI Python SDK 客户端，也可连接 OpenAI 兼容服务。
from openai import OpenAI

# ClientSession 是 MCP 客户端会话对象；StdioServerParameters 描述如何通过 stdio 启动/连接 MCP server。
from mcp import ClientSession, StdioServerParameters, types

# stdio_client 是 MCP 的 stdio 传输客户端；它会启动子进程并通过标准输入输出通信。
from mcp.client.stdio import stdio_client

# FastMCP 是 MCP Python SDK 提供的快速 server 开发封装，可用装饰器暴露 tools/resources/prompts。
from mcp.server.fastmcp import FastMCP
# 用了 官方 MCP Python SDK 里内置的 FastMCP


# 加载 .env 文件；override=False 表示不覆盖已经存在的系统环境变量。
load_dotenv(override=False)


# 创建 MCP server 对象；名字 WeatherMCPDemo 会出现在 MCP 客户端或 Inspector 中。
# json_response=True 让工具更倾向于返回结构化 JSON，便于客户端解析。
mcp = FastMCP("WeatherMCPDemo", json_response=True)


@mcp.tool()
def get_weather(city: str) -> dict[str, Any]:
    """
    MCP 工具：查询指定城市的演示天气信息。

    参数：
        city: 城市名，例如“北京”“上海”“深圳”“杭州”。

    返回：
        dict 格式的天气信息。

    说明：
        @mcp.tool() 会把这个普通 Python 函数注册为 MCP 工具。
        MCP server 会根据函数签名 city: str 和 docstring 自动生成工具 schema。
    """

    # 这里使用本地假数据，避免 demo 依赖真实天气 API。
    demo_weather_db = {
        "北京": {"condition": "晴", "temperature_c": 30.0, "humidity": "35%"},
        "上海": {"condition": "小雨", "temperature_c": 27.0, "humidity": "78%"},
        "深圳": {"condition": "多云", "temperature_c": 32.0, "humidity": "70%"},
        "杭州": {"condition": "阴", "temperature_c": 28.0, "humidity": "65%"},
    }

    # 如果 city 不在假数据库中，就返回一个默认天气，保证 demo 不会报 KeyError。
    weather = demo_weather_db.get(
        city,
        {"condition": "未知", "temperature_c": 26.0, "humidity": "未知"},
    )

    # 返回结构化 dict；MCP 会把它包装成 CallToolResult 返回给 client。
    return {
        "city": city,
        "condition": weather["condition"],
        "temperature_c": weather["temperature_c"],
        "humidity": weather["humidity"],
        "source": "demo_mcp_weather_server",
    }


@mcp.tool()
def celsius_to_fahrenheit(celsius: float) -> dict[str, float]:
    """
    MCP 工具：把摄氏温度转换为华氏温度。

    参数：
        celsius: 摄氏温度。

    返回：
        dict，包含原摄氏温度和华氏温度。
    """

    # 摄氏转华氏公式：F = C * 9 / 5 + 32。
    fahrenheit = celsius * 9 / 5 + 32

    # 返回结构化结果；round(..., 2) 用于保留两位小数。
    return {"celsius": celsius, "fahrenheit": round(fahrenheit, 2)}


@mcp.prompt()
def weather_agent_prompt(user_question: str) -> str:
    """
    MCP Prompt：返回一个可复用的天气 Agent 提示模板。

    参数：
        user_question: 用户原始问题。

    返回：
        一段提示词文本。

    说明：
        @mcp.prompt() 和 @mcp.tool() 类似，但它暴露的不是可执行函数，
        而是“提示模板”。MCP client 可以先发现 prompt，再按需获取 prompt 内容。
    """

    # 这里返回系统提示词；client 会把它放进 OpenAI messages 的 system 位置。
    return (
        "你是一个天气助手 Agent。"
        "你可以使用 MCP server 提供的工具查询天气和换算温度。"
        "当用户需要天气或温度换算时，必须优先调用工具，不要编造数据。"
        "最后用中文回答，并给出简洁穿衣建议。\n\n"
        f"用户问题：{user_question}"
    )


class MCPAgent:
    """
    一个最小 MCP Agent。

    它和 function calling 版的核心区别：
        - function calling 版：工具函数和工具 schema 都写在 Agent 代码内部；
        - MCP 版：工具函数放在 MCP server 中，Agent 先连接 server，再动态发现工具。

    也就是说，MCP 把“工具提供者”和“Agent 使用者”解耦了。
    """

    def __init__(
        self,
        model: str | None = None,
        apiKey: str | None = None,
        baseUrl: str | None = None,
        timeout: int | None = None,
    ) -> None:
        """
        初始化 MCP Agent，并创建 OpenAI 兼容客户端。

        参数说明和 function_call_agent.py 完全一致。
        """

        # 读取模型 ID；优先使用参数，其次使用环境变量。
        self.model = model or os.getenv("LLM_MODEL_ID")

        # 读取 API Key；变量名 apiKey 沿用你提供的写法。
        apiKey = apiKey or os.getenv("LLM_API_KEY")

        # 读取 OpenAI 兼容服务地址。
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")

        # 读取超时时间；没有配置时默认 60 秒。
        timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60))

        # 验证必要参数，避免请求时才报错。
        if not all([self.model, apiKey, baseUrl]):
            raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。")

        # 创建 OpenAI 兼容客户端；后续仍用它来驱动模型决策。
        self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout)

    def build_server_params(self) -> StdioServerParameters:
        """
        构造 MCP stdio server 启动参数。

        返回：
            StdioServerParameters 对象，告诉 MCP client 如何启动 server。
        """

        # sys.executable 是当前 Python 解释器路径，保证 client 和 server 使用同一个虚拟环境。
        python_executable = sys.executable

        # __file__ 是当前这个 mcp_agent.py 文件路径。
        current_file = __file__

        # StdioServerParameters 描述一个通过标准输入/输出通信的 MCP server 子进程。
        return StdioServerParameters(
            # command 是要执行的命令，这里就是当前 Python 解释器。
            command=python_executable,
            # args 是命令参数；相当于运行：python mcp_agent.py --server。
            args=[current_file, "--server"],
            # env 把当前环境变量传给子进程，保证 server 也能读取必要配置。
            env=os.environ.copy(),
        )

    async def print_mcp_capabilities(self, session: ClientSession) -> None:
        """
        打印 MCP server 暴露的 prompts 和 tools。

        参数：
            session: 已初始化的 MCP ClientSession。
        """

        # list_prompts 会向 MCP server 请求可用 prompt 列表。
        prompts_response = await session.list_prompts()

        # 打印 prompt 列表，方便观察 MCP 的“发现能力”。
        print("\n===== MCP Server 可用 Prompts =====")
        for prompt in prompts_response.prompts:
            # prompt.name 是 prompt 名称；prompt.description 是 docstring 生成的说明。
            print(f"- {prompt.name}: {prompt.description}")

        # list_tools 会向 MCP server 请求可用 tool 列表。
        tools_response = await session.list_tools()

        # 打印工具名称、描述和输入 schema，帮助看清 MCP 如何描述工具。
        print("\n===== MCP Server 可用 Tools =====")
        for tool in tools_response.tools:
            print(f"- name: {tool.name}")
            print(f"  description: {tool.description}")
            print(f"  inputSchema: {json.dumps(tool.inputSchema, ensure_ascii=False)}")

    async def get_prompt_from_mcp(self, session: ClientSession, user_question: str) -> str:
        """
        从 MCP server 获取 prompt 模板。

        参数：
            session: MCP 客户端会话。
            user_question: 用户问题。

        返回：
            prompt 文本，用于放入 OpenAI messages。
        """

        # get_prompt 调用 MCP prompt，并传入 prompt 所需参数。
        prompt_result = await session.get_prompt(
            # 这里的名字必须与 @mcp.prompt() 装饰的函数名一致。
            "weather_agent_prompt",
            # arguments 是 prompt 参数字典。
            arguments={"user_question": user_question},
        )

        # prompt_result.messages 是 MCP 返回的一组 prompt message。
        first_message = prompt_result.messages[0]

        # first_message.content 通常是 TextContent，其中 text 字段是真正文本。
        content = first_message.content

        # isinstance 用于判断 content 是否为 MCP TextContent 类型。
        if isinstance(content, types.TextContent):
            return content.text

        # 兜底：如果 SDK 返回了其他内容类型，就转成字符串。
        return str(content)

    async def convert_mcp_tools_to_openai_tools(self, session: ClientSession) -> list[dict[str, Any]]:
        """
        把 MCP tools 转换成 OpenAI Chat Completions tools 格式。

        参数：
            session: MCP 客户端会话。

        返回：
            OpenAI tools 列表。

        为什么需要转换：
            MCP server 能发现工具，但 OpenAI Chat Completions API 需要的是自己的 tools JSON 格式。
            因此 Agent 客户端要做一层适配。
        """

        # 从 MCP server 动态获取工具列表。
        tools_response = await session.list_tools()

        # 准备一个空列表，用来保存转换后的 OpenAI tool schema。
        openai_tools: list[dict[str, Any]] = []

        # 遍历 MCP server 暴露的每一个工具。
        for tool in tools_response.tools:
            # MCP v1 中工具入参 schema 字段通常叫 inputSchema。
            input_schema = tool.inputSchema

            # 按 OpenAI Chat Completions 的格式组织 function tool。
            openai_tools.append(
                {
                    "type": "function",
                    "function": {
                        # 工具名沿用 MCP tool.name。
                        "name": tool.name,
                        # 工具说明沿用 MCP tool.description；如果为空则给空字符串。
                        "description": tool.description or "",
                        # 参数 schema 直接使用 MCP 生成的 inputSchema。
                        "parameters": input_schema,
                    },
                }
            )

        # 返回可传给 OpenAI API 的工具定义。
        return openai_tools

    def call_llm(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> Any:
        """
        调用 LLM，让模型基于 MCP 工具 schema 决定是否 tool call。

        参数：
            messages: 对话历史。
            tools: 从 MCP tools 转换来的 OpenAI tools。

        返回：
            assistant message，可能包含自然语言 content，也可能包含 tool_calls。
        """

        # 调用 Chat Completions API。
        response = self.client.chat.completions.create(
            # 指定模型。
            model=self.model,
            # 传入当前上下文。
            messages=messages,
            # 传入从 MCP server 动态发现并转换来的工具定义。
            tools=tools,
            # 让模型自动决定是否调用工具。
            tool_choice="auto",
        )

        # 取第一条候选输出。
        return response.choices[0].message

    async def call_mcp_tool(self, session: ClientSession, tool_name: str, arguments: dict[str, Any]) -> str:
        """
        调用 MCP server 上的某个工具。

        参数：
            session: MCP 客户端会话。
            tool_name: 工具名，例如 get_weather。
            arguments: 工具参数，例如 {"city": "北京"}。

        返回：
            JSON 字符串，用作 OpenAI role=tool 的消息内容。
        """

        # session.call_tool 会通过 MCP 协议把工具名和参数发给 server 执行。
        result = await session.call_tool(tool_name, arguments=arguments)

        # 如果工具执行报错，isError 通常会为 True；这里转成 JSON 返回给模型。
        if getattr(result, "isError", False):
            return json.dumps({"error": self.extract_mcp_text_result(result)}, ensure_ascii=False)

        # MCP v1 的结构化结果字段通常是 structuredContent。
        structured_content = getattr(result, "structuredContent", None)

        # 如果存在结构化结果，优先返回结构化 JSON。
        if structured_content:
            return json.dumps(structured_content, ensure_ascii=False)

        # 兜底：如果没有 structuredContent，就从 content 中提取文本。
        return json.dumps({"text": self.extract_mcp_text_result(result)}, ensure_ascii=False)

    def extract_mcp_text_result(self, result: Any) -> str:
        """
        从 MCP CallToolResult 中提取文本内容。

        参数：
            result: MCP 工具调用结果。

        返回：
            拼接后的文本。
        """

        # 准备一个列表保存所有文本片段。
        texts: list[str] = []

        # result.content 是 MCP 内容块列表，可能包含文本、图片、资源等不同类型。
        for block in result.content:
            # 如果内容块是 TextContent，就取 block.text。
            if isinstance(block, types.TextContent):
                texts.append(block.text)
            # 如果是其他类型，这里简单转成字符串，真实项目中可按类型分别处理。
            else:
                texts.append(str(block))

        # 用换行拼接多个文本片段。
        return "\n".join(texts)

    async def run(self, user_question: str) -> str:
        """
        运行完整 MCP Agent。

        参数：
            user_question: 用户问题。

        返回：
            最终中文回答。
        """

        # 构造 stdio server 参数，准备启动 MCP server 子进程。
        server_params = self.build_server_params()

        # stdio_client 会启动 MCP server 子进程，并返回读写流。
        async with stdio_client(server_params) as (read_stream, write_stream):
            # ClientSession 封装了 MCP 初始化、工具发现、工具调用等能力。
            async with ClientSession(read_stream, write_stream) as session:
                # 初始化 MCP 连接；没有这一步，后续 list_tools/call_tool 不能正常工作。
                await session.initialize()

                # 打印 MCP server 暴露的工具和 prompt，方便你观察 MCP 的发现过程。
                await self.print_mcp_capabilities(session)

                # 从 MCP server 获取 prompt 模板。
                system_prompt = await self.get_prompt_from_mcp(session, user_question)

                # 把 MCP tools 转成 OpenAI tools，供模型选择调用。
                openai_tools = await self.convert_mcp_tools_to_openai_tools(session)

                # 构造初始对话历史。
                messages: list[dict[str, Any]] = [
                    # system 消息放 MCP prompt 返回的提示词。
                    {"role": "system", "content": system_prompt},
                    # user 消息放用户原始问题。
                    {"role": "user", "content": user_question},
                ]

                # 最多循环 5 轮，防止模型无限调用工具。
                for _ in range(5):
                    # 把 messages 和 MCP 转换出的 tools 发给模型。
                    assistant_message = self.call_llm(messages, openai_tools)

                    # 将 assistant message 加入上下文，保留模型发起的 tool_calls。
                    messages.append(assistant_message.model_dump(exclude_none=True))

                    # 如果没有 tool_calls，说明模型给出了最终回答。
                    if not assistant_message.tool_calls:
                        return assistant_message.content or ""

                    # 遍历模型请求的每个工具调用。
                    for tool_call in assistant_message.tool_calls:
                        # 读取模型想调用的工具名。
                        tool_name = tool_call.function.name

                        # 解析模型生成的 JSON 参数字符串。
                        arguments = json.loads(tool_call.function.arguments)

                        # 打印调用信息，方便你观察“模型决定调用什么工具”。
                        print("\n===== LLM 请求调用 MCP Tool =====")
                        print(f"tool_name = {tool_name}")
                        print(f"arguments = {arguments}")

                        # 通过 MCP 协议调用 server 上的工具。
                        tool_result_json = await self.call_mcp_tool(session, tool_name, arguments)

                        # 打印 MCP 工具返回值，方便调试。
                        print("===== MCP Tool 返回结果 =====")
                        print(tool_result_json)

                        # 把 MCP 工具结果作为 role=tool 消息发回模型。
                        messages.append(
                            {
                                "role": "tool",
                                "tool_call_id": tool_call.id,
                                "content": tool_result_json,
                            }
                        )

                # 如果超过最大轮数，返回保护性提示。
                return "工具调用轮数过多，Agent 已停止。"


# 只有直接运行本文件时，才进入这个分支。
if __name__ == "__main__":
    # 如果命令行参数中包含 --server，说明当前进程要作为 MCP server 运行。
    if "--server" in sys.argv:
        # 使用 stdio transport 启动 MCP server；client 会通过标准输入输出与它通信。
        mcp.run(transport="stdio")
    else:
        # 否则当前进程作为 MCP Agent client 运行。
        agent = MCPAgent()

        # 准备 demo 问题；可以改成 input("请输入问题：")。
        question = "请查询北京天气，并把摄氏温度换算成华氏温度，最后给出穿衣建议。"

        # asyncio.run 用于运行异步 main 流程。
        answer = asyncio.run(agent.run(question))

        # 打印最终回答。
        print("\n===== MCP Agent 最终回答 =====")
        print(answer)

===== MCP Server 可用 Prompts =====
- weather_agent_prompt: 
    MCP Prompt：返回一个可复用的天气 Agent 提示模板。

    参数：
        user_question: 用户原始问题。

    返回：
        一段提示词文本。

    说明：
        @mcp.prompt() 和 @mcp.tool() 类似，但它暴露的不是可执行函数，
        而是“提示模板”。MCP client 可以先发现 prompt，再按需获取 prompt 内容。
    

===== MCP Server 可用 Tools =====
- name: get_weather
  description: 
    MCP 工具：查询指定城市的演示天气信息。

    参数：
        city: 城市名，例如“北京”“上海”“深圳”“杭州”。

    返回：
        dict 格式的天气信息。

    说明：
        @mcp.tool() 会把这个普通 Python 函数注册为 MCP 工具。
        MCP server 会根据函数签名 city: str 和 docstring 自动生成工具 schema。
    
  inputSchema: {"properties": {"city": {"title": "City", "type": "string"}}, "required": ["city"], "title": "get_weatherArguments", "type": "object"}
- name: celsius_to_fahrenheit
  description: 
    MCP 工具：把摄氏温度转换为华氏温度。

    参数：
        celsius: 摄氏温度。

    返回：
        dict，包含原摄氏温度和华氏温度。
    
  inputSchema: {"properties": {"celsius": {"title": "Celsius", "type": "number"}}, "required": ["celsius"], "title": "celsius_to_fahrenheitArguments", "type": "object"}
[07/06/26 14:36:57] INFO     HTTP Request: POST                              _client.py:1025
                             https://your-llm-provider.com/v1/chat/completions                       
                             "HTTP/1.1 200 OK"                                              

===== LLM 请求调用 MCP Tool =====
tool_name = get_weather
arguments = {'city': '北京'}
===== MCP Tool 返回结果 =====
{"city": "北京", "condition": "晴", "temperature_c": 30.0, "humidity": "35%", "source": "demo_mcp_weather_server"}
[07/06/26 14:36:58] INFO     HTTP Request: POST                              _client.py:1025
                             https://your-llm-provider.com/v1/chat/completions                       
                             "HTTP/1.1 200 OK"                                              

===== LLM 请求调用 MCP Tool =====
tool_name = celsius_to_fahrenheit
arguments = {'celsius': 30}
===== MCP Tool 返回结果 =====
{"celsius": 30.0, "fahrenheit": 86.0}
[07/06/26 14:37:01] INFO     HTTP Request: POST                              _client.py:1025
                             https://your-llm-provider.com/v1/chat/completions                       
                             "HTTP/1.1 200 OK"                                              

===== MCP Agent 最终回答 =====
北京当前天气：晴，气温 30°C，湿度 35%。

换算成华氏温度：86°F。

穿衣建议：天气较热，建议穿轻薄、透气的夏装，如短袖、短裤/薄长裤；注意防晒，外出可带太阳镜和帽子，并及时补水。

用户问题
↓
LLM 判断需要天气工具
↓
调用 MCP tool: get_weather
↓
拿到北京天气
↓
LLM 判断需要温度换算工具
↓
调用 MCP tool: celsius_to_fahrenheit
↓
拿到华氏温度
↓
LLM 生成最终中文回答

weather_mcp_server.py
    ├── get_weather()
    └── celsius_to_fahrenheit()

openai_agent.py
    └── 不写 get_weather，只连接 MCP server

langchain_agent.py
    └── 不写 get_weather，只连接 MCP server

langgraph_agent.py
    └── 不写 get_weather，只连接 MCP server

claude_desktop_config.json
    └── 也可以连接同一个 MCP server

function_call_agent.py
    ├── get_weather()
    └── celsius_to_fahrenheit()

mcp_agent.py
    ├── get_weather()
    └── celsius_to_fahrenheit()

​

​

​

​

​

# MCP 提供对 GitHub 的标准化访问
github_mcp = MCPTool(server_command=["npx", "-y", "@modelcontextprotocol/server-github"])

# MCP 暴露的工具（简化示例）：
# - list_pull_requests(repo, state)
# - get_pull_request_details(pr_number)
# - list_pr_comments(pr_number)
# - create_pr_comment(pr_number, body)
# - get_file_content(repo, path, ref)
# - list_pr_files(pr_number)

---
name: code-review-workflow
description: 执行标准的代码审查流程，包括检查代码风格、安全问题、测试覆盖率等
---

# 代码审查工作流

## 审查清单

当执行代码审查时，按以下步骤进行：

1. **获取 PR 信息**：调用 `get_pull_request_details` 了解变更背景
2. **分析变更文件**：调用 `list_pr_files` 获取文件列表
3. **逐文件审查**：
   - 对于 `.py` 文件：检查是否符合 PEP 8，是否有明显的性能问题
   - 对于 `.js/.ts` 文件：检查是否有未处理的 Promise，是否使用了废弃的 API
   - 对于测试文件：验证是否覆盖了新增的代码路径
4. **安全检查**：
   - 是否硬编码了敏感信息（密钥、密码）
   - 是否有 SQL 注入或 XSS 风险
5. **提供反馈**：
   - 严重问题：使用 `create_pr_comment` 直接评论
   - 建议改进：在总结中提出

## 公司特定规范

- 所有数据库查询必须使用参数化查询
- API 端点必须有权限验证装饰器
- 新功能必须附带单元测试（覆盖率 > 80%）

## 示例评论模板

**严重问题**：

⚠️ 安全风险：第 45 行直接拼接 SQL 字符串，存在注入风险。
建议改用参数化查询：`cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))`

---
# === 必需字段 ===
name: skill-name
  # 技能的唯一标识符，使用 kebab-case 命名

description: >
  简洁但精确的描述，说明：
  1. 这个技能做什么
  2. 什么时候应该使用它
  3. 它的核心价值是什么
  # 注意：description 是智能体选择技能的唯一依据，必须写清楚！

# === 可选字段 ===
version: 1.0.0
  # 语义化版本号

allowed_tools: [tool1, tool2]
  # 此技能可以调用的工具列表（白名单）

required_context: [context_item1]
  # 此技能需要的上下文信息

license: MIT
  # 许可协议

author: Your Name <email@example.com>
  # 作者信息

tags: [database, analysis, sql]
  # 便于分类和搜索的标签
---

# 技能标题

## 概述
（对技能的详细介绍，包括使用场景、技术背景等）

## 前置条件
（使用此技能需要的环境配置、依赖项等）

## 工作流程
（详细的步骤说明，告诉智能体如何执行任务）

## 最佳实践
（经验总结、注意事项、常见陷阱等）

## 示例
（具体的使用案例，帮助智能体理解）

## 故障排查
（常见问题和解决方案）

用户输入任务
  ↓
SkillLoader 读取所有 SKILL.md 的 frontmatter
  ↓
LLM 根据 description 选择 skill
  ↓
Harness 加载被选中 skill 的 SKILL.md 正文
  ↓
Harness 根据 allowed_tools 暴露工具
  ↓
LLM 判断是否需要调用工具
  ↓
ToolExecutor 检查工具是否在白名单中
  ↓
执行工具
  ↓
工具结果返回给 LLM
  ↓
LLM 继续推理或输出最终答案

skills_agent_demo/
├── .env
├── requirements.txt
├── main.py
├── agent/
│   ├── __init__.py
│   ├── llm.py
│   ├── skill_loader.py
│   ├── tools.py
│   └── harness.py
└── skills/
    ├── web-research/
    │   ├── SKILL.md
    │   ├── advanced.md
    │   └── data/
    │       └── source_policy.json
    └── structured-writing/
        ├── SKILL.md
        └── advanced.md

┌────────────────────┐
│ 用户输入任务        │
└─────────┬──────────┘
          ↓
┌────────────────────┐
│ SkillLoader         │
│ 读取所有 SKILL.md   │
│ 的 frontmatter      │
└─────────┬──────────┘
          ↓
┌────────────────────┐
│ Skill Selector LLM  │
│ 只看 name +         │
│ description         │
└─────────┬──────────┘
          ↓
┌────────────────────┐
│ 选中某个 Skill      │
│ 例如 web-research   │
└─────────┬──────────┘
          ↓
┌────────────────────┐
│ Harness 加载该      │
│ Skill 的正文 body   │
└─────────┬──────────┘
          ↓
┌────────────────────┐
│ Harness 读取        │
│ allowed_tools       │
│ 并暴露工具 schema   │
└─────────┬──────────┘
          ↓
┌────────────────────┐
│ LLM 根据 Skill      │
│ 决定下一步           │
└─────────┬──────────┘
          ↓
┌────────────────────────────┐
│ 可能 1：直接回答            │
│ 可能 2：调用 tavily_search  │
│ 可能 3：读取 advanced.md    │
│ 可能 4：查询 data 文件      │
└─────────┬──────────────────┘
          ↓
┌────────────────────┐
│ ToolExecutor 执行   │
│ 工具并返回结果      │
└─────────┬──────────┘
          ↓
┌────────────────────┐
│ 工具结果重新放回    │
│ messages            │
└─────────┬──────────┘
          ↓
┌────────────────────┐
│ LLM 基于观察结果    │
│ 继续推理或最终回答  │
└────────────────────┘

from __future__ import annotations

import json
import re
from typing import Any

from agent.llm import ChatLLM
from agent.skill_loader import Skill, SkillLoader
from agent.tools import ToolExecutor


class SkillAgentHarness:
    """
    Skills Agent 的工程外壳。

    你可以把它理解成：

        LLM + Harness = Agent

    LLM 负责：
        1. 选择 skill。
        2. 根据上下文生成回答。
        3. 决定是否调用工具。

    Harness 负责：
        1. 加载 skills。
        2. 控制 skill 选择流程。
        3. 把选中的 SKILL.md 正文放进上下文。
        4. 根据 allowed_tools 暴露工具。
        5. 执行工具。
        6. 把工具结果塞回消息列表。
        7. 控制最大工具调用轮数。
    """

    def __init__(
        self,
        llm: ChatLLM,
        skill_loader: SkillLoader,
        max_tool_steps: int = 6,
        verbose: bool = True,
    ):
        """
        初始化 Agent Harness。

        参数：
            llm:
                ChatLLM 实例。

            skill_loader:
                SkillLoader 实例。

            max_tool_steps:
                最大工具调用轮数。
                防止模型无限循环调用工具。

            verbose:
                是否打印中间过程。
        """

        self.llm = llm
        self.skill_loader = skill_loader
        self.tool_executor = ToolExecutor(skill_loader)
        self.max_tool_steps = max_tool_steps
        self.verbose = verbose

    def run(self, user_task: str) -> str:
        """
        运行 Agent。

        完整流程：
        1. 根据用户任务选择 skill。
        2. 打印被选中的 skill。
        3. 使用被选中的 skill 执行任务。
        4. 返回最终答案。
        """

        selected_skill, reason = self._select_skill(user_task)

        if self.verbose:
            print("\n========== Skill Selection ==========")
            print(f"Selected skill: {selected_skill.name}")
            print(f"Reason: {reason}")
            print("Allowed tools:", selected_skill.allowed_tools)
            print("Available resources:", self.skill_loader.list_resources(selected_skill.name))

        return self._run_with_skill(user_task, selected_skill)

    def _select_skill(self, user_task: str) -> tuple[Skill, str]:
        """
        选择最合适的 skill。

        关键点：
            选择阶段只给 LLM 看：
            1. skill.name
            2. skill.description

            不给它看：
            1. SKILL.md 正文
            2. advanced.md
            3. data 文件

        这样可以模拟真实 Skills 系统中的第一层渐进式披露。
        """

        skill_candidates = self.skill_loader.list_for_selection()

        selector_messages = [
            {
                "role": "system",
                "content": (
                    "你是一个 skill selector。\n"
                    "你必须只根据每个 skill 的 description 判断应该选择哪个 skill。\n"
                    "name 只作为返回用的唯一标识，不作为语义判断依据。\n"
                    "请返回严格 JSON，不要输出多余文字。\n"
                    "JSON 格式：{\"skill_name\": \"...\", \"reason\": \"...\"}"
                ),
            },
            {
                "role": "user",
                "content": json.dumps(
                    {
                        "user_task": user_task,
                        "skills": skill_candidates,
                    },
                    ensure_ascii=False,
                    indent=2,
                ),
            },
        ]

        # 这里不传 tools，因为 skill 选择只是一个分类任务
        response = self.llm.chat(selector_messages, temperature=0.0)

        content = response.choices[0].message.content or ""

        # 尝试解析模型返回的 JSON
        data = self._parse_json_object(content)

        skill_name = data.get("skill_name")
        reason = data.get("reason", "")

        # 如果模型返回了合法 skill name，就使用它
        if skill_name in self.skill_loader.skills:
            return self.skill_loader.get(skill_name), reason

        # 如果模型输出格式错误，使用一个简单 fallback，防止 demo 中断
        fallback = self._fallback_select_skill(user_task)
        return fallback, "LLM 返回的 skill_name 无效，使用关键词 fallback。"

    def _fallback_select_skill(self, user_task: str) -> Skill:
        """
        简单的关键词 fallback。

        真实工程里可以替换成：
        1. embedding 检索
        2. BM25
        3. 规则系统
        4. 小模型分类器

        这里保留 fallback 是为了让 demo 更稳。
        """

        text = user_task.lower()

        # 这些关键词更像联网研究任务
        if any(
            keyword in text
            for keyword in [
                "搜索",
                "查找",
                "最新",
                "新闻",
                "联网",
                "资料",
                "来源",
                "调研",
                "research",
                "tavily",
                "官网",
            ]
        ):
            return self.skill_loader.get("web-research")

        # 默认使用写作 skill
        return self.skill_loader.get("structured-writing")

    def _run_with_skill(self, user_task: str, skill: Skill) -> str:
        """
        使用指定 skill 处理用户任务。

        这个函数是 Agent 的核心 loop。

        流程：
        1. 读取当前 skill 可用资源列表。
        2. 构造 system prompt。
        3. 根据 allowed_tools 暴露工具 schema。
        4. 调用 LLM。
        5. 如果 LLM 返回 tool_calls，就执行工具。
        6. 把工具结果追加到 messages。
        7. 再次调用 LLM。
        8. 如果 LLM 不再调用工具，就返回最终答案。
        """

        # 只列出资源路径，不读取资源内容
        resources = self.skill_loader.list_resources(skill.name)

        # 选中 skill 后，才把 SKILL.md 主体放进上下文
        system_prompt = f"""
你是一个支持 Skills 的 Agent。

当前已经触发的 Skill：

# Skill Name
{skill.name}

# Skill Description
{skill.description}

# Skill Body
{skill.body}

# 当前 Skill 允许使用的工具白名单
{json.dumps(skill.allowed_tools, ensure_ascii=False)}

# 当前 Skill 可按需读取的资源
{json.dumps(resources, ensure_ascii=False, indent=2)}

重要规则：
1. 你必须遵循当前 Skill 的工作流程。
2. 你只能调用 allowed_tools 中列出的工具。
3. 你不能假装已经读取 advanced.md 或 data 文件。
4. 如果需要高级策略、边缘情况或数据文件，必须调用 load_skill_resource 或 query_skill_data。
5. 如果工具返回错误，要先根据错误调整下一步，不要编造工具结果。
6. 最终回答使用中文，结构清晰。
""".strip()

        messages: list[dict[str, Any]] = [
            {
                "role": "system",
                "content": system_prompt,
            },
            {
                "role": "user",
                "content": user_task,
            },
        ]

        # 根据当前 skill 的 allowed_tools 获取工具 schema
        # 注意：这里不会暴露所有工具，只暴露白名单内工具
        tool_schemas = self.tool_executor.schemas_for(skill.allowed_tools)

        # 最多循环 max_tool_steps 轮，防止无限调用工具
        for step in range(1, self.max_tool_steps + 1):
            # 调用 LLM
            response = self.llm.chat(
                messages=messages,
                tools=tool_schemas,
                tool_choice="auto" if tool_schemas else None,
                temperature=0.2,
            )

            assistant_message = response.choices[0].message

            # 如果模型返回 tool_calls，说明它想调用工具
            tool_calls = assistant_message.tool_calls or []

            # 如果没有 tool call，说明模型已经准备好最终回答
            if not tool_calls:
                return assistant_message.content or ""

            # 把 assistant 的 tool call 消息追加到 messages
            messages.append(assistant_message.model_dump(exclude_none=True))

            # 逐个执行工具调用
            for tool_call in tool_calls:
                tool_name = tool_call.function.name

                try:
                    arguments = json.loads(tool_call.function.arguments or "{}")
                except json.JSONDecodeError:
                    arguments = {}

                if self.verbose:
                    print(f"\n[Tool Step {step}] {tool_name}")
                    print(json.dumps(arguments, ensure_ascii=False, indent=2))

                # 通过 ToolExecutor 执行工具
                # ToolExecutor 会再次检查 allowed_tools 白名单
                tool_result = self.tool_executor.execute(
                    skill=skill,
                    tool_name=tool_name,
                    arguments=arguments,
                )

                if self.verbose:
                    preview = tool_result[:1000]
                    print("[Tool Result Preview]")
                    print(preview + ("..." if len(tool_result) > 1000 else ""))

                # 把工具结果以 role=tool 的形式加入 messages
                # 下一轮 LLM 会基于这个观察结果继续推理
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": tool_result,
                    }
                )

        # 如果达到最大工具轮数还没结束，强制让模型总结已有信息
        messages.append(
            {
                "role": "system",
                "content": (
                    "工具调用次数已达到上限。"
                    "请基于已有信息给出最终答案。"
                    "如果信息不足，请明确说明不足之处。"
                ),
            }
        )

        final_response = self.llm.chat(messages=messages, temperature=0.2)

        return final_response.choices[0].message.content or ""

    def _parse_json_object(self, text: str) -> dict[str, Any]:
        """
        从模型输出中解析 JSON 对象。

        理想情况下，模型会直接输出：
            {"skill_name": "...", "reason": "..."}

        但模型偶尔可能加上一些解释文字。
        所以这里做两步解析：
        1. 先尝试直接 json.loads。
        2. 如果失败，再用正则提取第一个 {...}。
        """

        text = text.strip()

        try:
            return json.loads(text)
        except json.JSONDecodeError:
            pass

        match = re.search(r"\{.*\}", text, flags=re.DOTALL)

        if match:
            try:
                return json.loads(match.group(0))
            except json.JSONDecodeError:
                pass

        return {}

import os
from typing import Any

from openai import OpenAI


class ChatLLM:
    """
    OpenAI Chat Completions 的简单封装。

    这个类只负责一件事：
    调用 OpenAI 的 chat.completions.create 接口。

    注意：
    这里故意没有使用 LangChain。
    这样可以更清楚地看到：
    1. LLM 只是负责生成回复或 tool call。
    2. Agent 的 skill 选择、tool 执行、循环控制，都是 Harness 在做。
    """

    def __init__(self, model: str | None = None):
        """
        初始化 LLM 客户端。

        参数：
            model:
                可选模型名。
                如果不传，则读取环境变量 LLM_MODEL_ID。
                如果环境变量也没有，则默认使用 gpt-4.1-mini。
        """

        # 从环境变量中读取 LLM_API_KEY
        self.client = OpenAI(api_key=os.getenv("LLM_API_KEY"))

        # 优先使用传入模型，其次使用环境变量，最后使用默认模型
        self.model = model or os.getenv("LLM_MODEL_ID", "gpt-4.1-mini")

    def chat(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]] | None = None,
        tool_choice: str | dict[str, Any] | None = None,
        temperature: float = 0.2,
    ):
        """
        调用 OpenAI Chat Completions。

        参数：
            messages:
                对话消息列表。
                格式类似：
                [
                    {"role": "system", "content": "..."},
                    {"role": "user", "content": "..."}
                ]

            tools:
                可选工具 schema 列表。
                如果传入 tools，模型就可以返回 tool_calls。

            tool_choice:
                工具选择策略。
                常用值：
                - "auto"：模型自己决定是否调用工具
                - "none"：禁止调用工具
                - 指定某个工具：强制调用

            temperature:
                采样温度。
                越低越稳定，越高越发散。

        返回：
            OpenAI SDK 返回的 ChatCompletion 对象。
        """

        kwargs: dict[str, Any] = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature,
        }

        # 只有在存在工具时才传 tools 参数
        if tools:
            kwargs["tools"] = tools
            kwargs["tool_choice"] = tool_choice or "auto"

        return self.client.chat.completions.create(**kwargs)

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any
import json

import yaml


@dataclass
class Skill:
    """
    Skill 数据结构。

    一个 Skill 对应一个 skills/<skill-name>/SKILL.md 文件。

    字段说明：
        name:
            skill 的唯一标识符。

        description:
            skill 的简洁描述。
            在这个 demo 中，description 是 LLM 选择 skill 的唯一语义依据。

        version:
            skill 版本号，可选。

        allowed_tools:
            当前 skill 允许调用的工具白名单。

        required_context:
            当前 skill 需要的上下文说明。
            这个 demo 中暂时只展示，不做强校验。

        license:
            许可证信息。

        author:
            作者信息。

        tags:
            标签，用于分类。
            注意：在本 demo 的 skill 选择阶段，不使用 tags 作为语义判断依据。

        directory:
            skill 所在目录。

        body:
            SKILL.md 中 frontmatter 后面的正文。
            只有 skill 被选中后，才会放进 Agent 上下文。
    """

    name: str
    description: str
    version: str | None
    allowed_tools: list[str]
    required_context: list[str]
    license: str | None
    author: str | None
    tags: list[str]
    directory: Path
    body: str

    def metadata_for_selection(self) -> dict[str, str]:
        """
        返回给 skill selector 的最小信息。

        注意：
        这里故意只返回 name 和 description。

        原因：
        1. name 用于标识最终选择哪个 skill。
        2. description 是选择 skill 的唯一语义依据。
        3. 不把 SKILL.md 正文、advanced.md、data 文件提前塞给模型。
        """

        return {
            "name": self.name,
            "description": self.description,
        }


class SkillLoader:
    """
    Skill 加载器。

    主要职责：
    1. 扫描 skills 目录。
    2. 读取每个 SKILL.md。
    3. 解析 YAML frontmatter。
    4. 保存 skill 元信息和正文。
    5. 提供资源读取能力，用于渐进式披露。
    """

    def __init__(self, skills_root: Path):
        """
        初始化 SkillLoader。

        参数：
            skills_root:
                skills 根目录，例如 skills_agent_demo/skills。
        """

        self.skills_root = skills_root

        # key 是 skill.name，value 是 Skill 对象
        self.skills: dict[str, Skill] = {}

    def discover(self) -> dict[str, Skill]:
        """
        发现所有 skills。

        它会扫描：
            skills/*/SKILL.md

        返回：
            skill name 到 Skill 对象的映射。

        异常：
            如果 skills 目录不存在，会报错。
            如果没有发现任何 SKILL.md，会报错。
            如果 skill name 重复，会报错。
        """

        if not self.skills_root.exists():
            raise FileNotFoundError(f"skills 目录不存在：{self.skills_root}")

        # 遍历所有子目录里的 SKILL.md
        for skill_md in self.skills_root.glob("*/SKILL.md"):
            skill = self._load_skill_metadata_and_body(skill_md)

            if skill.name in self.skills:
                raise ValueError(f"重复的 skill name：{skill.name}")

            self.skills[skill.name] = skill

        if not self.skills:
            raise RuntimeError(f"没有发现任何 SKILL.md：{self.skills_root}")

        return self.skills

    def list_for_selection(self) -> list[dict[str, str]]:
        """
        返回用于选择 skill 的候选列表。

        只包含：
        1. name
        2. description

        不包含：
        1. SKILL.md 正文
        2. advanced.md
        3. data 文件
        4. references 文件
        """

        return [skill.metadata_for_selection() for skill in self.skills.values()]

    def get(self, skill_name: str) -> Skill:
        """
        根据 skill name 获取 Skill 对象。

        参数：
            skill_name:
                skill 的唯一名称。

        返回：
            Skill 对象。
        """

        if skill_name not in self.skills:
            raise KeyError(f"未知 skill：{skill_name}")

        return self.skills[skill_name]

    def list_resources(self, skill_name: str) -> list[str]:
        """
        列出某个 skill 目录下的附加资源。

        这也是渐进式披露的一部分。

        它只告诉模型：
            当前 skill 有哪些资源可用。

        但不会直接读取资源内容。

        例如可能返回：
            [
                "advanced.md",
                "data/source_policy.json"
            ]
        """

        skill = self.get(skill_name)
        resources: list[str] = []

        # 遍历 skill 目录下所有文件
        for path in skill.directory.rglob("*"):
            if path.is_file() and path.name != "SKILL.md":
                resources.append(path.relative_to(skill.directory).as_posix())

        return sorted(resources)

    def read_resource(self, skill_name: str, relative_path: str) -> str:
        """
        读取某个 skill 目录下的资源文件。

        参数：
            skill_name:
                当前 skill 名称。

            relative_path:
                相对于该 skill 目录的资源路径。
                例如：
                - advanced.md
                - data/source_policy.json

        安全限制：
            只能读取当前 skill 目录内的文件。
            禁止通过 ../ 逃逸到其他目录。
        """

        skill = self.get(skill_name)

        # 计算目标文件绝对路径
        target = (skill.directory / relative_path).resolve()

        # 当前 skill 目录绝对路径
        skill_dir = skill.directory.resolve()

        # 防止路径逃逸
        if not str(target).startswith(str(skill_dir)):
            raise ValueError("非法资源路径：禁止访问 skill 目录之外的文件")

        if not target.exists() or not target.is_file():
            raise FileNotFoundError(f"资源不存在：{relative_path}")

        return target.read_text(encoding="utf-8")

    def query_json_resource(
        self,
        skill_name: str,
        relative_path: str,
        query: str = "",
        limit: int = 5,
    ) -> str:
        """
        查询当前 skill 目录下的 JSON 数据文件。

        这个函数用于演示：
            大型数据文件不直接塞进 prompt，
            而是通过工具按需查询。

        参数：
            skill_name:
                当前 skill 名称。

            relative_path:
                JSON 文件相对路径。

            query:
                简单关键词过滤。
                为空时返回前 limit 条。

            limit:
                最多返回多少条。

        返回：
            JSON 字符串。
        """

        # 先读取 JSON 文件内容
        raw = self.read_resource(skill_name, relative_path)

        # 解析 JSON
        data = json.loads(raw)

        query_lower = query.lower().strip()

        # 把 dict/list 统一转换成列表，方便过滤
        if isinstance(data, dict):
            items = [{"key": k, "value": v} for k, v in data.items()]
        elif isinstance(data, list):
            items = data
        else:
            items = [{"value": data}]

        # 如果提供了 query，就做简单字符串匹配
        if query_lower:
            filtered = []

            for item in items:
                text = json.dumps(item, ensure_ascii=False).lower()

                if query_lower in text:
                    filtered.append(item)

            items = filtered

        # 返回前 limit 条
        return json.dumps(items[:limit], ensure_ascii=False, indent=2)

    def _load_skill_metadata_and_body(self, skill_md: Path) -> Skill:
        """
        读取并解析单个 SKILL.md 文件。

        SKILL.md 格式：

        ---
        name: xxx
        description: >
          xxx
        allowed_tools: [...]
        ---

        # 正文

        返回：
            Skill 对象。
        """

        text = skill_md.read_text(encoding="utf-8")

        if not text.startswith("---"):
            raise ValueError(f"{skill_md} 缺少 YAML frontmatter")

        try:
            # 分割：
            # 第一个 --- 前面为空
            # 第二部分是 YAML frontmatter
            # 第三部分是 Markdown 正文
            _, frontmatter_text, body = text.split("---", 2)
        except ValueError as exc:
            raise ValueError(f"{skill_md} frontmatter 格式错误") from exc

        # 解析 YAML
        meta: dict[str, Any] = yaml.safe_load(frontmatter_text) or {}

        # 检查必需字段
        required_fields = ["name", "description"]

        for field in required_fields:
            if field not in meta:
                raise ValueError(f"{skill_md} 缺少必需字段：{field}")

        # 构造 Skill 对象
        return Skill(
            name=str(meta["name"]),
            description=str(meta["description"]).strip(),
            version=meta.get("version"),
            allowed_tools=list(meta.get("allowed_tools", [])),
            required_context=list(meta.get("required_context", [])),
            license=meta.get("license"),
            author=meta.get("author"),
            tags=list(meta.get("tags", [])),
            directory=skill_md.parent,
            body=body.strip(),
        )

from __future__ import annotations

import json
import os
from typing import Any, Callable

from tavily import TavilyClient

from agent.skill_loader import Skill, SkillLoader


def tavily_search(query: str, max_results: int = 5, topic: str = "general") -> str:
    """
    使用 Tavily 搜索网页。

    这个工具用于 web-research skill。

    参数：
        query:
            搜索查询词。

        max_results:
            最多返回多少条结果。
            为了避免上下文太长，这里限制在 1 到 10 之间。

        topic:
            Tavily 的搜索类型。
            常见值：
            - general
            - news
            - finance

    返回：
        JSON 字符串，包含搜索结果列表。
    """

    api_key = os.getenv("TAVILY_API_KEY")

    if not api_key:
        return json.dumps(
            {
                "error": "缺少 TAVILY_API_KEY，请在 .env 文件中配置。",
            },
            ensure_ascii=False,
        )

    # 限制 max_results 范围，避免一次返回太多内容
    max_results = max(1, min(int(max_results), 10))

    # 初始化 Tavily 客户端
    client = TavilyClient(api_key=api_key)

    # 调用 Tavily 搜索
    response = client.search(
        query=query,
        max_results=max_results,
        topic=topic,
        include_answer=False,
    )

    # 压缩搜索结果，只保留 Agent 最常用的字段
    compact_results = []

    for item in response.get("results", []):
        compact_results.append(
            {
                "title": item.get("title"),
                "url": item.get("url"),
                "content": item.get("content"),
                "score": item.get("score"),
            }
        )

    return json.dumps(
        {
            "query": query,
            "results": compact_results,
        },
        ensure_ascii=False,
        indent=2,
    )


class ToolExecutor:
    """
    工具执行器。

    主要职责：
    1. 注册所有可用工具。
    2. 根据 skill.allowed_tools 暴露工具 schema。
    3. 执行工具前检查当前 skill 是否允许调用该工具。
    4. 执行工具并返回结果。
    """

    def __init__(self, skill_loader: SkillLoader):
        """
        初始化 ToolExecutor。

        参数：
            skill_loader:
                SkillLoader 实例。
                用于读取 skill 的 advanced.md、data 文件等资源。
        """

        self.skill_loader = skill_loader

        # 注册所有工具
        self.registry: dict[str, dict[str, Any]] = self._build_registry()

    def schemas_for(self, allowed_tool_names: list[str]) -> list[dict[str, Any]]:
        """
        根据 allowed_tools 返回工具 schema。

        参数：
            allowed_tool_names:
                当前 skill 允许调用的工具名称列表。

        返回：
            OpenAI Chat Completions tools 参数所需的 schema 列表。
        """

        schemas = []

        for name in allowed_tool_names:
            if name in self.registry:
                schemas.append(self.registry[name]["schema"])

        return schemas

    def execute(self, skill: Skill, tool_name: str, arguments: dict[str, Any]) -> str:
        """
        执行某个工具。

        参数：
            skill:
                当前被激活的 Skill。

            tool_name:
                模型想调用的工具名称。

            arguments:
                模型生成的工具参数。

        关键安全点：
            即使模型请求调用某个工具，
            Harness 也不会直接执行，
            而是先检查 tool_name 是否在当前 skill.allowed_tools 中。
        """

        # 白名单检查：防止模型越权调用工具
        if tool_name not in skill.allowed_tools:
            return json.dumps(
                {
                    "error": f"工具 {tool_name} 不在当前 skill 的 allowed_tools 白名单中。",
                    "allowed_tools": skill.allowed_tools,
                },
                ensure_ascii=False,
            )

        # 工具是否存在
        if tool_name not in self.registry:
            return json.dumps(
                {
                    "error": f"未知工具：{tool_name}",
                },
                ensure_ascii=False,
            )

        try:
            # load_skill_resource 是特殊工具：
            # 它不直接对应一个普通 Python 函数，
            # 而是调用 SkillLoader 读取当前 skill 目录下的资源。
            if tool_name == "load_skill_resource":
                relative_path = arguments["relative_path"]

                content = self.skill_loader.read_resource(
                    skill_name=skill.name,
                    relative_path=relative_path,
                )

                return json.dumps(
                    {
                        "skill": skill.name,
                        "resource": relative_path,
                        "content": content,
                    },
                    ensure_ascii=False,
                    indent=2,
                )

            # query_skill_data 也是特殊工具：
            # 它用于查询当前 skill 目录下的 JSON 数据文件。
            if tool_name == "query_skill_data":
                relative_path = arguments["relative_path"]
                query = arguments.get("query", "")
                limit = int(arguments.get("limit", 5))

                content = self.skill_loader.query_json_resource(
                    skill_name=skill.name,
                    relative_path=relative_path,
                    query=query,
                    limit=limit,
                )

                return json.dumps(
                    {
                        "skill": skill.name,
                        "resource": relative_path,
                        "query": query,
                        "matches": json.loads(content),
                    },
                    ensure_ascii=False,
                    indent=2,
                )

            # 普通工具直接从 registry 中取函数执行
            func: Callable[..., str] = self.registry[tool_name]["func"]

            return func(**arguments)

        except Exception as exc:
            # 工具执行失败时，把错误返回给模型
            # 让模型有机会调整下一步
            return json.dumps(
                {
                    "error": str(exc),
                    "tool": tool_name,
                },
                ensure_ascii=False,
            )

    def _build_registry(self) -> dict[str, dict[str, Any]]:
        """
        注册所有工具。

        每个工具包含两部分：
        1. func:
            Python 函数。
        2. schema:
            给 OpenAI tool calling 使用的 JSON schema。

        注意：
            这里只是“注册工具”。
            某个 skill 能不能用这些工具，
            取决于该 skill 的 allowed_tools。
        """

        return {
            "tavily_search": {
                "func": tavily_search,
                "schema": {
                    "type": "function",
                    "function": {
                        "name": "tavily_search",
                        "description": "使用 Tavily 搜索网页，适合查找最新信息、多来源资料、事实核查。",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "query": {
                                    "type": "string",
                                    "description": "搜索查询词，应该简洁明确。",
                                },
                                "max_results": {
                                    "type": "integer",
                                    "description": "返回结果数量，建议 3 到 5。",
                                    "default": 5,
                                },
                                "topic": {
                                    "type": "string",
                                    "enum": ["general", "news", "finance"],
                                    "description": "搜索类别。",
                                    "default": "general",
                                },
                            },
                            "required": ["query"],
                        },
                    },
                },
            },
            "load_skill_resource": {
                "func": None,
                "schema": {
                    "type": "function",
                    "function": {
                        "name": "load_skill_resource",
                        "description": "读取当前 skill 目录下的附加文档，例如 advanced.md。只能读取当前 skill 内部资源。",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "relative_path": {
                                    "type": "string",
                                    "description": "相对于当前 skill 目录的资源路径，例如 advanced.md。",
                                }
                            },
                            "required": ["relative_path"],
                        },
                    },
                },
            },
            "query_skill_data": {
                "func": None,
                "schema": {
                    "type": "function",
                    "function": {
                        "name": "query_skill_data",
                        "description": "查询当前 skill 目录下的 JSON 数据文件，只返回匹配片段，避免把大文件全部放进上下文。",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "relative_path": {
                                    "type": "string",
                                    "description": "相对于当前 skill 目录的数据文件路径，例如 data/source_policy.json。",
                                },
                                "query": {
                                    "type": "string",
                                    "description": "过滤关键词。为空时返回前几条。",
                                    "default": "",
                                },
                                "limit": {
                                    "type": "integer",
                                    "description": "最多返回多少条。",
                                    "default": 5,
                                },
                            },
                            "required": ["relative_path"],
                        },
                    },
                },
            },
        }

# Structured Writing Advanced Guide

## 正式语气

适合：

- 工作邮件
- 项目汇报
- 对外说明
- 课程作业
- 正式通知

常用表达：

- “目前”
- “预计”
- “建议”
- “请确认”
- “后续将”
- “为确保……”

避免表达：

- “反正”
- “随便”
- “差不多”
- “搞一下”
- “应该还行”

## 委婉语气

适合：

- 拒绝
- 延期
- 催促
- 说明问题
- 请求配合

可用表达：

- “可能需要更多时间”
- “为了确保质量”
- “是否方便在……前确认”
- “我们建议优先……”
- “如果方便的话，请……”

## 强硬但礼貌

适合：

- 明确边界
- 提醒风险
- 强调截止时间
- 要求对方行动

可用表达：

- “该事项需要在……前完成”
- “如果未能按时确认，可能会影响……”
- “为避免后续风险，请……”
- “请在……前完成确认”

## 营销风格

结构建议：

1. 先讲痛点
2. 再讲解决方案
3. 最后讲收益

示例结构：

> 你是否遇到过……  
> 现在，通过……  
> 你可以更轻松地……

---
name: structured-writing

description: >
  用于改写、润色、整理、扩写或压缩用户提供的文本，包括邮件、报告段落、说明文、社交媒体文案和学习笔记。
  当用户提供一段草稿，并希望它更清晰、更专业、更自然、更有结构、更正式、更委婉或更简洁时，应该使用此技能。
  核心价值是把粗糙表达转化成可直接使用的高质量文本，同时保留用户原意。

version: 1.0.0

allowed_tools: [load_skill_resource]

required_context: [user_draft, target_audience, tone]

license: MIT

author: Demo Author <demo@example.com>

tags: [writing, editing, polish, structure]
---

# Structured Writing Skill

## 概述

这个技能用于文本改写和结构化表达。

它不负责事实检索。

如果用户需要最新信息、事实核查或资料来源，应使用 `web-research` skill。

适用场景包括：

- 改写一句话
- 润色邮件
- 整理学习笔记
- 优化报告段落
- 改写社交媒体文案
- 把口语表达改得更正式
- 把冗长表达改得更简洁

## 前置条件

用户最好提供：

- 原始草稿
- 目标读者
- 语气要求
- 长度要求

如果缺少这些信息，可以根据上下文做合理默认：

- 目标读者：普通读者
- 语气：清晰、专业、自然
- 长度：中等

## 工作流程

1. 识别用户想要的改写目标：
   - 更专业
   - 更口语
   - 更简短
   - 更正式
   - 更委婉
   - 更有逻辑

2. 提取原文核心意思。

3. 保留用户原意，不新增未经提供的事实。

4. 优先改善：
   - 结构
   - 逻辑顺序
   - 表达清晰度
   - 语气一致性

5. 如果用户要求非常正式、委婉、强硬或营销风格，可以读取：
   - `load_skill_resource("advanced.md")`

6. 输出最终可直接使用的版本。

## 最佳实践

- 不要过度改写导致意思变化。
- 用户没有要求多版本时，默认给一个最佳版本。
- 如果用户要求“只要结果”，不要解释。
- 如果原文有明显歧义，可以在最终答案后简短说明。
- 不要为了“高级”而堆砌空话。
- 不要添加原文没有的承诺、数据、事实、时间点。

## 示例

用户：

> 帮我把这句话改得专业一点：这个功能现在还没做完，可能要晚点。

输出：

> 该功能目前仍在开发中，预计需要更多时间完成。我们会在进展明确后及时同步更新。

## 故障排查

- 如果用户要求“不要改变意思”，严格保留语义。
- 如果用户要求“更短”，优先删除重复信息。
- 如果用户要求“更强硬”，提高明确性，但避免攻击性表达。
- 如果用户要求事实补充，但没有提供资料，应提醒需要使用检索类能力。

[
  {
    "source_type": "official_docs",
    "trust_level": "high",
    "use_for": ["API usage", "configuration", "version behavior", "installation"],
    "caution": "官方文档适合确认 API、配置、版本行为，但可能不覆盖真实用户体验。"
  },
  {
    "source_type": "github",
    "trust_level": "high",
    "use_for": ["source code", "issues", "release notes", "examples"],
    "caution": "GitHub issue 中的信息可能只是个别案例，需要注意 issue 状态和时间。"
  },
  {
    "source_type": "academic_paper",
    "trust_level": "high",
    "use_for": ["method", "experiment", "formal definition"],
    "caution": "论文结果可能不代表工业落地效果。"
  },
  {
    "source_type": "news_media",
    "trust_level": "medium",
    "use_for": ["recent events", "public announcements", "market movement"],
    "caution": "注意发布时间、信息来源和媒体立场。"
  },
  {
    "source_type": "personal_blog",
    "trust_level": "medium",
    "use_for": ["practice experience", "tutorial", "engineering tricks"],
    "caution": "需要和官方资料交叉验证。"
  },
  {
    "source_type": "forum_or_social_media",
    "trust_level": "low",
    "use_for": ["user sentiment", "bug symptoms", "anecdotal evidence"],
    "caution": "不要把个别用户发言当成事实结论。"
  }
]

# Web Research Advanced Guide

## 查询改写策略

### 技术框架类问题

用户问：

> X 框架现在怎么用？

可以搜索：

- `X official docs`
- `X GitHub examples`
- `X agent framework tutorial`
- `X changelog`

### 最新动态类问题

用户问：

> 最近 X 有什么变化？

可以搜索：

- `X latest update`
- `X changelog`
- `X release notes`
- `X news`

### 事实核查类问题

至少搜索两个不同角度：

- 官方来源
- 独立媒体或第三方资料

## 来源优先级

一般优先级：

1. 官方文档
2. 官方 GitHub / changelog
3. 论文 / 标准文档
4. 主流媒体 / 专业机构
5. 个人博客
6. 论坛、社交媒体、二手转载

## 什么时候需要更多搜索

以下情况应该追加搜索：

- 搜索结果互相矛盾
- 用户要求“最新”
- 用户要求“准确引用来源”
- 涉及价格、政策、版本、API、法规
- 搜索结果明显不够权威

---
name: web-research

description: >
  用于回答需要联网检索、最新资料、多来源事实核查、资料综述或来源引用的问题。
  当用户的问题涉及最近变化、实时信息、外部网页资料、新闻、产品、框架、论文、技术动态或需要查证的信息时，应该使用此技能。
  核心价值是把开放网络信息检索、来源筛选、结果整合和不确定性说明组织成可靠的研究流程。

version: 1.0.0

allowed_tools: [tavily_search, load_skill_resource, query_skill_data]

required_context: [user_question]

license: MIT

author: Demo Author <demo@example.com>

tags: [web, research, tavily, fact-checking]
---

# Web Research Skill

## 概述

这个技能用于处理需要外部网页信息的问题。

它的重点不是“搜到一个结果就回答”，而是让 Agent：

1. 先理解用户真正想查什么。
2. 再设计合适的搜索 query。
3. 然后比较多个来源。
4. 最后综合回答，并说明不确定性。

适用场景包括：

- 最新技术动态
- 框架、工具、库的当前用法
- 产品、政策、价格、版本变化
- 需要来源支撑的事实核查
- 需要多来源综述的问题

## 前置条件

需要环境变量：

- `TAVILY_API_KEY`

如果没有这个 key，`tavily_search` 工具会返回错误信息。

## 工作流程

1. 判断用户问题是否需要最新信息或外部来源。

2. 将用户问题改写成简洁搜索 query。

3. 调用 `tavily_search` 获取 3 到 5 个结果。

4. 阅读每个结果的：
   - title
   - url
   - content
   - score

5. 优先使用更权威、更直接、更近期的来源。

6. 如果问题涉及来源可靠性、边缘情况或搜索策略，可以调用：
   - `load_skill_resource("advanced.md")`

7. 如果需要判断不同来源类型的可信度，可以调用：
   - `query_skill_data("data/source_policy.json", query="...")`

8. 最终回答必须：
   - 先给结论
   - 再给依据
   - 明确哪些信息来自搜索结果
   - 对不确定处说明“不确定”或“需要进一步验证”

## 最佳实践

- 查询词应短，不要把用户的长问题原样丢给搜索引擎。
- 技术问题优先找官方文档、GitHub、论文、主流技术博客。
- 新闻问题优先找多个独立媒体或官方发布。
- 不要把搜索结果里的营销话术直接当事实。
- 如果搜索结果不足，不要编造。
- 如果结果互相矛盾，应说明冲突，而不是强行给单一结论。

## 示例

用户：

> 请查一下 LangGraph 现在主要适合用来做什么？

推荐流程：

1. 搜索 `LangGraph current use cases agents`
2. 对比官方文档、GitHub、技术博客
3. 总结 LangGraph 的主要使用场景
4. 说明依据来自搜索结果

## 故障排查

- 如果 Tavily 返回结果很少，换更短 query。
- 如果结果质量差，加入 `official docs`、`GitHub`、`paper` 等限定词。
- 如果用户问的是观点类问题，不要只搜索一个来源。
- 如果用户要求“最新”，搜索词里可以加入年份或 `latest`。

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city} 的天气是晴天，25°C"

llm = ChatOpenAI(model="gpt-4o-mini")
tools = [get_weather]

# 仅把工具绑定到模型，让模型知道如何调用，但不自动执行
llm_with_tools = llm.bind_tools(tools)

# 调用后，如果模型决定调用工具，response 里会有 tool_calls 字段
response = llm_with_tools.invoke("北京今天天气怎么样？")
print(response.tool_calls)  # 输出：[{'name': 'get_weather', 'args': {'city': '北京'}, ...}]
# 但你需要自己写代码去执行这个 tool_calls，并把结果再喂给模型

# 注意：这是最新的推荐用法
from langchain import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city} 的天气是晴天，25°C"

llm = ChatOpenAI(model="gpt-4o-mini")
tools = [get_weather]

# create_agent 是 LangChain v1 的新函数，内部基于 LangGraph
agent = create_agent(model=llm, tools=tools)

# 使用方式和 LangGraph 一致
result = agent.invoke({"messages": [("user", "北京今天天气怎么样？")]})
print(result["messages"][-1].content)  # 输出：北京今天天气是晴天，25°C

from langchain import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import StructuredTool
import os
import yaml  # 需要安装：pip install pyyaml

# ============ 1. 加载 Skill ============
def load_skill(skill_folder_path: str) -> StructuredTool:
    """
    从文件夹加载 Skill，返回一个可调用的 Tool 对象
    """
    skill_md_path = os.path.join(skill_folder_path, "SKILL.md")
    
    # 解析 SKILL.md 的 YAML 头部
    with open(skill_md_path, "r", encoding="utf-8") as f:
        content = f.read()
    
    # 提取 YAML 头（--- 之间的内容）
    lines = content.split("\n")
    yaml_content = []
    in_yaml = False
    for line in lines:
        if line.strip() == "---":
            in_yaml = not in_yaml
            continue
        if in_yaml:
            yaml_content.append(line)
    
    yaml_data = yaml.safe_load("\n".join(yaml_content))
    skill_name = yaml_data.get("name", "unknown_skill")
    skill_description = yaml_data.get("description", "No description provided")
    
    # 导入执行器
    from skill_executor import execute_code_review
    
    # 创建 Tool 对象（核心：把 Skill 包装成 Tool）
    def skill_wrapper(code: str) -> str:
        """执行 Skill 的逻辑"""
        return execute_code_review(code, skill_folder_path)
    
    return StructuredTool.from_function(
        func=skill_wrapper,
        name=skill_name,
        description=skill_description
    )

# ============ 2. 注册 Skill 到 Agent ============
# 加载 Skill
skill_tool = load_skill("./my_skills/code_review_skill/")

# 创建 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 用 create_agent 创建 Agent（这是官方推荐的方式）
agent = create_agent(
    model=llm,
    tools=[skill_tool],  # 把 Skill 当作普通 Tool 注册进去
    system_prompt="你是一个专业的代码审查助手，擅长发现代码中的问题并给出改进建议。"
)

# ============ 3. 使用 Skill ============
user_input = """
帮我 review 这段 Python 代码：

def sum(a,b):
    return a+b
"""

# 调用 Agent
result = agent.invoke({
    "messages": [
        ("user", user_input)
    ]
})

# 打印结果
print(result["messages"][-1].content)

用户输入
   ↓
LLM
   │
   ├─ 普通生成
   │      ↓
   │   一段字符串
   │      ↓
   │   json-repair      ← 修 JSON 语法
   │      ↓
   │   Pydantic         ← 验证字段、类型、结构
   │
   ├─ Tool Calling
   │      ↓
   │   tool.arguments   ← API 层按参数 Schema 生成
   │      ↓
   │   你的 Python 函数
   │
   ├─ Structured Outputs
   │      ↓
   │   JSON Schema      ← API 层要求最终回答符合 Schema
   │      ↓
   │   Pydantic 对象
   │
   └─ 本地模型 / vLLM / Outlines
          │
          ├─ 受限解码，每生成一个 token 就限制合法 token
          │
          └─ SFT

流式场景：
token/chunk → token/chunk → token/chunk
                    ↓
            partial-json-parser
                    ↓
             当前已完整字段

prompt = """
必须返回 JSON！！！
绝对不允许输出其他内容！！！
格式错误将受到惩罚！！！
"""

好的，结果如下：

```json
{
  "name": "Tom",
  "age": 20
}
```

import re

match = re.search(r"\{.*\}", text, re.S)

再问一次模型：
“你的 JSON 格式错了，请重新回答！”

json.loads(...)

import json

json.loads("""
{
    'name': 'Tom',
    'age': 20,
}
""")

{
    "name": "Tom",
    "age": 20
}

from json_repair import loads

text = """
{
    'name': 'Tom',
    'age': 20,
"""

obj = loads(text)

print(obj)

{
    "name": "Tom",
    "age": 20
}

{
    "username": "Tom",
    "age": "二十岁"
}

{
    "name": "Tom",
    "age": 20
}

{
    "name": "小王",
    "age": 23,
    "skills": ["Python", "SQL"]
}

from pydantic import BaseModel


class Person(BaseModel):
    name: str
    age: int
    skills: list[str]

Person
├── name: 字符串
├── age: 整数
└── skills: 字符串数组

data = {
    "name": "小王",
    "age": 23,
    "skills": ["Python", "SQL"]
}

person = Person.model_validate(data)

print(person)

name='小王' age=23 skills=['Python', 'SQL']

person.name
person.age
person.skills

schema = Person.model_json_schema()

print(schema)

{
  "properties": {
    "name": {
      "title": "Name",
      "type": "string"
    },
    "age": {
      "title": "Age",
      "type": "integer"
    },
    "skills": {
      "items": {
        "type": "string"
      },
      "title": "Skills",
      "type": "array"
    }
  },
  "required": [
    "name",
    "age",
    "skills"
  ],
  "title": "Person",
  "type": "object"
}

Python 类型定义
       ↓
   Pydantic
       ↓
 JSON Schema
       ↓
OpenAI / vLLM / Tool Calling / Structured Outputs

Python 类型写一遍
JSON Schema 再手写一遍
校验代码再写一遍

LLM
 ↓
字符串
 ↓
json-repair
 ↓
Python dict
 ↓
Pydantic
 ↓
合法、带类型的数据对象

raw_text
    ↓
json_repair.loads()
    ↓
dict
    ↓
Person.model_validate()
    ↓
Person

repair JSON 语法
        ↓
Pydantic 检查业务结构

import os

from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel
from json_repair import loads as repair_loads


load_dotenv()

client = OpenAI(
    base_url=os.getenv("url"),
    api_key=os.getenv("apikey"),
)

MODEL = os.getenv("modelid")


class Person(BaseModel):
    name: str
    age: int
    skills: list[str]


response = client.chat.completions.create(
    model=MODEL,
    messages=[
        {
            "role": "user",
            "content": """
从下面文本中提取信息，并返回 JSON：

小王今年23岁，会Python和SQL。

需要字段：
name
age
skills
"""
        }
    ]
)

# 1. LLM 原始输出
raw = response.choices[0].message.content

print("LLM原始输出：")
print(raw)


# 2. 修复 / 解析 JSON
obj = repair_loads(raw)

print("\n修复后的dict：")
print(obj)


# 3. Pydantic验证
person = Person.model_validate(obj)

print("\nPydantic对象：")
print(person)

print("\n程序读取：")
print(person.name)
print(person.age)
print(person.skills)

                    API级结构约束
                         │
              ┌──────────┴──────────┐
              ↓                     ↓
        Tool Calling       Structured Outputs
              ↓                     ↓
       函数参数结构             最终回答结构

def create_user(name: str, age: int):
    ...

帮我创建用户小王，今年23岁

好的，我将创建小王。

{
    "name": "小王",
    "age": 23
}

create_user(
    name="小王",
    age=23
)

tools=[
    {
        "type": "function",
        "function": {
            "name": "...",
            "description": "...",
            "parameters": {
                # JSON Schema
            }
        }
    }
]

函数名称
函数说明
函数参数 JSON Schema

create_user(...)

{
    "name": "create_user",
    "arguments": "{\"name\":\"小王\",\"age\":23}"
}

tool_call = response.choices[0].message.tool_calls[0]

print(tool_call.function.name)
print(tool_call.function.arguments)

create_user

{"name":"小王","age":23}

args = json.loads(tool_call.function.arguments)

create_user(**args)

用户
 │
 │ “创建小王，23岁”
 ↓
LLM
 │
 │ tool call
 ↓
{
  name: create_user,
  arguments: {
      name: 小王,
      age: 23
  }
}
 │
 ↓
你的程序
 │
 ↓
create_user(name="小王", age=23)

class Person(BaseModel):
    name: str
    age: int
    skills: list[str]

openai.pydantic_function_tool(...)

Pydantic
   ↓
JSON Schema
   ↓
OpenAI function tool schema

import os
import openai
from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel

load_dotenv()

client = OpenAI(
    base_url=os.getenv("url"),
    api_key=os.getenv("apikey"),
)

MODEL = os.getenv("modelid")

class Person(BaseModel):
    name: str
    age: int
    skills: list[str]

# 创建辅助工具，自动将 Pydantic 模型转换为 OpenAI 要求的 Function Calling 格式
person_tool = openai.pydantic_function_tool(
    Person, # Pydantic 模型类，定义了数据结构
    name="submit_person", # 工具函数的名称，AI 调用时会用这个名字    
    description="提交抽取到的人员信息"# 告诉 AI 这个工具是做什么的，帮助 AI 决定何时调用
)

response = client.chat.completions.create(
    model=MODEL,
    messages=[
        {"role": "user", "content": "小王今年23岁，会Python和SQL，请提交他的人员信息。"}
    ],
    tools=[person_tool],
    tool_choice={ # tool_choice是个参数，强制调用某个函数
        "type": "function",
        "function": {"name": "submit_person"}
    }
    # 自动（默认）- AI 自己决定是否调用：tool_choice="auto"
    # 不调用任何函数：tool_choice="none"
)

# 处理响应
tool_call = response.choices[0].message.tool_calls[0]
print("调用的工具：", tool_call.function.name)
print("\n原始arguments：", tool_call.function.arguments)

# Pydantic验证
person = Person.model_validate_json(tool_call.function.arguments)
print("\nPydantic对象：", person)
print(f"\n下游读取：\n姓名：{person.name}\n年龄：{person.age}\n技能：{', '.join(person.skills)}")

调用的工具： submit_person
原始arguments： {"name":"小王","age":23,"skills":["Python","SQL"]}
Pydantic对象： name='小王' age=23 skills=['Python', 'SQL']
下游读取：
姓名：小王
年龄：23
技能：Python, SQL

{
    "type": "function",
    "function": {
        "name": "submit_person",
        "description": "提交抽取到的人员信息",
        "parameters": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "age": {"type": "integer"},
                "skills": {"type": "array", "items": {"type": "string"}}
            },
            "required": ["name", "age", "skills"]
        }
    }
}

person_tool = {
    "type": "function",
    "function": {
        "name": "submit_person",
        "description": "提交抽取到的人员信息",
        "parameters": {
            "type": "object",
            "properties": {
                "name": {
                    "type": "string",
                    "description": "人员姓名"
                },
                "age": {
                    "type": "integer",
                    "description": "人员年龄"
                },
                "skills": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "技能列表"
                }
            },
            "required": ["name", "age", "skills"]
        }
    }
}

def submit_person(name, age, skills):
    print("写入数据库：")
    print(name, age, skills)

submit_person(
    name=person.name,
    age=person.age,
    skills=person.skills
)

submit_person(**person.model_dump())

Prompt：
“拜托你严格按照JSON输出”

submit_person()

{
    "name": "小王",
    "age": 23,
    "skills": ["Python", "SQL"]
}

用户
 ↓
LLM
 ↓
任意文字

用户
 ↓
LLM
 │
 │ Schema = Person
 ↓
{
    "name": str,
    "age": int,
    "skills": list[str]
}

JSON Mode
    ↓
“是JSON就行”

Structured Outputs
    ↓
“不仅得是JSON，
 而且必须符合我规定的JSON Schema”

import os

from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel, ConfigDict


load_dotenv()

client = OpenAI(
    base_url=os.getenv("url"),
    api_key=os.getenv("apikey"),
)

MODEL = os.getenv("modelid")


class Person(BaseModel):
    # 不允许额外字段
    model_config = ConfigDict(extra="forbid")

    name: str
    age: int
    skills: list[str]


schema = Person.model_json_schema()

print("Pydantic生成的JSON Schema：")
print(schema)


response = client.chat.completions.create(
    model=MODEL,

    messages=[
        {
            "role": "user",
            "content": "小王今年23岁，会Python和SQL，请提取他的人员信息。"
        }
    ],

    response_format={
        "type": "json_schema",

        "json_schema": {
            "name": "person",
            "schema": schema,
            "strict": True
        }
    }
)


raw = response.choices[0].message.content

print("\n模型返回：")
print(raw)


person = Person.model_validate_json(raw)

print("\nPydantic对象：")
print(person)

print(person.name)
print(person.age)
print(person.skills)

response = client.chat.completions.parse(
    model=MODEL,
    messages=[
        {
            "role": "user",
            "content": "小王今年23岁，会Python和SQL，请提取他的人员信息。"
        }
    ],
    response_format=Person
)


person = response.choices[0].message.parsed

print(person)
print(person.name)

{"age":

token       probability

"          30%
2          25%
3          20%
null       10%
hello       5%
{           3%
...

所有 token
   ↓
根据 logits / probability
   ↓
选一个

{
    "age": "twenty three"
}

"age": {
    "type": "integer"
}

{"age":

空格
-
0~9
...

"
hello
[
{
true

模型 logits

"      9.1
2      8.7
hello  8.0
3      7.6
{      6.9

      ↓ JSON grammar mask

"      -∞
2      8.7
hello  -∞
3      7.6
{      -∞

      ↓

只能从 2 / 3 / ... 中采样

json.loads(buffer)

JSONDecodeError

{"name":"小王","age":

loads('{"key": "v')

LLM streaming
       ↓
buffer不断增长
       ↓
partial JSON parser
       ↓
React / Vue 页面不断刷新

from partial_json_parser import loads, Allow


chunks = [
    '{"name":',
    '"小王",',
    '"age":',
    '23,',
    '"skills":',
    '["Python",',
    '"SQL"]',
    '}'
]


buffer = ""

for chunk in chunks:

    buffer += chunk

    print("\n当前buffer：")
    print(buffer)

    try:
        obj = loads(
            buffer,
            Allow.OBJ | Allow.ARR
        )

        print("目前可解析：")
        print(obj)

    except Exception:
        pass

目前可解析：
{}

{
    'name': '小王'
}

{
    'name': '小王',
    'age': 23
}

{
    'name': '小王',
    'age': 23,
    'skills': ['Python', 'SQL']
}

import os

from dotenv import load_dotenv
from openai import OpenAI
from partial_json_parser import loads, Allow


load_dotenv()

client = OpenAI(
    base_url=os.getenv("url"),
    api_key=os.getenv("apikey"),
)

MODEL = os.getenv("modelid")


stream = client.chat.completions.create(
    model=MODEL,

    messages=[
        {
            "role": "user",
            "content": """
请用JSON返回下面的信息：

小王今年23岁，会Python和SQL。

格式：
{
    "name": "...",
    "age": 0,
    "skills": ["..."]
}
"""
        }
    ],

    # 至少保证最终是JSON
    response_format={
        "type": "json_object"
    },

    stream=True
)


buffer = ""


for chunk in stream:

    delta = chunk.choices[0].delta.content

    if not delta:
        continue

    buffer += delta

    print(delta, end="", flush=True)

    try:

        partial_data = loads(
            buffer,
            Allow.OBJ | Allow.ARR
        )

        print("\n\n当前已经可解析的数据：")
        print(partial_data)
        print("\n继续接收：")

    except Exception:
        pass

5秒后完整JSON突然出现

name    小王
          ↓
age     23
          ↓
skills  Python
          ↓
skills  Python, SQL

final_person = Person.model_validate_json(buffer)

class Output(BaseModel):
    ...


response = client.chat.completions.parse(
    model=MODEL,
    messages=messages,
    response_format=Output,
)

result = response.choices[0].message.parsed

查数据库
调用搜索
调用 MCP
发送邮件
查询设备
执行某个 action

class ToolArgs(BaseModel):
    ...


tools = [
    openai.pydantic_function_tool(
        ToolArgs,
        name="xxx"
    )
]

Prompt要求JSON
    ↓
json-repair
    ↓
Pydantic

Qwen / Llama
    ↓
vLLM

Pydantic
 ↓
JSON Schema
 ↓
vLLM structured_outputs
 ↓
受限解码