​

​

# ==================== 第一部分：Agent系统提示词 ====================
# 作用：定义Agent的核心行为准则和工作流程
# 在Agent中：作为系统级指令，指导LLM如何思考、行动和输出
AGENT_SYSTEM_PROMPT = """
你是一个智能旅行助手。你的任务是分析用户的请求，并使用可用工具一步步地解决问题。

# 可用工具:
- `get_weather(city: str)`: 查询指定城市的实时天气。
- `get_attraction(city: str, weather: str)`: 根据城市和天气搜索推荐的旅游景点。

# 输出格式要求:
你的每次回复必须严格遵循以下格式，包含一对Thought和Action：

Thought: [你的思考过程和下一步计划]
Action: [你要执行的具体行动]

Action的格式必须是以下之一：
1. 调用工具：function_name(arg_name="arg_value")
2. 结束任务：Finish[最终答案]

# 重要提示:
- 每次只输出一对Thought-Action
- Action必须在同一行，不要换行
- 当收集到足够信息可以回答用户问题时，必须使用 Action: Finish[最终答案] 格式结束

请开始吧！
"""


# ==================== 第二部分：工具函数定义 ====================
# 作用：提供Agent可调用的具体能力（工具）
# 在Agent中：每个工具函数对应一个具体操作，Agent通过调用它们来获取信息或执行任务

import requests

def get_weather(city: str) -> str:
    """
    作用：查询真实天气信息
    在Agent中：当Agent需要获取某个城市的天气时调用此函数
    输入：城市名称
    输出：格式化的天气描述字符串
    """
    # 调用第三方天气API获取实时数据
    url = f"https://wttr.in/{city}?format=j1"
    
    try:
        response = requests.get(url)
        response.raise_for_status() 
        data = response.json()
        
        # 从返回的JSON中提取温度、天气描述等信息
        current_condition = data['current_condition'][0]
        weather_desc = current_condition['weatherDesc'][0]['value']
        temp_c = current_condition['temp_C']
        
        return f"{city}当前天气:{weather_desc}，气温{temp_c}摄氏度"
        
    except requests.exceptions.RequestException as e:
        return f"错误:查询天气时遇到网络问题 - {e}"
    except (KeyError, IndexError) as e:
        return f"错误:解析天气数据失败，可能是城市名称无效 - {e}"


import os
from tavily import TavilyClient

def get_attraction(city: str, weather: str) -> str:
    """
    作用：根据天气和城市推荐景点
    在Agent中：当Agent获取天气后，根据天气情况推荐合适的景点
    输入：城市名称、天气描述
    输出：景点推荐信息字符串
    """
    # 使用Tavily搜索API进行智能搜索
    api_key = os.environ.get("TAVILY_API_KEY")
    if not api_key:
        return "错误:未配置TAVILY_API_KEY环境变量。"

    tavily = TavilyClient(api_key=api_key)
    
    # 构造精确的搜索查询，将天气条件纳入考虑
    query = f"'{city}' 在'{weather}'天气下最值得去的旅游景点推荐及理由"
    
    try:
        response = tavily.search(query=query, search_depth="basic", include_answer=True)
        
        # 优先返回Tavily生成的综合回答
        if response.get("answer"):
            return response["answer"]
        
        # 否则返回搜索结果摘要
        formatted_results = []
        for result in response.get("results", []):
            formatted_results.append(f"- {result['title']}: {result['content']}")
        
        if not formatted_results:
             return "抱歉，没有找到相关的旅游景点推荐。"

        return "根据搜索，为您找到以下信息:\n" + "\n".join(formatted_results)

    except Exception as e:
        return f"错误:执行Tavily搜索时出现问题 - {e}"

# 将工具函数注册到字典中，便于Agent动态调用
# 作用：建立工具名称到函数对象的映射
# 在Agent中：当Agent解析出要调用的工具名称后，从此字典查找对应的函数
available_tools = {
    "get_weather": get_weather,
    "get_attraction": get_attraction,
}


# ==================== 第三部分：LLM客户端封装 ====================
# 作用：封装与大语言模型的通信逻辑
# 在Agent中：负责将Agent的思考过程（prompt）发送给LLM，获取下一步行动计划

from openai import OpenAI

class OpenAICompatibleClient:
    """
    作用：统一接口调用各种兼容OpenAI API的LLM服务
    在Agent中：作为Agent的"大脑"，每次循环调用此客户端获取Thought-Action
    """
    def __init__(self, model: str, api_key: str, base_url: str):
        self.model = model
        self.client = OpenAI(api_key=api_key, base_url=base_url)

    def generate(self, prompt: str, system_prompt: str) -> str:
        """调用LLM API来生成回应。"""
        print("正在调用大语言模型...")
        try:
            messages = [
                {'role': 'system', 'content': system_prompt},
                {'role': 'user', 'content': prompt}
            ]
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                stream=False
            )
            answer = response.choices[0].message.content
            print("大语言模型响应成功。")
            return answer
        except Exception as e:
            print(f"调用LLM API时发生错误: {e}")
            return "错误:调用语言模型服务时出错。"


# ==================== 第四部分：Agent主循环 ====================
# 作用：实现ReAct（Reasoning + Acting）模式的Agent核心逻辑
# 在Agent中：协调思考（Reasoning）和行动（Acting）的完整流程

import re

# --- 4.1 配置LLM客户端和初始化环境 ---
# 作用：设置Agent运行所需的API密钥和模型参数
# 在Agent中：决定使用哪个LLM作为推理引擎
API_KEY = "YOUR_API_KEY"
BASE_URL = "YOUR_BASE_URL"
MODEL_ID = "YOUR_MODEL_ID"
TAVILY_API_KEY="YOUR_Tavily_KEY"
os.environ['TAVILY_API_KEY'] = "YOUR_TAVILY_API_KEY"

llm = OpenAICompatibleClient(
    model=MODEL_ID,
    api_key=API_KEY,
    base_url=BASE_URL
)

# --- 4.2 初始化用户请求和对话历史 ---
# 作用：记录完整的交互过程
# 在Agent中：维护Thought-Action-Observation的循环历史，保持上下文连贯性
user_prompt = "你好，请帮我查询一下今天北京的天气，然后根据天气推荐一个合适的旅游景点。"
prompt_history = [f"用户请求: {user_prompt}"]

print(f"用户输入: {user_prompt}\n" + "="*40)

# --- 4.3 Agent主循环（最多5次）---
# 作用：执行ReAct循环直到任务完成或达到最大次数
# 在Agent中：每个循环包含 思考(Thought)->行动(Action)->观察(Observation) 三步骤
for i in range(5):
    print(f"--- 循环 {i+1} ---\n")
    
    # 步骤1：构建完整prompt（包含历史）
    # 作用：为LLM提供完整的上下文
    full_prompt = "\n".join(prompt_history)
    
    # 步骤2：调用LLM进行思考，获取下一步行动
    # 作用：Agent的"思考"环节
    llm_output = llm.generate(full_prompt, system_prompt=AGENT_SYSTEM_PROMPT)
    
    # 清理输出：确保只保留第一个Thought-Action对
    # 作用：防止LLM一次性生成多个动作，保持Agent的单步执行特性
    match = re.search(r'(Thought:.*?Action:.*?)(?=\n\s*(?:Thought:|Action:|Observation:)|\Z)', llm_output, re.DOTALL)
    if match:
        truncated = match.group(1).strip()
        if truncated != llm_output.strip():
            llm_output = truncated
            print("已截断多余的 Thought-Action 对")
    print(f"模型输出:\n{llm_output}\n")
    prompt_history.append(llm_output)
    
    # 步骤3：解析Action字段
    # 作用：从LLM输出中提取具体要执行的操作
    action_match = re.search(r"Action: (.*)", llm_output, re.DOTALL)
    if not action_match:
        observation = "错误: 未能解析到 Action 字段。请确保你的回复严格遵循 'Thought: ... Action: ...' 的格式。"
        observation_str = f"Observation: {observation}"
        print(f"{observation_str}\n" + "="*40)
        prompt_history.append(observation_str)
        continue
    
    action_str = action_match.group(1).strip()

    # 步骤4：判断是否完成任务
    # 作用：检查Agent是否准备好输出最终答案
    if action_str.startswith("Finish"):
        final_answer = re.match(r"Finish\[(.*)\]", action_str).group(1)
        print(f"任务完成，最终答案: {final_answer}")
        break
    
    # 步骤5：解析工具调用
    # 作用：从Action字符串中提取工具名称和参数
    tool_name = re.search(r"(\w+)\(", action_str).group(1)
    args_str = re.search(r"\((.*)\)", action_str).group(1)
    kwargs = dict(re.findall(r'(\w+)="([^"]*)"', args_str))

    # 步骤6：执行工具函数（Agent的"行动"环节）
    # 作用：根据解析出的工具名称调用对应的函数
    if tool_name in available_tools:
        observation = available_tools[tool_name](**kwargs)
    else:
        observation = f"错误:未定义的工具 '{tool_name}'"

    # 步骤7：记录观察结果（Observation）
    # 作用：将工具执行结果反馈给Agent，用于下一轮思考
    observation_str = f"Observation: {observation}"
    print(f"{observation_str}\n" + "="*40)
    prompt_history.append(observation_str)

用户输入："查询北京天气，然后根据天气推荐景点"
         ↓
    ┌─────────────────────────────────────┐
    │  循环1：感知+思考                     │
    │  Thought: 需要先获取北京天气          │
    │  Action: get_weather(city="北京")    │
    └─────────────────────────────────────┘
         ↓
    【工具执行：调用天气API】
         ↓
    ┌─────────────────────────────────────┐
    │  观察(Observation):                  │
    │  北京当前天气:晴，气温25摄氏度        │
    └─────────────────────────────────────┘
         ↓
    ┌─────────────────────────────────────┐
    │  循环2：基于观察结果再次思考           │
    │  Thought: 已获得天气信息(晴天25°C)，  │
    │          现在需要推荐景点             │
    │  Action: get_attraction(city="北京", │
    │                      weather="晴")   │
    └─────────────────────────────────────┘
         ↓
    【工具执行：搜索景点推荐】
         ↓
    ┌─────────────────────────────────────┐
    │  观察(Observation):                  │
    │  推荐颐和园、故宫、圆明园...          │
    └─────────────────────────────────────┘
         ↓
    ┌─────────────────────────────────────┐
    │  循环3：最终思考                      │
    │  Thought: 信息已完整，可以回答用户    │
    │  Action: Finish[北京今天晴天25°C，   │
    │                  推荐去颐和园...]     │
    └─────────────────────────────────────┘

​

​

"""
ReAct Agent 实现
功能：创建一个能够调用外部工具（如搜索引擎）的智能体，通过思考-行动-观察循环回答问题
"""

import os
import re
from typing import List, Dict, Any
from openai import OpenAI
from dotenv import load_dotenv
from serpapi import SerpApiClient

# ==================== 1. 配置加载模块 ====================
# 作用：加载 .env 文件中的环境变量（API密钥、模型配置等）
load_dotenv()


# ==================== 2. LLM客户端封装模块 ====================
class HelloAgentsLLM:
    """
    作用：封装大语言模型（LLM）客户端
    功能：
    - 初始化OpenAI兼容的API连接
    - 提供 think() 方法调用模型进行思考（流式响应）
    """
    
    def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None):
        """
        初始化LLM客户端
        优先使用传入参数，否则从环境变量读取
        """
        self.model = model or os.getenv("LLM_MODEL_ID")
        apiKey = apiKey or os.getenv("LLM_API_KEY")
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")
        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 think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
        """
        调用LLM进行思考
        参数：
        - messages: 对话消息列表
        - temperature: 温度参数（控制随机性，默认0表示确定性输出）
        返回：模型响应的文本内容
        """
        print(f"🧠 正在调用 {self.model} 模型...")
        try:
            # 发起流式请求
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                stream=True,
            )
            
            # 处理流式响应
            print("✅ 大语言模型响应成功:")
            collected_content = []
            for chunk in response:
                if not chunk.choices:
                    continue
                content = chunk.choices[0].delta.content or ""
                print(content, end="", flush=True)  # 实时打印
                collected_content.append(content)
            print()  # 换行
            return "".join(collected_content)
        
        except Exception as e:
            print(f"❌ 调用LLM API时发生错误: {e}")
            return None


# ==================== 3. 工具定义模块 ====================
def search(query: str) -> str:
    """
    作用：搜索引擎工具（基于SerpApi）
    功能：
    - 执行Google搜索
    - 智能解析结果：优先返回答案框/知识图谱，否则返回前3个搜索摘要
    """
    print(f"🔍 正在执行 [SerpApi] 网页搜索: {query}")
    try:
        api_key = os.getenv("SERPAPI_API_KEY")
        if not api_key:
            return "错误：SERPAPI_API_KEY 未在 .env 文件中配置。"
        
        # 配置搜索参数
        params = {
            "engine": "google",
            "q": query,
            "api_key": api_key,
            "gl": "cn",      # 地区：中国
            "hl": "zh-cn",   # 语言：简体中文
        }
        
        # 执行搜索
        client = SerpApiClient(params)
        results = client.get_dict()
        
        # 智能解析结果（优先级从高到低）
        # 1. 答案列表
        if "answer_box_list" in results:
            return "\n".join(results["answer_box_list"])
        # 2. 答案框
        if "answer_box" in results and "answer" in results["answer_box"]:
            return results["answer_box"]["answer"]
        # 3. 知识图谱描述
        if "knowledge_graph" in results and "description" in results["knowledge_graph"]:
            return results["knowledge_graph"]["description"]
        # 4. 前3个有机搜索结果摘要
        if "organic_results" in results and results["organic_results"]:
            snippets = [
                f"[{i+1}] {res.get('title', '')}\n{res.get('snippet', '')}"
                for i, res in enumerate(results["organic_results"][:3])
            ]
            return "\n\n".join(snippets)
        
        return f"对不起，没有找到关于 '{query}' 的信息。"
    
    except Exception as e:
        return f"搜索时发生错误: {e}"


