# ==================== 第一部分：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

​

​

​

{
  "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}