# ==================== 4. 工具执行器模块 ====================
class ToolExecutor:
    """
    作用：管理和执行工具的注册器
    功能：
    - registerTool(): 注册新工具
    - getTool(): 根据名称获取工具函数
    - getAvailableTools(): 获取所有可用工具的描述
    """
    
    def __init__(self):
        self.tools: Dict[str, Dict[str, Any]] = {}  # 存储工具：{名称: {描述, 函数}}
    
    def registerTool(self, name: str, description: str, func: callable):
        """注册一个新工具"""
        if name in self.tools:
            print(f"警告：工具 '{name}' 已存在，将被覆盖。")
        self.tools[name] = {"description": description, "func": func}
        print(f"工具 '{name}' 已注册。")
    
    def getTool(self, name: str) -> callable:
        """根据名称获取工具函数"""
        return self.tools.get(name, {}).get("func")
    
    def getAvailableTools(self) -> str:
        """获取所有可用工具的格式化描述（用于提示词）"""
        return "\n".join([
            f"- {name}: {info['description']}" 
            for name, info in self.tools.items()
        ])
        //ToDO：还可以进一步完善，输出tool所需的参数（数量及格式等），以便LLM执行。


# ==================== 5. ReAct提示词模板 ====================
# 作用：定义LLM的响应格式规范
# 强制LLM输出：Thought（思考）+ Action（行动）的结构
REACT_PROMPT_TEMPLATE = """
请注意，你是一个有能力调用外部工具的智能助手。

可用工具如下：
{tools}

请严格按照以下格式进行回应：

Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
Action: 你决定采取的行动，必须是以下格式之一：
- `{{tool_name}}[{{tool_input}}]`：调用一个可用工具。
- `Finish[最终答案]`：当你认为已经获得最终答案时。

**重要**：Finish 中的最终答案应该简洁明了，不要包含换行符。如果需要多行内容，请用空格或标点符号分隔。
**注意**：每次只能输出【一对】Thought 和 Action，完成当前 Action 并获得观察结果后，才能进行下一轮思考。

现在，请开始解决以下问题：
Question: {question}
History: {history}
"""


# ==================== 6. ReAct智能体核心模块 ====================
class ReActAgent:
    """
    作用：实现ReAct（Reasoning + Acting）范式的智能体
    工作流程：
    1. 思考（Thought）：分析问题并决定下一步
    2. 行动（Action）：调用工具或输出最终答案
    3. 观察（Observation）：获取工具执行结果
    4. 循环直至获得答案或达到最大步数
    """
    
    def __init__(self, llm_client: HelloAgentsLLM, tool_executor: ToolExecutor, max_steps: int = 6):
        """
        初始化ReAct智能体
        参数：
        - llm_client: LLM客户端
        - tool_executor: 工具执行器
        - max_steps: 最大循环步数（防止无限循环）
        """
        self.llm_client = llm_client
        self.tool_executor = tool_executor
        self.max_steps = max_steps
        self.history = []  # 记录行动-观察历史
    
    def run(self, question: str):
        """
        运行ReAct智能体解决问题
        这是主要入口函数，包含核心的思考-行动循环
        """
        self.history = []  # 重置历史
        current_step = 0
        
        while current_step < self.max_steps:
            current_step += 1
            print(f"\n--- 第 {current_step} 步 ---")
            
            # 步骤1：格式化提示词
            tools_desc = self.tool_executor.getAvailableTools()
            history_str = "\n".join(self.history)
            prompt = REACT_PROMPT_TEMPLATE.format(
                tools=tools_desc,
                question=question,
                history=history_str
            )
            
            # 步骤2：调用LLM进行思考
            messages = [{"role": "user", "content": prompt}]
            response_text = self.llm_client.think(messages=messages)
            
            if not response_text:
                print("错误：LLM未能返回有效响应。")
                break
            
            # 步骤3：解析LLM的输出（提取Thought和Action）
            thought, action = self._parse_output(response_text)
            
            if thought:
                print(f"💭 思考: {thought}")
            
            if not action:
                print("警告：未能解析出有效的Action，流程终止。")
                break
            
            # 清理 action 字符串（移除多余的换行和空格）
            action = action.strip()
            
            # 步骤4：执行Action
            # 情况A：输出最终答案（修复正则表达式以支持多行内容）
            if action.startswith("Finish"):
                # 使用更灵活的正则表达式匹配 Finish[...]，支持跨行匹配
                finish_match = re.match(r"Finish\[(.*?)\]", action, re.DOTALL)
                if finish_match:
                    final_answer = finish_match.group(1).strip()
                    print(f"🎉 最终答案: {final_answer}")
                    return final_answer
                else:
                    # 如果无法匹配标准格式，尝试提取 Finish 之后的所有内容
                    print(f"警告：无法解析 Finish 格式: {action}")
                    # 降级处理：移除 "Finish" 前缀后作为答案
                    final_answer = action.replace("Finish", "").strip("[] \n")
                    if final_answer:
                        print(f"🎉 最终答案（降级解析）: {final_answer}")
                        return final_answer
                    else:
                        print("错误：Finish 指令格式错误，无法提取答案。")
                        break
            
            # 情况B：调用工具
            tool_name, tool_input = self._parse_action(action)
            if not tool_name or not tool_input:
                print(f"警告：无法解析Action格式: {action}")
                # 记录无效action到历史，避免重复
                self.history.append(f"Action: {action}")
                self.history.append(f"Observation: 错误：Action格式无效，请使用正确格式 ToolName[input] 或 Finish[答案]")
                continue
            
            print(f"🎬 行动: {tool_name}[{tool_input}]")
            
            # 执行工具
            tool_function = self.tool_executor.getTool(tool_name)
            if not tool_function:
                observation = f"错误：未找到名为 '{tool_name}' 的工具。"
            else:
                observation = tool_function(tool_input)
            
            print(f"👀 观察: {observation}")
            
            # 记录历史（供下一轮LLM参考）
            self.history.append(f"Action: {action}")
            self.history.append(f"Observation: {observation}")
        
        # 超出最大步数
        print("⚠️ 已达到最大步数，流程终止。")
        return None
    
    def _parse_output(self, text: str):
        """
        作用：从LLM响应中提取Thought和Action
        使用正则表达式匹配格式：
        Thought: ... 
        Action: ...
        
        修复：支持 Action 内容跨越多行的情况
        """
        # 提取Thought（到Action或文本末尾）
        thought_match = re.search(r"Thought:\s*(.*?)(?=\nAction:|$)", text, re.DOTALL)
        
        # 提取Action（改进版：支持多行内容，直到遇到下一个Thought或文本末尾）
        # 使用非贪婪匹配，在遇到下一个 "Thought:" 或字符串结束时停止
        action_match = re.search(r"Action:\s*(.*?)(?=\nThought:|$)", text, re.DOTALL)
        
        # 如果没有匹配到，尝试匹配到字符串末尾
        if not action_match:
            action_match = re.search(r"Action:\s*(.*)$", text, re.DOTALL)
        
        thought = thought_match.group(1).strip() if thought_match else None
        action = action_match.group(1).strip() if action_match else None
        
        return thought, action
    
    def _parse_action(self, action_text: str):
        """
        作用：解析Action字符串，提取工具名称和输入参数
        格式：ToolName[input]
        
        修复：支持 input 中包含各种字符（包括中文标点、括号等）
        """
        # 改进正则表达式：支持工具名称后跟方括号，匹配到最后一个闭合方括号
        # 使用贪婪匹配确保获取完整的 input 内容
        match = re.match(r"(\w+)\[(.*)\]$", action_text, re.DOTALL)
        if match:
            return match.group(1), match.group(2).strip()
        
        # 如果上述匹配失败，尝试更宽松的匹配（允许输入中包含未转义的括号）
        match = re.match(r"(\w+)\[(.*)", action_text, re.DOTALL)
        if match:
            tool_name = match.group(1)
            tool_input = match.group(2).rstrip("]").strip()
            return tool_name, tool_input
        
        return None, None


# ==================== 7. 主程序入口 ====================
if __name__ == '__main__':
    # 1. 初始化LLM客户端
    llm = HelloAgentsLLM()
    
    # 2. 创建工具执行器并注册搜索引擎工具
    tool_executor = ToolExecutor()
    search_desc = "一个网页搜索引擎。当你需要回答关于时事、事实以及在你的知识库中找不到的信息时，应使用此工具。"
    tool_executor.registerTool("Search", search_desc, search)
    
    # 3. 创建ReAct智能体
    agent = ReActAgent(llm_client=llm, tool_executor=tool_executor, max_steps=6)
    
    # 4. 定义问题并运行
    question = "苹果最新发布的手机是什么（截至今天，2026年5月）？它的主要卖点是什么？"
    agent.run(question)

PS D:\Microsoft VS Code\vs work\codeworkvs\other\Agent\ReactDemo> python .\demo.py
工具 'Search' 已注册。

--- 第 1 步 ---
🧠 正在调用 xiaomi-mimo-v2.5-pro 模型...
✅ 大语言模型响应成功:
Thought: 用户想知道截至2026年5月苹果最新发布的手机型号及其主要卖点。由于我的知识库截止于2025年7月，我需要通过网络搜索获取最新信息。首先，我需要搜索苹果在2026年发布的新款手机信息，然后提取其主要卖点。

Action: Search[苹果 2026年 最新发布 手机 型号]
💭 思考: 用户想知道截至2026年5月苹果最新发布的手机型号及其主要卖点。由于我的知识库截止于2025年7月，我需要通过网络搜索获取最新信息。首先，我需要搜索苹果在2026年发布的新款手机信息，然后提取其主要卖点。
🎬 行动: Search[苹果 2026年 最新发布 手机 型号]
🔍 正在执行 [SerpApi] 网页搜索: 苹果 2026年 最新发布 手机 型号
👀 观察: [1] 苹果2026年新品前瞻：折叠屏iPhone在列，超20款产品将登场
... 发布Mac或是iPad。 iPhone 18 Pro系列，相比于往年，今年苹果将改变手机发布策略，秋季发布会上将只有两款Pro机型，标准版则会放到第二年的春季。这代 ...

[2] 2026年苹果手机推荐：五款旗舰机型深度解析与选购指南
五款机型——iPhone 13 Pro、iPhone Air、iPhone 17、iPhone 15和iPhone 14 Pro，每一款都展现了不同的特色与风格。 🕴️ 商务场合首选 续航表现卓越…

[3] 7 款新品齊發! 2026年的蘋果太瘋狂了! | 日更第43天| iPhone18 ...
7 款新品齊發! 2026年的蘋果太瘋狂了! | 日更第43天| iPhone18 / iPhone折疊機/ Siri 2.0 / 20週年Macbook / HomePod | 大耳朵TV.

--- 第 2 步 ---
🧠 正在调用 xiaomi-mimo-v2.5-pro 模型...
✅ 大语言模型响应成功:
Thought: 用户询问截至2026年5月，苹果最新发布的手机是什么及其主要卖点。根据提供的历史搜索结果，之前的搜索似乎包含了对2026年的预测或展望（例如提到iPhone 18 Pro系列和折叠屏iPhone），但这些可能并非“最新发布”的确切信息，特别是用户指定了“截至今天，2026年5月”。历史结果中的第二条似乎列出了一些旧型号（如iPhone 13 Pro、iPhone 17），这不太可能是2026年“最新”的。因此，我需要更精确地搜索2026年5月之前苹果实际发布的新款手机型号。我将使用搜索工具，查询关键词聚焦于2026年已发布或近期发布的iPhone型号。

Action: Search[苹果 2026年 已发布 手机 主要卖点]
💭 思考: 用户询问苹果截至2026年5月最新发布的手机及其主要卖点。由于我的知识截止到2025年，且当前日期是2026年5月，我需要搜索最新信息来回答这个问题。我将使用搜索引擎查找相关信息。
🎬 行动: Search[苹果 2026年 最新发布 手机 主要卖点]
🔍 正在执行 [SerpApi] 网页搜索: 苹果 2026年 已发布 手机 主要卖点
👀 观察: [1] 苹果首款折叠屏手机计划今年秋季发布！这些A股公司或受益
如无意外，苹果公司首款折叠屏手机将于今年秋季与苹果新款iPhone一起发布，目前各界对其出货量持乐观态度，预计A股苹果公司供应链企业业绩将获提振。

[2] 接下来，苹果的iPhone 和小米手机一样多
接下来，苹果的iPhone 和小米手机一样多 · iPhone 一年7 台，价位全覆盖 · 不过，我们都知道今年真正的主角，将会是苹果首款折叠屏iPhone Fold，真正的定名可能是「iPhone Ultra」 ...

[3] 2026年苹果iPhone手机推荐哪一款性价比最高？（8000字选 ...
iPhone 17系列的更新重点主要集中在A19系列处理器、高刷下放、摄影系统提升以及VC散热技术等方面，使得性能、摄影能力、显示技术和散热表现等方面都有了显著的提升。特别 ...

--- 第 3 步 ---
🧠 正在调用 xiaomi-mimo-v2.5-pro 模型...
✅ 大语言模型响应成功:
Thought: 我需要根据用户提供的搜索结果和当前日期（2026年5月）来回答。搜索结果显示苹果首款折叠屏手机计划于2026年秋季发布，因此尚未发布。而搜索结果[3]提到了iPhone 17系列及其更新重点，这暗示iPhone 17系列可能是截至2026年5月最新发布的手机。主要卖点从搜索结果[3]中提取，包括A19系列处理器、高刷下放、摄影系统提升和VC散热技术。

Action: Finish[苹果最新发布的手机是iPhone 17系列，主要卖点是搭载A19系列处理器、高刷新率显示、摄影系统提升和VC散热技术。]
💭 思考: 我需要根据用户提供的搜索结果和当前日期（2026年5月）来回答。搜索结果显示苹果首款折叠屏手机计划于2026年秋季发布，因此尚未发布。而搜索结果[3]提到了iPhone 17系列及其更新重点，这暗示iPhone 17系列可能是截至2026年5月最新发布的手机。主要卖点从搜索结果[3]中提取，包括A19系列处理器、高刷下放、摄影系统提升和VC散热技术。
🎉 最终答案: 苹果最新发布的手机是iPhone 17系列，主要卖点是搭载A19系列处理器、高刷新率显示、摄影系统提升和VC散热技术。

# OpenAI SDK 风格
from openai import OpenAI

client = OpenAI()
response = self.client.chat.completions.create(
# 统一接口，直接传递参数即可
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}], # 需要区分角色
    temperature=0.7,
    max_tokens=100,
    stream=True
)


# Transformers 库风格
from transformers import AutoModelForCausalLM, AutoTokenizer

model = AutoModelForCausalLM.from_pretrained("gpt2")
tokenizer = AutoTokenizer.from_pretrained("gpt2") # 需要先手动分词
inputs = tokenizer("Hello", return_tensors="pt")
streamer = TextStreamer(tokenizer, skip_prompt=True) # 流式需要先行定义类
outputs = model.generate(**inputs, max_length=50, streamer=streamer, temperature=0.7)

​

PLANNER_PROMPT_TEMPLATE = """
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。

问题: {question}

请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的:
```python
["步骤1", "步骤2", "步骤3", ...]
```
"""

EXECUTOR_PROMPT_TEMPLATE = """
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决“当前步骤”，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。

# 原始问题:
{question}

# 完整计划:
{plan}

# 历史步骤与结果:
{history}

# 当前步骤:
{current_step}

请仅输出针对“当前步骤”的回答:
"""

import os
import re
from typing import List, Dict, Any
from openai import OpenAI
from dotenv import load_dotenv
from serpapi import SerpApiClient
import ast
from typing import Any

load_dotenv()

class HelloAgentsLLM:
    """
    作用：封装大语言模型（LLM）客户端
    功能：
    - 初始化OpenAI兼容的API连接
    - 提供 think() 方法调用模型进行思考（流式响应）
    """
    
    def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None):
        """
        初始化LLM客户端
        优先使用传入参数，否则从环境变量读取
        """
        self.model = model or os.getenv("LLM_MODEL_ID")
        apiKey = apiKey or os.getenv("LLM_API_KEY")
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")
        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 think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
        """
        调用LLM进行思考
        参数：
        - messages: 对话消息列表
        - temperature: 温度参数（控制随机性，默认0表示确定性输出）
        返回：模型响应的文本内容
        """
        print(f"🧠 正在调用 {self.model} 模型...")
        try:
            # 发起流式请求
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                stream=True,
            )
            
            # 处理流式响应
            print("✅ 大语言模型响应成功:")
            collected_content = []
            for chunk in response:
                if not chunk.choices:
                    continue
                content = chunk.choices[0].delta.content or ""
                print(content, end="", flush=True)  # 实时打印
                collected_content.append(content)
            print()  # 换行
            return "".join(collected_content)
        
        except Exception as e:
            print(f"❌ 调用LLM API时发生错误: {e}")
            return None


PLANNER_PROMPT_TEMPLATE = """
    你是一个顶级的 AI 规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
    请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
    你的输出必须是一个 Python 列表，其中每个元素都是一个描述子任务的字符串。

    问题：
    {question}

    请严格按照以下格式输出计划。```python 与 ``` 作为前后缀是必要的：
    ```python
    ["步骤1", "步骤2", "步骤3", ...]
    ```
    """

EXECUTOR_PROMPT_TEMPLATE = """
    你是一位顶级的 AI 执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
    你将收到原始问题、完整计划，以及到目前为止已经完成的步骤和结果。
    请专注于解决“当前步骤”，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。

    原始问题：
    {question}

    完整计划：
    {plan}

    历史步骤与结果：
    {history}

    当前步骤：
    {current_step}

    请仅输出针对“当前步骤”的回答：
    """

class Planner:
    """负责将复杂问题拆分为多个可执行步骤。"""
    def __init__(self, llm_client: Any) -> None:
        self.llm_client = llm_client

    def plan(self, question: str) -> list[str]:
        """
        根据用户问题生成行动计划。

        Args:
            question: 用户提出的问题。

        Returns:
            由多个步骤组成的字符串列表。解析失败时返回空列表。
        """
        prompt = PLANNER_PROMPT_TEMPLATE.format(question=question)
        messages = [{"role": "user", "content": prompt}]

        print("\n--- 正在生成计划 ---")

        response_text = self.llm_client.think(messages=messages) or ""

        print(f"✅ 计划已生成：\n{response_text}")

        return self._parse_plan(response_text)

    @staticmethod
    def _parse_plan(response_text: str) -> list[str]:
        """
        从大语言模型的回复中提取 Python 列表。

        Args:
            response_text: 大语言模型返回的原始文本。

        Returns:
            合法的步骤列表。解析失败时返回空列表。
        """
        try:
            # 提取 ```python 和 ``` 之间的内容
            plan_str = response_text.split("```python", maxsplit=1)[1]
            plan_str = plan_str.split("```", maxsplit=1)[0].strip()

            # 安全解析 Python 字面量
            plan = ast.literal_eval(plan_str)

            # 检查列表中的每个元素是否都是非空字符串
            if not isinstance(plan, list):
                raise TypeError("计划必须是 Python 列表。")

            if not all(isinstance(step, str) and step.strip() for step in plan):
                raise TypeError("计划中的每个步骤都必须是非空字符串。")

            return plan

        except (ValueError, SyntaxError, IndexError, TypeError) as error:
            print(f"❌ 解析计划时出错：{error}")
            print(f"原始响应：\n{response_text}")
            return []
    
class Executor:
    """负责按照规划器生成的计划逐步执行任务。"""
    def __init__(self, llm_client: Any) -> None:
        self.llm_client = llm_client

    def execute(self, question: str, plan: list[str]) -> str:
        """
        根据行动计划逐步执行任务。

        Args:
            question: 用户提出的原始问题。
            plan: 规划器生成的行动计划。

        Returns:
            最后一个步骤的执行结果。
        """
        if not plan:
            return ""

        history = ""
        final_answer = ""

        print("\n--- 正在执行计划 ---")

        for index, current_step in enumerate(plan, start=1):
            print(f"\n-> 正在执行步骤 {index}/{len(plan)}：{current_step}")

            prompt = EXECUTOR_PROMPT_TEMPLATE.format(
                question=question,
                plan=plan,
                history=history or "无",
                current_step=current_step,
            )

            messages = [{"role": "user", "content": prompt}]

            response_text = self.llm_client.think(messages=messages) or ""

            history += (
                f"步骤 {index}：{current_step}\n"
                f"结果：{response_text}\n\n"
            )

            final_answer = response_text

            print(f"✅ 步骤 {index} 已完成，结果：\n{response_text}")

        return final_answer
    
class PlanAndSolveAgent:
    """先生成计划，再逐步执行计划。"""
    def __init__(self, llm_client: Any) -> None:
        self.planner = Planner(llm_client)
        self.executor = Executor(llm_client)

    def run(self, question: str) -> str:
        """
        运行完整的 Plan-and-Solve 流程。

        Args:
            question: 用户提出的问题。

        Returns:
            最终答案。规划失败时返回空字符串。
        """
        print(f"\n--- 开始处理问题 ---\n问题：{question}")

        # 1. 生成计划
        plan = self.planner.plan(question)

        if not plan:
            print("\n--- 任务终止 ---")
            print("无法生成有效的行动计划。")
            return ""

        # 2. 执行计划
        final_answer = self.executor.execute(question, plan)

        print(f"\n--- 任务完成 ---\n最终答案：\n{final_answer}")

        return final_answer
    
def main() -> None:
    """程序入口。"""
    llm_client = HelloAgentsLLM()

    agent = PlanAndSolveAgent(llm_client)

    question = input("请输入需要解决的问题：").strip()

    if not question:
        print("❌ 问题不能为空。")
        return

    agent.run(question)
    
if __name__ == "__main__":
    main()

PS D:\Microsoft VS Code\vs work\codeworkvs\other\Agent\demo> python .\PaSdemo.py
请输入需要解决的问题：计算1 + (15 + 7) × 3 - 10 的结果

--- 开始处理问题 ---
问题：计算1 + (15 + 7) × 3 - 10 的结果

--- 正在生成计划 ---
🧠 正在调用 gpt-5.5 模型...
✅ 大语言模型响应成功:
```python
["先计算括号内的加法：15 + 7", "将括号内的结果乘以 3", "计算 1 加上上一步的乘积", "从上一步的结果中减去 10", "得到最终计算结果"]
```
✅ 计划已生成：
```python
["先计算括号内的加法：15 + 7", "将括号内的结果乘以 3", "计算 1 加上上一步的乘积", "从上一步的结果中减去 10", "得到最终计算结果"]
```

--- 正在执行计划 ---

-> 正在执行步骤 1/5：先计算括号内的加法：15 + 7
🧠 正在调用 gpt-5.5 模型...
✅ 大语言模型响应成功:
22
✅ 步骤 1 已完成，结果：
22

-> 正在执行步骤 2/5：将括号内的结果乘以 3
🧠 正在调用 gpt-5.5 模型...
✅ 大语言模型响应成功:
66
✅ 步骤 2 已完成，结果：
66

-> 正在执行步骤 3/5：计算 1 加上上一步的乘积
🧠 正在调用 gpt-5.5 模型...
✅ 大语言模型响应成功:
67
✅ 步骤 3 已完成，结果：
67

-> 正在执行步骤 4/5：从上一步的结果中减去 10
🧠 正在调用 gpt-5.5 模型...
✅ 大语言模型响应成功:
57
✅ 步骤 4 已完成，结果：
57

-> 正在执行步骤 5/5：得到最终计算结果
🧠 正在调用 gpt-5.5 模型...
✅ 大语言模型响应成功:
57
✅ 步骤 5 已完成，结果：
57

--- 任务完成 ---
最终答案：
57

​

from typing import List, Dict, Any, Optional

class Memory:
    """
    一个简单的短期记忆模块，用于存储智能体的行动与反思轨迹。
    """

    def __init__(self):
        """
        初始化一个空列表来存储所有记录。
        """
        self.records: List[Dict[str, Any]] = []

    def add_record(self, record_type: str, content: str):
        """
        向记忆中添加一条新记录。

        参数:
        - record_type (str): 记录的类型 ('execution' 或 'reflection')。
        - content (str): 记录的具体内容 (例如，生成的代码或反思的反馈)。
        """
        record = {"type": record_type, "content": content}
        self.records.append(record)
        print(f"📝 记忆已更新，新增一条 '{record_type}' 记录。")

    def get_trajectory(self) -> str:
        """
        将所有记忆记录格式化为一个连贯的字符串文本，用于构建提示词。
        """
        trajectory_parts = []
        for record in self.records:
            if record['type'] == 'execution':
                trajectory_parts.append(f"--- 上一轮尝试 (代码) ---\n{record['content']}")
            elif record['type'] == 'reflection':
                trajectory_parts.append(f"--- 评审员反馈 ---\n{record['content']}")
        
        return "\n\n".join(trajectory_parts)

    def get_last_execution(self) -> Optional[str]:
        """
        获取最近一次的执行结果以供反思 (例如，最新生成的代码)。
        如果不存在，则返回 None。
        """
        for record in reversed(self.records):
            if record['type'] == 'execution':
                return record['content']
        return None

import os
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv
from openai import OpenAI

# 加载环境变量
load_dotenv()

class Memory:
    """
    一个简单的短期记忆模块，用于存储智能体的行动与反思轨迹。
    """

    def __init__(self):
        """
        初始化一个空列表来存储所有记录。
        """
        self.records: List[Dict[str, Any]] = []

    def add_record(self, record_type: str, content: str):
        """
        向记忆中添加一条新记录。

        参数:
        - record_type (str): 记录的类型 ('execution' 或 'reflection')。
        - content (str): 记录的具体内容 (例如，生成的代码或反思的反馈)。
        """
        record = {"type": record_type, "content": content}
        self.records.append(record)
        print(f"📝 记忆已更新，新增一条 '{record_type}' 记录。")

    def get_trajectory(self) -> str:
        """
        将所有记忆记录格式化为一个连贯的字符串文本，用于构建提示词。
        """
        trajectory_parts = []
        for record in self.records:
            if record['type'] == 'execution':
                trajectory_parts.append(f"--- 上一轮尝试 (代码) ---\n{record['content']}")
            elif record['type'] == 'reflection':
                trajectory_parts.append(f"--- 评审员反馈 ---\n{record['content']}")
        
        return "\n\n".join(trajectory_parts)

    def get_last_execution(self) -> Optional[str]:
        """
        获取最近一次的执行结果 (例如，最新生成的代码)。
        如果不存在，则返回 None。
        """
        for record in reversed(self.records):
            if record['type'] == 'execution':
                return record['content']
        return None


class HelloAgentsLLM:
    """
    作用：封装大语言模型（LLM）客户端
    功能：
    - 初始化OpenAI兼容的API连接
    - 提供 think() 方法调用模型进行思考（流式响应）
    """
    
    def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None):
        """
        初始化LLM客户端
        优先使用传入参数，否则从环境变量读取
        """
        self.model = model or os.getenv("LLM_MODEL_ID")
        apiKey = apiKey or os.getenv("LLM_API_KEY")
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")
        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 think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
        """
        调用LLM进行思考
        参数：
        - messages: 对话消息列表
        - temperature: 温度参数（控制随机性，默认0表示确定性输出）
        返回：模型响应的文本内容
        """
        print(f"🧠 正在调用 {self.model} 模型...")
        try:
            # 发起流式请求
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                stream=True,
            )
            
            # 处理流式响应
            print("✅ 大语言模型响应成功:")
            collected_content = []
            for chunk in response:
                if not chunk.choices:
                    continue
                content = chunk.choices[0].delta.content or ""
                print(content, end="", flush=True)  # 实时打印
                collected_content.append(content)
            print()  # 换行
            return "".join(collected_content)
        
        except Exception as e:
            print(f"❌ 调用LLM API时发生错误: {e}")
            return None


# 提示词模板
INITIAL_PROMPT_TEMPLATE = """
你是一位资深的Python程序员。请根据以下要求，编写一个Python函数。
你的代码必须包含完整的函数签名、文档字符串，并遵循PEP 8编码规范。

要求: {task}

请直接输出代码，不要包含任何额外的解释。
"""

REFLECT_PROMPT_TEMPLATE = """
你是一位极其严格的代码评审专家和资深算法工程师，对代码的性能有极致的要求。
你的任务是审查以下Python代码，并专注于找出其在<strong>算法效率</strong>上的主要瓶颈。

# 原始任务:
{task}

# 待审查的代码:
```python
{code}
```

请分析该代码的时间复杂度，并思考是否存在一种<strong>算法上更优</strong>的解决方案来显著提升性能。
如果存在，请清晰地指出当前算法的不足，并提出具体的、可行的改进算法建议（例如，使用筛法替代试除法）。
如果代码在算法层面已经达到最优，才能回答“无需改进”。

请直接输出你的反馈，不要包含任何额外的解释。
"""

REFINE_PROMPT_TEMPLATE = """
你是一位资深的Python程序员。你正在根据一位代码评审专家的反馈来优化你的代码。

# 原始任务:
{task}

# 你上一轮尝试的代码:
{last_code_attempt}
评审员的反馈：
{feedback}

请根据评审员的反馈，生成一个优化后的新版本代码。
你的代码必须包含完整的函数签名、文档字符串，并遵循PEP 8编码规范。
请直接输出优化后的代码，不要包含任何额外的解释。
"""

class ReflectionAgent:
    def __init__(self, llm_client, max_iterations=3):
        self.llm_client = llm_client
        self.memory = Memory()
        self.max_iterations = max_iterations

    def run(self, task: str):
        print(f"\n--- 开始处理任务 ---\n任务: {task}")

        # --- 1. 初始执行 ---
        print("\n--- 正在进行初始尝试 ---")
        initial_prompt = INITIAL_PROMPT_TEMPLATE.format(task=task)
        initial_code = self._get_llm_response(initial_prompt)
        self.memory.add_record("execution", initial_code)

        # --- 2. 迭代循环:反思与优化 ---
        for i in range(self.max_iterations):
            print(f"\n--- 第 {i+1}/{self.max_iterations} 轮迭代 ---")

            # a. 反思
            print("\n-> 正在进行反思...")
            last_code = self.memory.get_last_execution()
            reflect_prompt = REFLECT_PROMPT_TEMPLATE.format(task=task, code=last_code)
            feedback = self._get_llm_response(reflect_prompt)
            self.memory.add_record("reflection", feedback)

            # b. 检查是否需要停止
            if "无需改进" in feedback:
                print("\n✅ 反思认为代码已无需改进，任务完成。")
                break

            # c. 优化
            print("\n-> 正在进行优化...")
            refine_prompt = REFINE_PROMPT_TEMPLATE.format(
                task=task,
                last_code_attempt=last_code,
                feedback=feedback
            )
            refined_code = self._get_llm_response(refine_prompt)
            self.memory.add_record("execution", refined_code)
        
        final_code = self.memory.get_last_execution()
        print(f"\n--- 任务完成 ---\n最终生成的代码:\n```python\n{final_code}\n```")
        return final_code

    def _get_llm_response(self, prompt: str) -> str:
        """一个辅助方法，用于调用LLM并获取完整的流式响应。"""
        messages = [{"role": "user", "content": prompt}]
        response_text = self.llm_client.think(messages=messages) or ""
        return response_text

def simple_main():
    """
    简化版主函数，适合快速测试
    """
    # 初始化LLM客户端
    try:
        llm_client = HelloAgentsLLM()
        print("✅ LLM客户端初始化成功")
    except ValueError as e:
        print(f"❌ LLM客户端初始化失败: {e}")
        print("请检查.env文件配置")
        return
    
    # 创建反思代理
    agent = ReflectionAgent(llm_client, max_iterations=3)
    
    # 定义任务
    task = """
    编写一个函数 find_primes(n)，返回所有小于等于n的质数列表。
    要求算法尽可能高效。
    """
    
    # 运行代理
    final_code = agent.run(task)
    
    # # 保存代码
    # if final_code:
    #     with open("generated_code.py", "w", encoding="utf-8") as f:
    #         f.write(final_code)
    #     print("\n📁 最终代码已保存到 generated_code.py")


# 选择使用哪个main函数
if __name__ == "__main__":
    simple_main()

PS D:\Microsoft VS Code\vs work\codeworkvs\other\Agent\reflectiondemo> python .\demo.py
✅ LLM客户端初始化成功

--- 开始处理任务 ---
任务: 
    编写一个函数 find_primes(n)，返回所有小于等于n的质数列表。
    要求算法尽可能高效。
    

--- 正在进行初始尝试 ---
🧠 正在调用 gpt-5.5 模型...
✅ 大语言模型响应成功:
from math import isqrt


def find_primes(n: int) -> list[int]:
    """返回所有小于等于 n 的质数列表。"""
    if n < 2:
        return []

    sieve = bytearray(b"\x01") * (n + 1)
    sieve[0:2] = b"\x00\x00"

    for number in range(2, isqrt(n) + 1):
        if sieve[number]:
            start = number * number
            count = (n - start) // number + 1
            sieve[start:n + 1:number] = b"\x00" * count

    return [number for number, is_prime in enumerate(sieve) if is_prime]
📝 记忆已更新，新增一条 'execution' 记录。

--- 第 1/3 轮迭代 ---

-> 正在进行反思...
🧠 正在调用 gpt-5.5 模型...
✅ 大语言模型响应成功:
代码当前已经使用埃氏筛，不是试除法。

结论：算法层面无需替换为其他筛法；当前主算法已是实践中非常高效的方案。

关键评审反馈：

1. 时间复杂度

```text
O(n log log n)
```

原因是埃氏筛对每个质数标记其倍数，总标记次数约为：

```text
n/2 + n/3 + n/5 + ... = O(n log log n)
```

最终列表生成：

```python
[number for number, is_prime in enumerate(sieve) if is_prime]
```

需要完整扫描 `sieve`，复杂度为：

```text
O(n)
```

因此总复杂度仍为：

```text
O(n log log n)
```

2. 空间复杂度

```text
O(n)
```

`bytearray` 长度为 `n + 1`。

3. 当前主要瓶颈

主要瓶颈在于：

```python
sieve[start:n + 1:number] = b"\x00" * count
```

这是埃氏筛的核心标记操作，但该写法已经利用了 `bytearray` 切片赋值，底层由 C 层执行，比 Python 层循环逐个标记高效得多。

4. 是否存在算法上显著更优的方案

不存在值得替换的显著更优算法。

线性筛理论复杂度是：

```text
O(n)
```

但在 Python 中通常需要大量 Python 层循环和列表操作，实际性能往往不如当前这种基于 `bytearray` 切片的埃氏筛。

阿特金筛理论上也可能更优，但实现复杂、常数大，在 Python 中通常也不如当前方案稳定高效。

5. 可做的改进

只能做常数级优化，不是算法复杂度级别提升：

- 只筛奇数，减少约一半内存和标记量；
- 对超大 `n` 使用分段筛，降低内存占用并改善缓存局部性；
- 当前实现本身已经是高质量埃氏筛实现。

最终结论：

```text
无需进行算法层面的改进。
当前代码已经采用了适合该任务的高效筛法。
```
📝 记忆已更新，新增一条 'reflection' 记录。

-> 正在进行优化...
🧠 正在调用 gpt-5.5 模型...
✅ 大语言模型响应成功:
from math import isqrt


def find_primes(n: int) -> list[int]:
    """Return all prime numbers less than or equal to n."""
    if n < 2:
        return []
    if n == 2:
        return [2]

    size = (n + 1) // 2
    sieve = bytearray(b"\x01") * size
    sieve[0] = 0

    for number in range(3, isqrt(n) + 1, 2):
        index = number // 2
        if sieve[index]:
            start = number * number // 2
            count = (size - 1 - start) // number + 1
            sieve[start::number] = b"\x00" * count

    return [2] + [
        2 * index + 1
        for index in range(1, size)
        if sieve[index]
    ]
📝 记忆已更新，新增一条 'execution' 记录。

--- 第 2/3 轮迭代 ---

-> 正在进行反思...
🧠 正在调用 gpt-5.5 模型...
✅ 大语言模型响应成功:
时间复杂度：`O(n log log n)`  
空间复杂度：`O(n)`

该代码使用的是**只筛奇数的埃拉托色尼筛法**，不是低效的试除法。核心标记逻辑：

```python
sieve[start::number] = b"\x00" * count
```

整体仍然是标准埃氏筛的复杂度，只是通过跳过偶数将常数约减半。

主要算法瓶颈：

1. 仍然需要对每个质数的倍数进行批量标记，总标记规模为 `O(n log log n)`。
2. 需要维护长度约为 `n / 2` 的筛数组，空间为 `O(n)`。
3. 最后构造结果列表需要线性扫描筛数组，额外 `O(n)`。

是否存在算法上显著更优的方案：

- 对于“返回所有 `<= n` 的质数”这一任务，该实现已经是非常接近最优的通用方案。
- 理论上可以使用**线性筛**将复杂度降到 `O(n)`，但在 Python 中线性筛需要大量 Python 层循环和列表操作，实际性能通常不如当前这种基于 `bytearray` 切片批量赋值的埃氏筛。
- 分段筛可以降低内存占用并改善超大 `n` 时的缓存局部性，但时间复杂度仍然是 `O(n log log n)`，不是算法阶数上的显著提升。
- Atkin 筛理论复杂度更优，但常数大、实现复杂，在 Python 中通常也不如当前实现稳定高效。

结论：

当前代码在算法层面已经使用了高效筛法，并且做了跳过偶数的优化。不存在明显的、在 Python 中能显著提升性能的更优算法替代方案。

无需改进。
📝 记忆已更新，新增一条 'reflection' 记录。

✅ 反思认为代码已无需改进，任务完成。

--- 任务完成 ---
最终生成的代码:
```python
from math import isqrt


def find_primes(n: int) -> list[int]:
    """Return all prime numbers less than or equal to n."""
    if n < 2:
        return []
    if n == 2:
        return [2]

    size = (n + 1) // 2
    sieve = bytearray(b"\x01") * size
    sieve[0] = 0

    for number in range(3, isqrt(n) + 1, 2):
        index = number // 2
        if sieve[index]:
            start = number * number // 2
            count = (size - 1 - start) // number + 1
            sieve[start::number] = b"\x00" * count

    return [2] + [
        2 * index + 1
        for index in range(1, size)
        if sieve[index]
    ]
```

​

​

​

import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.messages import AIMessage, ToolMessage # 观察到模型决策的结果


# 使用 @tool 装饰器，将普通 Python 函数注册为 Agent 可以调用的工具。
# 函数名称、参数类型和文档字符串都会提供给大语言模型，
# 帮助模型理解工具的用途以及调用时需要传入哪些参数。
@tool
def calculate_rectangle_area(length: float, width: float) -> str:
    """
    计算矩形面积。

    Args:
        length: 矩形的长度。
        width: 矩形的宽度。

    Returns:
        计算结果。
    """
    area = length * width
    return f"矩形面积为 {area}"


@tool
def calculate_circle_area(radius: float) -> str:
    """
    计算圆形面积。

    Args:
        radius: 圆形的半径。

    Returns:
        计算结果。
    """
    pi = 3.1415926
    area = pi * radius * radius
    return f"圆形面积约为 {area:.2f}"


def create_model() -> ChatOpenAI:
    """
    创建大语言模型客户端。

    ChatOpenAI 不仅可以连接 OpenAI 官方模型，
    也可以连接兼容 OpenAI Chat Completions 协议的模型服务。
    本例使用阿里云百炼提供的兼容接口调用千问模型。
    """
    api_key = os.getenv("LLM_API_KEY")

    # 如果没有配置 API Key，则主动抛出异常，
    # 避免程序在调用模型时才出现不容易定位的错误。
    if not api_key:
        raise RuntimeError("请先在 .env 文件中设置 API_KEY")

    return ChatOpenAI(
        # 可以通过环境变量更换模型。
        # 如果没有配置，则默认使用 qwen-plus。
        model=os.getenv("LLM_MODEL_ID", "qwen-plus"),

        # 从环境变量读取 API Key，避免在源代码中硬编码密钥。
        api_key=api_key, # type: ignore

        # 阿里云百炼提供的 OpenAI 兼容接口地址。
        base_url=os.getenv(
            "LLM_BASE_URL",
            "https://dashscope.aliyuncs.com/compatible-mode/v1",
        ),

        # 温度设置为 0，使模型输出相对稳定。
        temperature=0,
    )


def build_agent():
    """
    创建几何计算 Agent。

    create_agent() 会自动管理工具调用循环：
    1. 将用户问题发送给模型；
    2. 让模型选择工具；
    3. 执行工具；
    4. 将工具结果返回给模型；
    5. 输出最终回答。
    """
    return create_agent(
        model=create_model(),

        # Agent 可以使用的工具列表。
        tools=[
            calculate_rectangle_area,
            calculate_circle_area,
        ],

        # 系统提示词用于约束 Agent 的行为。
        # 明确要求调用工具，可以避免模型直接口算。
        system_prompt=(
            "你是一个几何计算助手。"
            "遇到矩形面积或圆形面积问题时，必须调用合适的工具完成计算；"
            "获得工具结果后，请用简洁的中文回答用户。"
        ),
    )


def run_agent() -> None:
    """
    程序入口。
    """
    # 读取项目根目录下的 .env 文件。
    load_dotenv()

    # 创建 Agent。
    agent = build_agent()

    # invoke() 用于运行 Agent。
    # messages 表示发送给 Agent 的对话消息。
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "一个长为 8、宽为 5 的矩形面积是多少？",
                }
            ]
        }
    )

    # create_agent() 会返回完整的消息历史。
    # 最后一条消息就是 Agent 的最终回答。
    print(result["messages"][-1].content)

def run_agent_with_trace(agent, question: str) -> None:
    """
    运行 Agent，并实时打印中间执行轨迹。

    stream_mode="updates" 表示：
    每当 Agent 完成一个步骤，就返回一次状态更新。

    对于一次普通的工具调用，通常会依次看到：
    1. 模型决定调用工具；
    2. 工具完成执行并返回结果；
    3. 模型根据工具结果生成最终回答。
    """
    print("\n========== 用户问题 ==========")
    print(question)

    # 与 invoke() 不同，stream() 不会等到全部步骤结束后才返回结果。
    # 它会在 Agent 运行过程中逐步返回中间状态。
    for chunk in agent.stream( # 每当 Agent 完成一个步骤，就取出一次更新，并将这次更新保存到变量 chunk 中。
        {
            "messages": [  # 向 Agent 发送用户消息
                {
                    "role": "user",
                    "content": question,
                }
            ]
        },

        # updates：每完成一个 Agent 步骤，就返回一次更新。
        stream_mode="updates",

        # v2：使用统一的数据格式。
        version="v2",
    ):
        print("\n========== 新的 chunk ==========")
        print(chunk)  # 打印原始更新内容，便于观察数据结构。
        # 使用 v2 格式时，每个 chunk 都包含 type 和 data 字段。
        if chunk["type"] != "updates":
            continue
        # 在这种格式下，每次返回的 chunk 通常具有类似结构：
        # {
        #     "type": "updates",
        #     "data": {
        #         "model": {
        #             "messages": [...]
        #         },
        #         "tools": {
        #             "messages": [...]
        #         }
        #     }
        # }

        # 一次更新中可能包含一个或多个节点的执行结果。
        for step_name, step_data in chunk["data"].items():
            print(f"当前节点：{step_name}")
            latest_message = step_data["messages"][-1]

            # AIMessage 表示模型生成的消息。
            if isinstance(latest_message, AIMessage):
                # 如果 tool_calls 不为空，说明模型决定调用工具。
                if latest_message.tool_calls:
                    print("\n========== 模型决策 ==========")

                    for tool_call in latest_message.tool_calls:
                        print(f"调用工具：{tool_call['name']}")
                        print(f"工具参数：{tool_call['args']}")

                # 如果没有工具调用但存在文本，则通常是最终回答。
                elif latest_message.content:
                    print("\n========== 最终回答 ==========")
                    print(latest_message.content)

            # ToolMessage 表示工具执行完成后返回给模型的结果。
            elif isinstance(latest_message, ToolMessage):
                print("\n========== 工具执行结果 ==========")
                print(f"工具名称：{latest_message.name}")
                print(f"返回内容：{latest_message.content}")


if __name__ == "__main__":
    run_agent()
    print("\n\n========== 带执行轨迹的演示 ==========")
    run_agent_with_trace(build_agent(), "一个半径为 3 的圆形面积是多少？")

PS D:\Microsoft VS Code\vs work\codeworkvs\py\agent\demo> python .\LangChaindemo.py                                     
========== 直接执行的演示 ==========

========== 用户问题 ==========
一个长为 8、宽为 5 的矩形面积是多少？

矩形面积是 40。


========== 带执行轨迹的演示 ==========

========== 用户问题 ==========
一个半径为 3 的圆形面积是多少？

========== 新的 chunk ==========
{'type': 'updates', 'ns': (), 'data': {'model': {'messages': [AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 19, 'prompt_tokens': 255, 'total_tokens': 274, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5.5-2026-04-24', 'system_fingerprint': None, 'id': 'chatcmpl-DoVOPq0X5MK0cWD6Lc3fV19vkdXS5', 'service_tier': 'default', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--019ea7aa-d2db-7132-9e56-83ce51045e5e-0', tool_calls=[{'name': 'calculate_circle_area', 'args': {'radius': 3}, 'id': 'call_HeeE0R8S5Zw0hQ3mkYOTM4Bx', 'type': 'tool_call'}], invalid_tool_calls=[], usage_metadata={'input_tokens': 255, 'output_tokens': 19, 'total_tokens': 274, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})]}}}
当前节点：model

========== 模型决策 ==========
调用工具：calculate_circle_area
工具参数：{'radius': 3}

========== 新的 chunk ==========
{'type': 'updates', 'ns': (), 'data': {'tools': {'messages': [ToolMessage(content='圆形面积约为 28.27', name='calculate_circle_area', id='d000d484-1567-4cdd-a21b-ac4d502acc65', tool_call_id='call_HeeE0R8S5Zw0hQ3mkYOTM4Bx')]}}}
当前节点：tools

========== 工具执行结果 ==========
工具名称：calculate_circle_area
返回内容：圆形面积约为 28.27

========== 新的 chunk ==========
{'type': 'updates', 'ns': (), 'data': {'model': {'messages': [AIMessage(content='半径为 3 的圆形面积约为 28.27。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 20, 'prompt_tokens': 294, 'total_tokens': 314, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5.5-2026-04-24', 'system_fingerprint': None, 'id': 'chatcmpl-DoVOQJhMxZIoGT9vRvawXvzvLT47W', 'service_tier': 'default', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019ea7aa-d909-79b1-a909-7b8aa27c2163-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 294, 'output_tokens': 20, 'total_tokens': 314, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})]}}}
当前节点：model

========== 最终回答 ==========
半径为 3 的圆形面积约为 28.27。

import asyncio
import os
import time

from dotenv import load_dotenv
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai_like import OpenAILike


async def query_delivery_status(order_id: str) -> str:
    """
    查询订单的配送状态。

    Args:
        order_id: 需要查询的订单编号。

    Returns:
        订单当前的配送状态。
    """
    print(f"[工具开始执行] 正在查询订单 {order_id} ...")

    # 使用 asyncio.sleep() 模拟调用远程物流 API。
    # await 表示当前工具暂时进入等待状态。
    # 等待期间，事件循环可以调度其他任务继续执行。
    await asyncio.sleep(2)

    print(f"[工具执行完成] 已获得订单 {order_id} 的查询结果")

    return f"订单 {order_id} 已到达配送站，预计今天送达。"


def create_model() -> OpenAILike:
    """
    创建大语言模型客户端。

    OpenAILike 是 LlamaIndex 提供的兼容层，
    用于接入支持 OpenAI API 协议的第三方模型服务。
    """
    api_key = os.getenv("LLM_API_KEY")

    # 如果没有配置 API Key，则主动抛出异常，
    # 避免程序在调用模型时才出现不容易定位的错误。
    if not api_key:
        raise RuntimeError("请先在 .env 文件中设置 API_KEY")

    return OpenAILike(
        # 可以通过环境变量更换模型。
        # 如果没有配置，则默认使用 qwen-plus。
        model=os.getenv("LLM_MODEL_ID", "qwen-plus"),

        # 从环境变量读取 API Key，避免在源代码中硬编码密钥。
        api_key=api_key, # type: ignore

        # 阿里云百炼提供的 OpenAI 兼容接口地址。
        api_base=os.getenv(
            "LLM_BASE_URL",
            "https://dashscope.aliyuncs.com/compatible-mode/v1",
        ),

        # 表示当前模型使用 Chat Completions 接口，
        # 而不是传统的文本补全接口。
        is_chat_model=True,

        # 表示模型支持 Function Calling。
        # FunctionAgent 需要模型能够根据问题选择并调用工具。
        is_function_calling_model=True,

        # 上下文窗口大小。
        context_window=int(
            os.getenv("MODEL_CONTEXT_WINDOW", "32768")
        ),

        # 温度设置为 0，使模型输出相对稳定。
        temperature=0,
    )


def build_agent() -> FunctionAgent:
    """
    创建 FunctionAgent。
    创建订单查询 Agent。

    LlamaIndex 可以直接读取 Python 函数的名称、
    参数类型和文档字符串，并将其转换为工具描述。
    """
    # 将异步 Python 函数转换为 Agent 可以调用的工具。
    # FunctionTool 是对现有 Python 函数的简单包装，并且同步函数和异步函数都支持。
    # 在异步工具场景中，显式包装比直接放入tool中注册更清楚。
    delivery_tool = FunctionTool.from_defaults(
        async_fn=query_delivery_status,
    )

    return FunctionAgent(
        llm=create_model(),

        # 为 Agent 注册异步工具。
        tools=[
            delivery_tool,
        ],

        # 明确要求 Agent 使用工具查询订单状态，
        # 避免模型直接编造答案。
        system_prompt=(
            "你是一个订单物流查询助手。"
            "当用户询问订单配送状态时，必须调用 query_delivery_status 工具。"
            "获得工具结果后，请用简洁的中文回答用户。"
        ),
    )

async def ask_agent(order_id: str) -> str:
    """
    创建一个 Agent，并查询指定订单。

    每一个订单查询任务使用一个独立 Agent，
    避免多个用户请求共享同一份对话状态。
    """
    agent = build_agent()

    response = await agent.run(
        user_msg=f"请查询订单 {order_id} 的配送状态。"
    )

    return str(response)


async def run_sequentially() -> None:
    """
    顺序执行两个订单查询任务。

    第二个任务必须等待第一个任务结束后才能开始。
    """
    print("\n========== 顺序执行 ==========")

    start_time = time.perf_counter()

    result_a = await ask_agent("A100")
    result_b = await ask_agent("B200")

    elapsed_time = time.perf_counter() - start_time

    print("\n订单 A100 的回答：")
    print(result_a)

    print("\n订单 B200 的回答：")
    print(result_b)

    print(f"\n顺序执行总耗时：{elapsed_time:.2f} 秒")


async def run_concurrently() -> None:
    """
    并发执行两个订单查询任务。

    asyncio.gather() 会同时启动两个任务。
    当一个任务等待远程响应时，另一个任务可以继续运行。
    """
    print("\n========== 并发执行 ==========")

    start_time = time.perf_counter()

    result_a, result_b = await asyncio.gather(
        ask_agent("A100"),
        ask_agent("B200"),
    )

    elapsed_time = time.perf_counter() - start_time

    print("\n订单 A100 的回答：")
    print(result_a)

    print("\n订单 B200 的回答：")
    print(result_b)

    print(f"\n并发执行总耗时：{elapsed_time:.2f} 秒")


async def main() -> None:
    """
    异步程序入口。

    先顺序执行两个订单查询任务，
    再并发执行相同的两个任务，
    通过运行时间对比观察异步并发的作用。
    """
    load_dotenv()

    await run_sequentially()
    await run_concurrently()


if __name__ == "__main__":
    # 创建并启动事件循环。
    asyncio.run(main())

========== 顺序执行 ==========
[工具开始执行] 正在查询订单 A100 ...
[工具执行完成] 已获得订单 A100 的查询结果
[工具开始执行] 正在查询订单 B200 ...
[工具执行完成] 已获得订单 B200 的查询结果

订单 A100 的回答：
订单 A100 已到达配送站，预计今天送达。

订单 B200 的回答：
订单 B200 已到达配送站，预计今天送达。

顺序执行总耗时：13.33 秒

========== 并发执行 ==========
[工具开始执行] 正在查询订单 A100 ...
[工具开始执行] 正在查询订单 B200 ...
[工具执行完成] 已获得订单 A100 的查询结果
[工具执行完成] 已获得订单 B200 的查询结果

订单 A100 的回答：
订单 A100 已到达配送站，预计今天送达。

订单 B200 的回答：
订单 B200 已到达配送站，预计今天送达。

并发执行总耗时：7.32 秒

import os

from dotenv import load_dotenv
from llama_index.core import Document, Settings, VectorStoreIndex
from llama_index.embeddings.openai_like import OpenAILikeEmbedding
from llama_index.llms.openai_like import OpenAILike


def configure_models() -> None:
    """
    配置大语言模型和 Embedding 模型。

    RAG 中需要使用两个模型：

    1. Embedding 模型：
       将知识库文本和用户问题转换为向量，
       用于计算文本之间的语义相似度。

    2. 大语言模型：
       读取检索到的知识库内容，
       并生成最终回答。
    """
    api_key = os.getenv("DASHSCOPE_API_KEY")

    if not api_key:
        raise RuntimeError("请先在 .env 文件中设置 DASHSCOPE_API_KEY")

    api_base = os.getenv(
        "OPENAI_BASE_URL",
        "https://dashscope.aliyuncs.com/compatible-mode/v1",
    )

    # 配置用于生成回答的大语言模型。
    Settings.llm = OpenAILike(
        model=os.getenv("MODEL_NAME", "qwen-plus"),
        api_base=api_base,
        api_key=api_key,

        # 表示当前模型使用 Chat Completions 接口。
        is_chat_model=True,

        # RAG 查询不需要 Function Calling，
        # 因此不需要设置 is_function_calling_model。
        context_window=int(
            os.getenv("MODEL_CONTEXT_WINDOW", "32768")
        ),

        # 温度设置为 0，使回答相对稳定。
        temperature=0,
    )

    # 配置用于向量检索的 Embedding 模型。
    Settings.embed_model = OpenAILikeEmbedding(
        model_name=os.getenv(
            "EMBEDDING_MODEL_NAME",
            "text-embedding-v4",
        ),
        api_base=api_base,
        api_key=api_key,

        # text-embedding-v4 默认支持 1024 维向量。
        dimensions=int(
            os.getenv("EMBEDDING_DIMENSIONS", "1024")
        ),
    )


def create_knowledge_base() -> list[Document]:
    """
    创建一个很小的本地知识库。

    在实际项目中，这些内容通常来自：
    - 本地 TXT 文件；
    - PDF 文档；
    - Word 文档；
    - 数据库；
    - 企业内部知识库。

    为了让示例尽量简单，这里直接在代码中创建文档。
    """
    return [
        Document(
            text=(
                "RAG 是 Retrieval-Augmented Generation 的缩写，"
                "中文通常翻译为检索增强生成。"
                "它的基本流程是：先根据用户问题检索相关文档，"
                "再将检索结果作为上下文交给大语言模型生成回答。"
            )
        ),
        Document(
            text=(
                "RAG 和模型微调不是同一种技术。"
                "RAG 主要通过外部知识库为模型补充信息，"
                "通常不需要修改模型参数。"
                "模型微调则会使用训练数据更新模型参数。"
            )
        ),
        Document(
            text=(
                "DPR 是 Dense Passage Retrieval 的缩写。"
                "它是一种稠密检索方法，"
                "通常使用两个编码器分别将问题和文档转换为向量。"
            )
        ),
        Document(
            text=(
                "FAISS 是一个高效的向量检索库。"
                "它可以在大量向量中快速查找与查询向量最相似的内容。"
            )
        ),
    ]


def build_query_engine():
    """
    创建 RAG 查询引擎。

    VectorStoreIndex.from_documents() 会执行以下工作：

    1. 读取文档；
    2. 将文档切分为较小的文本块；
    3. 使用 Embedding 模型将文本块转换为向量；
    4. 建立向量索引。

    本示例未连接外部向量数据库，
    因此向量索引默认存储在内存中。
    程序退出后，索引会消失。
    """
    documents = create_knowledge_base()

    index = VectorStoreIndex.from_documents(
        documents=documents,

        # 在终端显示构建索引的进度。
        show_progress=True,
    )

    # 将向量索引转换为查询引擎。
    #
    # similarity_top_k=2 表示：
    # 每次收到问题后，检索相似度最高的两个文本块，
    # 再将它们提供给大语言模型。
    return index.as_query_engine(
        similarity_top_k=2,
    )


def print_retrieved_nodes(response) -> None:
    """
    打印本次查询检索到的文本块。

    这一步不是 RAG 必需的，
    但有助于理解检索过程和排查问题。
    """
    print("\n========== 检索到的知识库内容 ==========")

    for index, source_node in enumerate(
        response.source_nodes,
        start=1,
    ):
        print(f"\n--- 文本块 {index} ---")

        # score 表示该文本块与问题的语义相似度。
        # 数值通常越高，说明文本块与问题越相关。
        print(f"相似度分数：{source_node.score}")

        # node.text 表示文本块的实际内容。
        print(f"文本内容：{source_node.node.text}")


def main() -> None:
    """
    程序入口。
    """
    # 加载 .env 文件中的环境变量。
    load_dotenv()

    # 配置大模型和 Embedding 模型。
    configure_models()

    # 构建向量索引，并创建查询引擎。
    query_engine = build_query_engine()

    # 用户问题。
    question = "RAG 和模型微调有什么区别？"

    print("\n========== 用户问题 ==========")
    print(question)

    # query() 会执行完整的 RAG 流程：
    #
    # 1. 将问题转换为向量；
    # 2. 在向量索引中检索相关文本；
    # 3. 将检索结果和问题发送给大语言模型；
    # 4. 返回模型生成的答案。
    response = query_engine.query(question)

    # 输出检索到的文本块。
    print_retrieved_nodes(response)

    # 输出大语言模型根据检索结果生成的答案。
    print("\n========== 最终回答 ==========")
    print(str(response))


if __name__ == "__main__":
    main()

​

import asyncio
import os
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.agents import UserProxyAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.messages import Console


def create_openai_model_client():
    """创建并配置 OpenAI 模型客户端"""
    return OpenAIChatCompletionClient(
        model=os.getenv("LLM_MODEL_ID", "gpt-4o"),
        api_key=os.getenv("LLM_API_KEY"),
        base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1")
    )


def create_product_manager(model_client):
    """创建产品经理智能体"""
    system_message = """你是一位经验丰富的产品经理，专门负责软件产品的需求分析和项目规划。

你的核心职责包括：
1. **需求分析**：深入理解用户需求，识别核心功能和边界条件
2. **技术规划**：基于需求制定清晰的技术实现路径
3. **风险评估**：识别潜在的技术风险和用户体验问题
4. **协调沟通**：与工程师和其他团队成员进行有效沟通

当接到开发任务时，请按以下结构进行分析：
1. 需求理解与分析
2. 功能模块划分
3. 技术选型建议
4. 实现优先级排序
5. 验收标准定义

请简洁明了地回应，并在分析完成后说"请工程师开始实现"。"""

    return AssistantAgent(
        name="ProductManager",
        model_client=model_client,
        system_message=system_message,
    )


def create_engineer(model_client):
    """创建软件工程师智能体"""
    system_message = """你是一位资深的软件工程师，擅长 Python 开发和 Web 应用构建。

你的技术专长包括：
1. **Python 编程**：熟练掌握 Python 语法和最佳实践
2. **Web 开发**：精通 Streamlit、Flask、Django 等框架
3. **API 集成**：有丰富的第三方 API 集成经验
4. **错误处理**：注重代码的健壮性和异常处理

当收到开发任务时，请：
1. 仔细分析技术需求
2. 选择合适的技术方案
3. 编写完整的代码实现
4. 添加必要的注释和说明
5. 考虑边界情况和异常处理

请提供完整的可运行代码，并在完成后说"请代码审查员检查"。"""

    return AssistantAgent(
        name="Engineer",
        model_client=model_client,
        system_message=system_message,
    )


def create_code_reviewer(model_client):
    """创建代码审查员智能体"""
    system_message = """你是一位经验丰富的代码审查专家，专注于代码质量和最佳实践。

你的审查重点包括：
1. **代码质量**：检查代码的可读性、可维护性和性能
2. **安全性**：识别潜在的安全漏洞和风险点
3. **最佳实践**：确保代码遵循行业标准和最佳实践
4. **错误处理**：验证异常处理的完整性和合理性

审查流程：
1. 仔细阅读和理解代码逻辑
2. 检查代码规范和最佳实践
3. 识别潜在问题和改进点
4. 提供具体的修改建议
5. 评估代码的整体质量

请提供具体的审查意见，完成后说"代码审查完成，请用户代理测试"。"""

    return AssistantAgent(
        name="CodeReviewer",
        model_client=model_client,
        system_message=system_message,
    )


def create_user_proxy():
    """创建用户代理智能体"""
    
    return UserProxyAgent(
        name="UserProxy",
        description="""用户代理，负责以下职责：
1. 代表用户提出开发需求
2. 执行最终的代码实现
3. 验证功能是否符合预期
4. 提供用户反馈和建议

完成测试后请回复 TERMINATE。""",
    )


async def run_software_development_team():
    """运行软件开发团队协作"""
    # 初始化模型客户端
    model_client = create_openai_model_client()
    
    # 创建所有智能体
    product_manager = create_product_manager(model_client)
    engineer = create_engineer(model_client)
    code_reviewer = create_code_reviewer(model_client)
    user_proxy = create_user_proxy()
    
    # 定义团队聊天和协作规则
    team_chat = RoundRobinGroupChat(
        participants=[product_manager, engineer, code_reviewer, user_proxy],
        termination_condition=TextMentionTermination("TERMINATE"),
        max_turns=20,
    )
    
    # 定义任务描述
    task = """我们需要开发一个比特币价格显示应用，具体要求如下：
核心功能：
- 实时显示比特币当前价格（USD）
- 显示24小时价格变化趋势（涨跌幅和涨跌额）
- 提供价格刷新功能

技术要求：
- 使用 Streamlit 框架创建 Web 应用
- 界面简洁美观，用户友好
- 添加适当的错误处理和加载状态

请团队协作完成这个任务，从需求分析到最终实现。"""
    
    # 异步执行团队协作，并流式输出对话过程
    result = await Console(team_chat.run_stream(task=task)) # 异步等待单个任务，并非并行执行
    return result


# 主程序入口
if __name__ == "__main__":
    result = asyncio.run(run_software_development_team())

🔧 正在初始化模型客户端...
👥 正在创建智能体团队...
🚀 启动 AutoGen 软件开发团队协作...
============================================================
---------- TextMessage (user) ----------
我们需要开发一个比特币价格显示应用，具体要求如下：
...
请团队协作完成这个任务，从需求分析到最终实现。
---------- TextMessage (ProductManager) ----------
### 1. 需求理解与分析
...
请工程师开始实现。
---------- TextMessage (Engineer) ----------
### 技术方案实施
...
请代码审查员检查。
---------- TextMessage (CodeReviewer) ----------
### 代码审查
...
代码审查完成，请用户代理测试。    
---------- TextMessage (UserProxy) ----------
已经完成需求
---------- TextMessage (ProductManager) ----------
太好了，感谢您的反馈！如果在使用过程中有任何问题，或者有其他功能需求和改进建议，请随时告知我们。我们会持续提供支持和改进。期待您对我们的应用
有愉快的使用体验！
---------- TextMessage (Engineer) ----------
很高兴听到项目顺利完成。如果您或用户有任何问题或者需要帮助，请随时联系我们。感谢您对我们工作的支持，让我们一起确保应用稳定运行并不断优化用户
体验！
---------- TextMessage (CodeReviewer) ----------
非常感谢大家的努力与协作，使得项目能够顺利完成。未来若有更多技术支持的需求或者需要改进的地方，我们愿意为项目的持续优化贡
献力量。期待用户能够享受到流畅的体验，同时也欢迎提出更多的反馈与建议。再次感谢团队的合作！
---------- TextMessage (UserProxy) ----------
Enter your response: TERMINATE
============================================================
✅ 团队协作完成！

📋 协作结果摘要：
- 参与智能体数量：4个
- 任务完成状态：成功

from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
    model_info={
        "function_calling": True,
        "max_tokens": 4096,
        "context_length": 32768,
        "vision": False,
        "json_output": True,
        "family": "deepseek",
        "structured_output": True,
    }
)

​

from agentscope.message import Msg

# 消息的标准结构
message = Msg(
    name="Alice",           # 发送者名称
    content="Hello, Bob!",  # 消息内容
    role="user",           # 角色类型
    metadata={             # 元数据信息
        "timestamp": "2024-01-15T10:30:00Z",
        "message_type": "text",
        "priority": "normal"
    }
)

from agentscope.agents import AgentBase

class CustomAgent(AgentBase):
    def __init__(self, name: str, **kwargs):
        super().__init__(name=name, **kwargs)
        # 智能体初始化逻辑
    
    def reply(self, x: Msg) -> Msg:
        # 智能体的核心响应逻辑
        response = self.model(x.content)
        return Msg(name=self.name, content=response, role="assistant")
    
    def observe(self, x: Msg) -> None:
        # 智能体的观察逻辑（可选）
        self.memory.add(x)

async def werewolf_phase(self, round_num: int):
    """狼人阶段 - 展示消息驱动的协作模式"""
    if not self.werewolves:
        return None
        
    # 通过消息中心建立狼人专属通信频道
    async with MsgHub(
        self.werewolves,
        enable_auto_broadcast=True,
        announcement=await self.moderator.announce(
            f"狼人们，请讨论今晚的击杀目标。存活玩家：{format_player_list(self.alive_players)}"
        ),
    ) as werewolves_hub:
        # 讨论阶段：狼人通过消息交换策略
        for _ in range(MAX_DISCUSSION_ROUND):
            for wolf in self.werewolves:
                await wolf(structured_model=DiscussionModelCN)
        
        # 投票阶段：收集并统计狼人的击杀决策
        werewolves_hub.set_auto_broadcast(False)
        kill_votes = await fanout_pipeline(
            self.werewolves,
            msg=await self.moderator.announce("请选择击杀目标"),
            structured_model=WerewolfKillModelCN,
            enable_gather=False,
        )

class DiscussionModelCN(BaseModel):
    """讨论阶段的输出格式"""
    reach_agreement: bool = Field(
        description="是否已达成一致意见",
        default=False
    )
    confidence_level: int = Field(
        description="对当前推理的信心程度(1-10)",
        ge=1, le=10,
        default=5
    )
    key_evidence: Optional[str] = Field(
        description="支持你观点的关键证据",
        default=None
    )

class WitchActionModelCN(BaseModel):
    """女巫行动的输出格式"""
    use_antidote: bool = Field(description="是否使用解药")
    use_poison: bool = Field(description="是否使用毒药")
    target_name: Optional[str] = Field(description="毒药目标玩家姓名")

def get_role_prompt(role: str, character: str) -> str:
    """获取角色提示词 - 融合游戏规则与人物性格"""
    base_prompt = f"""你是{character}，在这场三国狼人杀游戏中扮演{role}。

重要规则：
1. 你只能通过对话和推理参与游戏
2. 不要尝试调用任何外部工具或函数
3. 严格按照要求的JSON格式回复

角色特点：
"""
    
    if role == "狼人":
        return base_prompt + f"""
- 你是狼人阵营，目标是消灭所有好人
- 夜晚可以与其他狼人协商击杀目标
- 白天要隐藏身份，误导好人
- 以{character}的性格说话和行动
"""

# 并行收集所有玩家的投票决策
vote_msgs = await fanout_pipeline(
    self.alive_players,
    await self.moderator.announce("请投票选择要淘汰的玩家"),
    structured_model=get_vote_model_cn(self.alive_players),
    enable_gather=False,
)

try:
    response = await wolf(
        "请分析当前局势并表达你的观点。",
        structured_model=DiscussionModelCN
    )
except Exception as e:
    print(f"⚠️ {wolf.name} 讨论时出错: {e}")
    # 创建默认响应，确保游戏继续进行
    default_response = DiscussionModelCN(
        reach_agreement=False,
        confidence_level=5,
        key_evidence="暂时无法分析"
    )

​

from colorama import Fore
from camel.societies import RolePlaying
from camel.utils import print_text_animated
from camel.models import ModelFactory
from camel.types import ModelPlatformType
from dotenv import load_dotenv
import os

load_dotenv()
LLM_API_KEY = os.getenv("LLM_API_KEY")
LLM_BASE_URL = os.getenv("LLM_BASE_URL")
LLM_MODEL = os.getenv("LLM_MODEL")

#创建模型,在这里以Qwen为例,调用的百炼大模型平台API
model = ModelFactory.create(
    model_platform=ModelPlatformType.QWEN,
    model_type=LLM_MODEL,
    url=LLM_BASE_URL,
    api_key=LLM_API_KEY
)

# 定义协作任务
task_prompt = """
创作一本关于"拖延症心理学"的短篇电子书，目标读者是对心理学感兴趣的普通大众。
要求：
1. 内容科学严谨，基于实证研究
2. 语言通俗易懂，避免过多专业术语
3. 包含实用的改善建议和案例分析
4. 篇幅控制在8000-10000字
5. 结构清晰，包含引言、核心章节和总结
"""

print(Fore.YELLOW + f"协作任务:\n{task_prompt}\n")

# 初始化角色扮演会话
# AI 作家作为 "user"，负责提出写作结构和要求
# AI 心理学家作为 "assistant"，负责提供专业知识和内容
role_play_session = RolePlaying(
    assistant_role_name="心理学家",
    user_role_name="作家",
    task_prompt=task_prompt,
    model=model,
    with_task_specify=False, # 在本例中，我们直接使用给定的task_prompt
)

print(Fore.CYAN + f"具体任务描述:\n{role_play_session.task_prompt}\n")

# 开始协作对话
chat_turn_limit, n = 30, 0
# 调用 init_chat() 来获得由 AI 生成的初始对话消息
input_msg = role_play_session.init_chat()

while n < chat_turn_limit:
    n += 1
    # step() 方法驱动一轮完整的对话，AI 用户和 AI 助理各发言一次
    assistant_response, user_response = role_play_session.step(input_msg)
    
    # 检查是否有消息返回，防止对话提前终止
    if assistant_response.msg is None or user_response.msg is None:
        break
    
    print_text_animated(Fore.BLUE + f"作家 (AI User):\n\n{user_response.msg.content}\n")
    print_text_animated(Fore.GREEN + f"心理学家 (AI Assistant):\n\n{assistant_response.msg.content}\n")
    
    # 检查任务完成标志
    if "<CAMEL_TASK_DONE>" in user_response.msg.content or "<CAMEL_TASK_DONE>" in assistant_response.msg.content:
        print(Fore.MAGENTA + "✅ 电子书创作完成！")
        break
    
    # 将助理的回复作为下一轮对话的输入
    input_msg = assistant_response.msg

print(Fore.YELLOW + f"总共进行了 {n} 轮协作对话")

from langgraph.graph import StateGraph, END

# 初始化一个状态图，并绑定我们定义的状态结构
workflow = StateGraph(AgentState)

# 将节点函数添加到图中
workflow.add_node("planner", planner_node)
workflow.add_node("executor", executor_node)

# 设置图的入口点
workflow.set_entry_point("planner")

# 添加常规边，连接 planner 和 executor
workflow.add_edge("planner", "executor")

# 添加条件边，实现动态路由
workflow.add_conditional_edges(
    # 起始节点
    "executor",
    # 判断函数
    should_continue,
    # 路由映射：将判断函数的返回值映射到目标节点
    {
        "continue_to_planner": "planner", # 如果返回"continue_to_planner"，则跳回planner节点
        "end_workflow": END               # 如果返回"end_workflow"，则结束流程
    }
)

# 编译图，生成可执行的应用
app = workflow.compile()

# 运行图
inputs = {"current_task": "分析最近的AI行业新闻", "messages": []}
for event in app.stream(inputs):
    print(event)

"""
智能搜索助手 - 基于 LangGraph + Tavily API 的真实搜索系统
1. 理解用户需求
2. 使用Tavily API真实搜索信息  
3. 生成基于搜索结果的回答
"""

import asyncio
from typing import TypedDict, Annotated
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
import os
from dotenv import load_dotenv
from tavily import TavilyClient

# 加载环境变量
load_dotenv()

# 定义状态结构
class SearchState(TypedDict):
    messages: Annotated[list, add_messages]
    user_query: str        # 用户查询
    search_query: str      # 优化后的搜索查询
    search_results: str    # Tavily搜索结果
    final_answer: str      # 最终答案
    step: str              # 当前步骤

# 初始化模型和Tavily客户端
llm = ChatOpenAI(
    model=os.getenv("LLM_MODEL_ID", "gpt-4o-mini"),
    api_key=os.getenv("LLM_API_KEY"), # type: ignore
    base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"),
    temperature=0.7
)

# 初始化Tavily客户端
tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

def understand_query_node(state: SearchState) -> SearchState:
    """步骤1：理解用户查询并生成搜索关键词"""
    
    # 获取最新的用户消息
    user_message = ""
    for msg in reversed(state["messages"]):
        if isinstance(msg, HumanMessage):
            user_message = msg.content
            break
    
    understand_prompt = f"""分析用户的查询："{user_message}"

        请完成两个任务：
        1. 简洁总结用户想要了解什么
        2. 生成最适合搜索的关键词（中英文均可，要精准）

        格式：
        理解：[用户需求总结]
        搜索词：[最佳搜索关键词]
    """

    response = llm.invoke([SystemMessage(content=understand_prompt)]) # invoke方法同LangChain，会返回一个AIMessage对象，包含模型的回复内容
    
    # 提取搜索关键词
    response_text = response.content
    search_query = user_message  # 默认使用原始查询
    
    if "搜索词：" in response_text:
        search_query = response_text.split("搜索词：")[1].strip()
    elif "搜索关键词：" in response_text:
        search_query = response_text.split("搜索关键词：")[1].strip()
    
    return {
        "user_query": response.content,
        "search_query": search_query,
        "step": "understood",
        "messages": [AIMessage(content=f"我理解您的需求：{response.content}")]
    }

def tavily_search_node(state: SearchState) -> SearchState:
    """步骤2：使用Tavily API进行真实搜索"""
    
    search_query = state["search_query"]
    
    try:
        print(f"🔍 正在搜索: {search_query}")
        
        # 调用Tavily搜索API
        response = tavily_client.search(
            query=search_query,
            search_depth="basic",
            include_answer=True,
            include_raw_content=False,
            max_results=5
        )
        
        # 处理搜索结果
        search_results = ""
        
        # 优先使用Tavily的综合答案
        if response.get("answer"):
            search_results = f"综合答案：\n{response['answer']}\n\n"
        
        # 添加具体的搜索结果
        if response.get("results"):
            search_results += "相关信息：\n"
            for i, result in enumerate(response["results"][:3], 1):
                title = result.get("title", "")
                content = result.get("content", "")
                url = result.get("url", "")
                search_results += f"{i}. {title}\n{content}\n来源：{url}\n\n"
        
        if not search_results:
            search_results = "抱歉，没有找到相关信息。"
        
        return {
            "search_results": search_results,
            "step": "searched",
            "messages": [AIMessage(content=f"✅ 搜索完成！找到了相关信息，正在为您整理答案...")]
        }
        
    except Exception as e:
        error_msg = f"搜索时发生错误: {str(e)}"
        print(f"❌ {error_msg}")
        
        return {
            "search_results": f"搜索失败：{error_msg}",
            "step": "search_failed",
            "messages": [AIMessage(content="❌ 搜索遇到问题，我将基于已有知识为您回答")]
        }

def generate_answer_node(state: SearchState) -> SearchState:
    """步骤3：基于搜索结果生成最终答案"""
    
    # 检查是否有搜索结果
    if state["step"] == "search_failed":
        # 如果搜索失败，基于LLM知识回答
        fallback_prompt = f"""搜索API暂时不可用，请基于您的知识回答用户的问题：

            用户问题：{state['user_query']}

            请提供一个有用的回答，并说明这是基于已有知识的回答。
        """
        
        response = llm.invoke([SystemMessage(content=fallback_prompt)])
        
        return {
            "final_answer": response.content,
            "step": "completed",
            "messages": [AIMessage(content=response.content)]
        }
    
    # 基于搜索结果生成答案
    answer_prompt = f"""基于以下搜索结果为用户提供完整、准确的答案：

        用户问题：{state['user_query']}

        搜索结果：
        {state['search_results']}

        请要求：
        1. 综合搜索结果，提供准确、有用的回答
        2. 如果是技术问题，提供具体的解决方案或代码
        3. 引用重要信息的来源
        4. 回答要结构清晰、易于理解
        5. 如果搜索结果不够完整，请说明并提供补充建议
    """

    response = llm.invoke([SystemMessage(content=answer_prompt)])
    
    return {
        "final_answer": response.content,
        "step": "completed",
        "messages": [AIMessage(content=response.content)]
    }

# 构建搜索工作流
def create_search_assistant():
    workflow = StateGraph(SearchState)
    
    # 添加三个节点
    workflow.add_node("understand", understand_query_node)
    workflow.add_node("search", tavily_search_node)
    workflow.add_node("answer", generate_answer_node)
    
    # 设置线性流程
    workflow.add_edge(START, "understand")
    workflow.add_edge("understand", "search")
    workflow.add_edge("search", "answer")
    workflow.add_edge("answer", END)
    
    # 编译图
    memory = InMemorySaver()
    app = workflow.compile(checkpointer=memory)
    
    return app

async def main():
    """主函数：运行智能搜索助手"""
    
    # 检查API密钥
    if not os.getenv("TAVILY_API_KEY"):
        print("❌ 错误：请在.env文件中配置TAVILY_API_KEY")
        return
    
    app = create_search_assistant()
    
    print("🔍 智能搜索助手启动！")
    print("我会使用Tavily API为您搜索最新、最准确的信息")
    print("支持各种问题：新闻、技术、知识问答等")
    print("(输入 'quit' 退出)\n")
    
    session_count = 0
    
    while True:
        user_input = input("🤔 您想了解什么: ").strip()
        
        if user_input.lower() in ['quit', 'q', '退出', 'exit']:
            print("感谢使用！再见！👋")
            break
        
        if not user_input:
            continue
        
        session_count += 1
        config = {"configurable": {"thread_id": f"search-session-{session_count}"}}
        
        # 初始状态
        initial_state = {
            "messages": [HumanMessage(content=user_input)],
            "user_query": "",
            "search_query": "",
            "search_results": "",
            "final_answer": "",
            "step": "start"
        }
        
        try:
            print("\n" + "="*60)
            
            # 执行工作流
            async for output in app.astream(initial_state, config=config):
                for node_name, node_output in output.items():
                    if "messages" in node_output and node_output["messages"]:
                        latest_message = node_output["messages"][-1]
                        if isinstance(latest_message, AIMessage):
                            if node_name == "understand":
                                print(f"🧠 理解阶段: {latest_message.content}")
                            elif node_name == "search":
                                print(f"🔍 搜索阶段: {latest_message.content}")
                            elif node_name == "answer":
                                print(f"\n💡 最终回答:\n{latest_message.content}")
            
            print("\n" + "="*60 + "\n")
        
        except Exception as e:
            print(f"❌ 发生错误: {e}")
            print("请重新输入您的问题。\n")

if __name__ == "__main__":
    asyncio.run(main())

PS D:\Microsoft VS Code\vs work\codeworkvs\py\myother\agent\project> python .\demo.py
🔍 智能搜索助手启动！
我会使用Tavily API为您搜索最新、最准确的信息
支持各种问题：新闻、技术、知识问答等
(输入 'quit' 退出)

🤔 您想了解什么: 下周我要去青岛，天气怎么样？有合适的景点吗？

============================================================
🧠 理解阶段: 我理解您的需求：理解：用户计划下周去青岛旅行，需要了解当地天气情况以及推荐合适的旅游景点。
搜索词：青岛下周天气预报；青岛旅游景点推荐
🔍 正在搜索: 青岛下周天气预报；青岛旅游景点推荐
🔍 搜索阶段: ✅ 搜索完成！找到了相关信息，正在为您整理答案...

💡 最终回答:
根据您“下周去青岛旅行”的计划，我为您综合整理了当地的天气预报和旅游景点推荐，以帮助您更好地准备行程。

### **一、天气与着装建议**
- **天气预报摘要**：根据搜索结果，青岛**下周预计为多云天气，气温在20-24摄氏度之间**，气候总体温和舒适。
- **着装建议**：鉴于此温度范围和青岛秋季（9-11月）气候宜人、偶有微风的特点，**建议您穿着舒适的长袖衬衫、薄外套或风衣**。由于沿海城市可能湿度较高且早晚温差存在，携带一件方便穿脱的外套是明智之选。
- **气候参考**：青岛秋季（9-11月）平均气温在17.2-23.6°C，雨量减少，阳光明媚，非常适合户外活动。

**请注意**：以上天气预报综合自搜索结果中的信息。为确保准确，请您在出发前通过**中国天气网（weather.com.cn）** 或手机天气应用查询最新的**青岛未来7天具体天气预报**。

### **二、旅游景点推荐**
青岛作为“海滨明珠”，集山、海、城、湾于一体。以下为您分类推荐必游景点：

#### **1. 经典必游（文化与地标）**
- **栈桥**：青岛的象征和标志，是感受城市历史与海景的绝佳起点。
- **青岛啤酒博物馆**：深入了解享誉世界的青岛啤酒历史与酿造工艺，并可品尝新鲜啤酒，是青岛独特的文化体验。

#### **2. 海滨风光（休闲与漫步）**
- **石老人海水浴场**：著名的海滩，沙质细腻，适合踏浪、散步和观海。
- **小青岛公园 & 鲁迅公园**：优美的海滨公园，适合漫步、拍照，欣赏海岸线风光。
- **八大关风景区**：以各国风格别墅闻名的街区，绿树成荫，尤其秋季景色极佳。

#### **3. 文化与历史体验**
- **历史德式建筑群**：漫步老城区，观赏保存完好的德式建筑，感受独特的欧陆风情。
- **天主教堂（圣弥厄尔大教堂）**：典型的哥特式建筑，是拍照和了解宗教文化的好去处。

#### **4. 自然与风光**
- **崂山风景区**：道教名山，拥有壮丽的山海风光和道教庙宇，适合半日或一日游，是登山爱好者的首选。

### **三、补充建议与节庆活动**
- **最佳旅行季节**：资料显示，**8-11月**是青岛的最佳旅行时间。秋季气候宜人，海鲜肥美，且能赶上热闹的节庆。
- **特色节庆（如时间契合）**：
    - **青岛国际啤酒节**：通常在8月第二个周末开始，为期两周，是盛大的狂欢派对。
    - **秋季海鲜季**：此时各类海鲜上市，是品尝美食的好时机。
- **实用贴士**：建议穿着舒适的鞋子以便步行。市内酒店、餐厅普遍有Wi-Fi覆盖，出行交通便利。

**总结**：下周您前往青岛旅行，天气以多云温和为主，适合户外活动。请务必带一件薄外套。行程上，**栈桥、青岛啤酒博物馆、石老人海滩和八大关**是经典组合，若时间充裕，**崂山**也值得一游。

（信息综合自KKday青岛旅游指南及天气网相关资料）

"""
智能搜索 Agent - 基于 LangGraph + Tavily API

Agent 可以自行决定：
1. 是否需要调用工具
2. 调用哪个工具
3. 工具参数是什么
4. 是否需要连续调用多个工具
5. 什么时候结束并直接回答用户
"""

import ast
import operator
import os
from datetime import datetime
from zoneinfo import ZoneInfo
from typing import Any

from dotenv import load_dotenv
from tavily import TavilyClient

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver


# ============================================================
# 1. 初始化
# ============================================================

load_dotenv()

llm = ChatOpenAI(
    model=os.getenv("LLM_MODEL_ID", "gpt-4o-mini"),
    api_key=os.getenv("LLM_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"),
    # 工具选择任务通常使用较低温度，使行为更稳定
    temperature=0,
)

tavily_client = TavilyClient(
    api_key=os.getenv("TAVILY_API_KEY")
)


# ============================================================
# 2. 定义工具
# ============================================================

@tool
def tavily_search(query: str) -> str:
    """
    搜索互联网中的最新信息和外部资料。

    适用于：
    - 新闻、热点、最新技术进展
    - 软件版本、产品信息、政策、价格等可能变化的信息
    - 用户明确要求查询、搜索或核实的信息
    - 需要引用外部来源的信息

    不适用于：
    - 简单寒暄
    - 基础算术
    - 可以直接回答的稳定知识

    Args:
        query: 精确、适合搜索引擎使用的搜索关键词。
    """

    response = tavily_client.search(
        query=query,
        search_depth="basic",
        include_answer=True,
        include_raw_content=False,
        max_results=5,
    )

    sections: list[str] = []

    if response.get("answer"):
        sections.append(
            f"搜索摘要：\n{response['answer']}"
        )

    results = response.get("results", [])

    if results:
        formatted_results = ["搜索来源："]

        for index, result in enumerate(results, start=1):
            title = result.get("title", "无标题")
            content = result.get("content", "")
            url = result.get("url", "")

            formatted_results.append(
                f"{index}. {title}\n"
                f"{content}\n"
                f"URL: {url}"
            )

        sections.append("\n\n".join(formatted_results))

    if not sections:
        return "没有检索到有效结果。"

    return "\n\n".join(sections)


_ALLOWED_BINARY_OPERATORS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.FloorDiv: operator.floordiv,
    ast.Mod: operator.mod,
    ast.Pow: operator.pow,
}

_ALLOWED_UNARY_OPERATORS = {
    ast.UAdd: operator.pos,
    ast.USub: operator.neg,
}


def _evaluate_expression(node: ast.AST) -> int | float:
    """安全地解析基础算术表达式，避免直接使用 eval。"""

    if isinstance(node, ast.Constant):
        if isinstance(node.value, (int, float)):
            return node.value

        raise ValueError("表达式中只能出现数字。")

    if isinstance(node, ast.BinOp):
        operator_type = type(node.op)

        if operator_type not in _ALLOWED_BINARY_OPERATORS:
            raise ValueError("包含不支持的运算符。")

        left = _evaluate_expression(node.left)
        right = _evaluate_expression(node.right)

        # 防止模型生成过大的指数运算
        if isinstance(node.op, ast.Pow) and abs(right) > 20:
            raise ValueError("指数过大。")

        return _ALLOWED_BINARY_OPERATORS[operator_type](left, right)

    if isinstance(node, ast.UnaryOp):
        operator_type = type(node.op)

        if operator_type not in _ALLOWED_UNARY_OPERATORS:
            raise ValueError("包含不支持的一元运算符。")

        value = _evaluate_expression(node.operand)
        return _ALLOWED_UNARY_OPERATORS[operator_type](value)

    raise ValueError("表达式格式不受支持。")


@tool
def calculator(expression: str) -> str:
    """
    计算基础算术表达式。

    适用于：
    - 加减乘除
    - 取模
    - 幂运算
    - 带括号的算术表达式

    Args:
        expression: 算术表达式，例如 "(125 + 75) * 3"。
    """

    try:
        parsed = ast.parse(expression, mode="eval")
        result = _evaluate_expression(parsed.body)
        return f"计算结果：{result}"

    except Exception as exc:
        return f"计算失败：{exc}"


@tool
def get_current_time(timezone_name: str = "Asia/Singapore") -> str:
    """
    获取指定时区的当前日期和时间。

    适用于：
    - 查询当前时间
    - 查询今天的日期
    - 回答涉及“今天”“现在”等相对时间的问题

    Args:
        timezone_name: IANA 时区名称，例如 Asia/Shanghai、Asia/Singapore。
    """

    try:
        current_time = datetime.now(ZoneInfo(timezone_name))

        return (
            f"时区：{timezone_name}\n"
            f"当前时间：{current_time.strftime('%Y-%m-%d %H:%M:%S')}"
        )

    except Exception:
        return (
            "无法识别该时区。请使用 IANA 时区名称，"
            "例如 Asia/Shanghai 或 Asia/Singapore。"
        )


tools = [
    tavily_search,
    calculator,
    get_current_time,
]

# bind_tools 只是将工具结构提供给模型。
# 模型会自行判断是否生成 tool_call。
llm_with_tools = llm.bind_tools(tools)


# ============================================================
# 3. 定义 Agent 节点
# ============================================================

SYSTEM_PROMPT = """
你是一名智能搜索助手。

你可以自行判断是否调用工具。请遵循以下规则：

1. 用户询问新闻、最新进展、软件版本、政策、产品信息、
   实时数据或明确要求核实信息时，优先调用 tavily_search。
2. 用户提出算术问题时，调用 calculator。
3. 用户询问当前日期、时间或涉及时区时，调用 get_current_time。
4. 用户提出基础知识问题、解释性问题或简单寒暄时，
   可以直接回答，不要为了调用工具而调用工具。
5. 一次工具调用不足以回答问题时，可以继续调用工具。
6. 使用 tavily_search 时，请自行将用户问题改写为精确的搜索关键词。
7. 搜索结果包含 URL 时，在最终回答中列出关键来源。
8. 搜索结果不充分时，请明确说明信息不足，不要编造事实。
"""


def agent_node(state: MessagesState) -> dict[str, list[AIMessage]]:
    """
    Agent 节点：
    - 阅读历史消息
    - 自行决定直接回答或调用工具
    """

    response = llm_with_tools.invoke(
        [
            SystemMessage(content=SYSTEM_PROMPT),
            *state["messages"],
        ]
    )

    return {
        "messages": [response]
    }


# ============================================================
# 4. 构建动态工具调用图
# ============================================================

def create_agent():
    workflow = StateGraph(MessagesState) # MessagesState 是 LangGraph 为聊天模型场景提供的预构建状态，它包含一个 messages 键，并使用 add_messages reducer 合并新消息。

    # LLM 节点：负责思考和选择工具
    workflow.add_node("agent", agent_node)

    # 工具节点：负责执行模型选中的工具
    workflow.add_node("tools", ToolNode(tools))

    workflow.add_edge(START, "agent")

    # tools_condition 会检查 agent 返回的最后一条 AIMessage：
    # - 存在 tool_calls：进入 tools
    # - 不存在 tool_calls：进入 END
    workflow.add_conditional_edges(
        "agent",
        tools_condition,
        {
            "tools": "tools",
            END: END,
        },
    )

    # 工具执行完成后回到 agent。
    # Agent 可以决定继续调用工具，也可以输出最终回答。
    workflow.add_edge("tools", "agent")

    memory = InMemorySaver()

    return workflow.compile(
        checkpointer=memory
    )


# ============================================================
# 5. 交互式运行
# ============================================================

async def main():
    if not os.getenv("LLM_API_KEY"):
        print("❌ 错误：请在 .env 文件中配置 LLM_API_KEY")
        return

    if not os.getenv("TAVILY_API_KEY"):
        print("❌ 错误：请在 .env 文件中配置 TAVILY_API_KEY")
        return

    app = create_agent()

    print("🤖 LangGraph 智能搜索 Agent 已启动！")
    print("Agent 会自行判断是否需要使用工具。")
    print("输入 'quit' 退出。\n")

    # 使用同一个 thread_id，保留同一轮对话中的上下文。
    config = {
        "configurable": {
            "thread_id": "search-session-1"
        }
    }

    while True:
        user_input = input("🤔 您想了解什么：").strip()

        if user_input.lower() in ["quit", "q", "退出", "exit"]:
            print("感谢使用！再见！👋")
            break

        if not user_input:
            continue

        print("\n" + "=" * 60)

        try:
            async for update in app.astream(
                {
                    "messages": [
                        HumanMessage(content=user_input)
                    ]
                },
                config=config,
                stream_mode="updates",
            ):
                for node_name, node_output in update.items():
                    messages = node_output.get("messages", [])

                    if not messages:
                        continue

                    latest_message = messages[-1]

                    if node_name == "agent":
                        if isinstance(latest_message, AIMessage):
                            if latest_message.tool_calls:
                                for tool_call in latest_message.tool_calls:
                                    print(
                                        "🛠️ Agent 选择工具："
                                        f"{tool_call['name']}\n"
                                        f"   参数：{tool_call['args']}"
                                    )

                            elif latest_message.content:
                                print(
                                    "\n💡 最终回答：\n"
                                    f"{latest_message.content}"
                                )

                    elif node_name == "tools":
                        print("✅ 工具执行完成，Agent 正在继续分析。")

        except Exception as exc:
            print(f"❌ 发生错误：{exc}")

        print("=" * 60 + "\n")


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

PS D:\Microsoft VS Code\vs work\codeworkvs\py\myother\agent\project> python .\demo.py
🤖 LangGraph 智能搜索 Agent 已启动！
Agent 会自行判断是否需要使用工具。
输入 'quit' 退出。

🤔 您想了解什么：下周我要去青岛，天气怎么样？有合适的景点吗？

============================================================
🛠️ Agent 选择工具：get_current_time
   参数：{'timezone_name': 'Asia/Shanghai'}
✅ 工具执行完成，Agent 正在继续分析。
🛠️ Agent 选择工具：tavily_search
   参数：{'query': '青岛 2026年6月16日 天气预报 下周'}
✅ 工具执行完成，Agent 正在继续分析。
🛠️ Agent 选择工具：tavily_search
   参数：{'query': '青岛 旅游景点 推荐 必去 景点 2026'}
✅ 工具执行完成，Agent 正在继续分析。

💡 最终回答：
根据搜索到的信息，我来为您整理一下下周（6月16日-22日）青岛的天气和旅游景点建议：

## 🌤️ 天气情况
- **6月16日**：气温17°C~29°C，以多云天气为主
- **整体趋势**：青岛6月平均气温约20.4°C，最高温通常在22-24°C，最低温17-19°C
- **降水**：6月降水量约87mm，无明显持续性降水
- **水温**：海水温度约19°C，适合海边活动
- **建议**：早晚温差较大，建议带薄外套；多云天气适合户外活动

## 🏖️ 推荐景点
根据2026年最新旅游攻略，青岛必去景点包括：

### 经典地标
1. **栈桥**：青岛百年地标，红色八角亭“回澜阁”是青岛啤酒商标的原型
2. **圣弥厄尔教堂**：德国文艺复兴风格天主教堂，奶油色花岗岩外墙搭配橘红色尖顶
3. **八大关**：万国建筑博览馆，百栋风格各异的老建筑，四季景色不同

### 自然风光
4. **崂山**：“海上第一名山”，海拔1132.7米，道教圣地，可游览“九水十八潭”
5. **小鱼山公园**：海拔60米的最佳观景台，可一览红瓦碧海的经典青岛风貌
6. **第二海水浴场**：沙细浪小人少，与八大关相邻，适合散步拍照

### 文化体验
7. **青岛啤酒博物馆**：了解啤酒百年历史，品尝原浆啤酒（门票60元）
8. **德国总督楼旧址博物馆**：百年德式古堡，红蓝绿三色屋顶极具特色
9. **五四广场**：现代城市地标，30米高“五月的风”雕塑夜晚灯光壮观

### 小众打卡
10. **宫崎骏漫画街**：老城小巷改造，充满吉卜力风格
11. **银鱼巷**：百年老建筑改造的潮流艺术街区
12. **李慰農公園**：年轻人拍照热点，树影与海景同框

## 💡 游玩建议
- **行程参考**：可规划4-5天，重点游览栈桥-德国总督楼-小鱼山-天主教堂-啤酒博物馆-八大关-第二海水浴场
- **崂山游览**：景区较大，建议只选1-2个片区深度游，可乘观光车节省体力
- **最佳时间**：傍晚前游览小鱼山或栈桥，可欣赏金色日落光线
- **注意事项**：6月是旅游旺季，建议提前预约崂山等热门景点

## 📌 重要提醒
以上信息来自互联网搜索，天气预报和景点开放情况可能有变化。建议您：
1. 出发前3-5天再次查询最新天气预报
2. 提前预约崂山、啤酒博物馆等热门景点
3. 关注当地旅游公众号获取实时信息

祝您青岛之旅愉快！🌊
============================================================

🤔 您想了解什么：你好。

============================================================

💡 最终回答：
你好！很高兴为您服务。请问有什么我可以帮助您的吗？无论是查询信息、解决问题还是需要推荐，我都很乐意为您提供帮助。😊
============================================================

🤔 您想了解什么：计算321*（1.23+6）

============================================================
🛠️ Agent 选择工具：calculator
   参数：{'expression': '321 * (1.23 + 6)'}
✅ 工具执行完成，Agent 正在继续分析。

💡 最终回答：
计算结果是：**2320.83**

计算过程：
- 先计算括号内：1.23 + 6 = 7.23
- 再计算乘法：321 × 7.23 = 2320.83

如果您有其他计算需求或问题，请随时告诉我！

{
    "name": "calculator",
    "args": {
        "expression": "321 * (1.23 + 6)"
    }
}

🛠️ Agent 选择工具：calculator
   参数：{'expression': '321 * (1.23 + 6)'}

@tool
def calculator(expression: str) -> str:

[
    HumanMessage(
        content="计算321*（1.23+6）"
    ),

    AIMessage(
        content="",
        tool_calls=[
            {
                "name": "calculator",
                "args": {
                    "expression": "321 * (1.23 + 6)"
                }
            }
        ]
    ),

    ToolMessage(
        content="计算结果：2320.83"
    )
]

计算结果是：2320.83

计算过程：
- 先计算括号内：1.23 + 6 = 7.23
- 再计算乘法：321 × 7.23 = 2320.83

用户输入：
计算321*（1.23+6）

        ↓

第一次调用 agent_node
模型决定调用 calculator

        ↓

AIMessage.tool_calls：
calculator("321 * (1.23 + 6)")

        ↓

ToolNode 执行 calculator

        ↓

calculator 返回：
"计算结果：2320.83"

        ↓

ToolNode 将结果包装为 ToolMessage

        ↓

第二次调用 agent_node
模型读取原始问题和工具结果

        ↓

模型生成自然语言回答：
结果 + 计算步骤 + 客套话

        ↓

没有新的 tool_calls

        ↓

tools_condition 路由到 END

​

​

​

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,))`

​

​

​

​

​

​

{
  "id": "simple_001",
  "question": "What's the weather like in Beijing today?",
  "function": [
    {
      "name": "get_weather",
      "description": "Get the current weather for a location",
      "parameters": {
        "type": "object",
        "properties": {
          "location": {
            "type": "string",
            "description": "The city name"
          }
        },
        "required": ["location"]
      }
    }
  ],
  "ground_truth": [
    {
      "name": "get_weather",
      "arguments": {
        "location": "Beijing"
      }
    }
  ]
}

# 安装BFCL评估工具
pip install bfcl

# 运行官方评估
bfcl evaluate \
    --model-result-path ./results.json \
    --test-category simple_python

# 克隆BFCL仓库
git clone https://github.com/ShishirPatil/gorilla.git temp_gorilla
cd temp_gorilla/berkeley-function-call-leaderboard

# 查看BFCL v4数据集
ls bfcl_eval/data/
# 输出: BFCL_v4_simple_python.json  BFCL_v4_multiple.json  BFCL_v4_parallel.json  ...

# 查看ground truth
ls bfcl_eval/data/possible_answer/
# 输出: BFCL_v4_simple_python.json  BFCL_v4_multiple.json  ...

{
  "task_id": "gaia_001",
  "Question": "What is the total population of the top 3 most populous cities in California?",
  "Level": 2,
  "Final answer": "12847521",
  "file_name": "",
  "file_path": "",
  "Annotator Metadata": {
    "Steps": [
      "Search for most populous cities in California",
      "Get population data for top 3 cities",
      "Sum the populations"
    ],
    "Number of steps": 3,
    "How long did this take?": "5 minutes",
    "Tools": ["web_search", "calculator"]
  }
}

GAIA_SYSTEM_PROMPT = """You are a general AI assistant. I will ask you a question. Report your thoughts, and finish your answer with the following template: FINAL ANSWER: [YOUR FINAL ANSWER].

YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings.

If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise.

If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise.

If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string."""

from huggingface_hub import snapshot_download
import os

# 设置Token
os.environ["HF_TOKEN"] = "hf_your_token_here"

# 下载数据集
snapshot_download(
    repo_id="gaia-benchmark/GAIA",
    repo_type="dataset",
    local_dir="./data/gaia",
    token=os.getenv("HF_TOKEN")
)

# 查看数据集统计
stats = dataset.get_statistics()
print(f"总样本数: {stats['total_samples']}")
print(f"级别分布: {stats['level_distribution']}")
# 输出:
# 总样本数: 165
# 级别分布: {1: 53, 2: 62, 3: 50}