import os, getpass
import asyncio
import nest_asyncio
from typing import List
from dotenv import load_dotenv
import logging
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool as langchain_tool
from langchain.agents import create_tool_calling_agent, AgentExecutor

# 取消注释
# 安全地提示用户并将API密钥设置为环境变量
os.environ["GOOGLE_API_KEY"] = getpass.getpass("输入您的Google API密钥: ")
os.environ["OPENAI_API_KEY"] = getpass.getpass("输入您的OpenAI API密钥: ")

try:
    # 需要具有函数/工具调用能力的模型。
    llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
    print(f"✅ 语言模型已初始化: {llm.model}")
except Exception as e:
    print(f"🛑 初始化语言模型时出错: {e}")
    llm = None

# --- 定义工具 ---
@langchain_tool
def search_information(query: str) -> str:
    """
    提供给定主题的事实信息。使用此工具查找
    如'法国首都'或'伦敦天气？'等短语的答案。
    """
    print(f"\n--- 🛠️ 工具调用: search_information，查询: '{query}' ---")
    
    # 使用预定义结果字典模拟搜索工具。
    simulated_results = {
        "伦敦天气": "伦敦目前多云，温度15°C。",
        "法国首都": "法国的首都是巴黎。",
        "地球人口": "地球的估计人口约为80亿人。",
        "最高山峰": "珠穆朗玛峰是海拔最高的山峰。",
        "默认": f"'{query}'的模拟搜索结果: 未找到特定信息，但主题似乎很有趣。"
    }
    
    result = simulated_results.get(query.lower(), simulated_results["默认"])
    print(f"--- 工具结果: {result} ---")
    return result

tools = [search_information]

# --- 创建工具调用智能体 ---
if llm:
    # 此提示模板需要`agent_scratchpad`占位符用于智能体的内部步骤。
    agent_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个有用的助手。"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])
    
    # 创建智能体，将LLM、工具和提示绑定在一起。
    agent = create_tool_calling_agent(llm, tools, agent_prompt)
    
    # AgentExecutor是调用智能体并执行所选工具的运行时。
    # 这里不需要'tools'参数，因为它们已经绑定到智能体。
    agent_executor = AgentExecutor(agent=agent, verbose=True, tools=tools)

async def run_agent_with_tool(query: str):
    """使用查询调用智能体执行器并打印最终响应。"""
    print(f"\n--- 🏃 使用查询运行智能体: '{query}' ---")
    try:
        response = await agent_executor.ainvoke({"input": query})
        print("\n--- ✅ 最终智能体响应 ---")
        print(response["output"])
    except Exception as e:
        print(f"\n🛑 智能体执行期间发生错误: {e}")

async def main():
    """并发运行所有智能体查询。"""
    tasks = [
        run_agent_with_tool("法国的首都是什么？"),
        run_agent_with_tool("伦敦的天气如何？"),
        run_agent_with_tool("告诉我关于狗的事情。")  # 应该触发默认工具响应
    ]
    await asyncio.gather(*tasks)

nest_asyncio.apply()
asyncio.run(main())

# pip install crewai langchain-openai

import os
from crewai import Agent, Task, Crew
from crewai.tools import tool
import logging

# --- 最佳实践: 配置日志记录 ---
# 基本日志设置有助于调试和跟踪crew的执行。
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# --- 设置您的API密钥 ---
# 对于生产，建议使用更安全的密钥管理方法
# 如运行时加载的环境变量或密钥管理器。
#
# 设置您选择的LLM提供商的环境变量（如OPENAI_API_KEY）
# os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
# os.environ["OPENAI_MODEL_NAME"] = "gpt-4o"

# --- 1. 重构工具: 返回清洁数据 ---
# 工具现在返回原始数据（浮点数）或引发标准Python错误。
# 这使其更可重用并强制智能体正确处理结果。
@tool("股票价格查找工具")
def get_stock_price(ticker: str) -> float:
    """
    获取给定股票代码符号的最新模拟股票价格。
    返回价格作为浮点数。如果未找到代码，则引发ValueError。
    """
    logging.info(f"工具调用: get_stock_price，代码 '{ticker}'")
    simulated_prices = {
        "AAPL": 178.15,
        "GOOGL": 1750.30,
        "MSFT": 425.50,
    }
    
    price = simulated_prices.get(ticker.upper())
    if price is not None:
        return price
    else:
        # 引发特定错误比返回字符串更好。
        # 智能体配备处理异常并可以决定下一步行动。
        raise ValueError(f"未找到代码 '{ticker.upper()}' 的模拟价格。")

# --- 2. 定义智能体 ---
# 智能体定义保持不变，但现在将利用改进的工具。
financial_analyst_agent = Agent(
  role='高级金融分析师',
  goal='使用提供的工具分析股票数据并报告关键价格。',
  backstory="你是一名经验丰富的金融分析师，擅长使用数据源查找股票信息。你提供清晰、直接的答案。",
  verbose=True,
  tools=[get_stock_price],
  # 允许委托可能有用，但对于此简单任务不是必需的。
  allow_delegation=False,
)

# --- 3. 完善任务: 更清晰的指令和错误处理 ---
# 任务描述更具体，指导智能体如何反应
# 成功数据检索和潜在错误。
analyze_aapl_task = Task(
  description=(
      "苹果（代码: AAPL）的当前模拟股票价格是什么？ "
      "使用'股票价格查找工具'查找它。 "
      "如果未找到代码，你必须报告你无法检索价格。"
  ),
  expected_output=(
      "关于AAPL模拟股票价格的单一、清晰句子。 "
      "例如: 'AAPL的模拟股票价格是$178.15。' "
      "如果找不到价格，请明确说明。"
  ),
  agent=financial_analyst_agent,
)

# --- 4. 制定Crew ---
# crew编排智能体和任务如何协同工作。
financial_crew = Crew(
  agents=[financial_analyst_agent],
  tasks=[analyze_aapl_task],
  verbose=True # 在生产中设置为False以减少详细日志
)

# --- 5. 在主执行块内运行Crew ---
# 使用__name__ == "__main__":块是标准Python最佳实践。
def main():
    """运行crew的主函数。"""
    # 在开始前检查API密钥以避免运行时错误。
    if not os.environ.get("OPENAI_API_KEY"):
        print("错误: 未设置OPENAI_API_KEY环境变量。")
        print("请在运行脚本之前设置它。")
        return
    
    print("\n## 启动金融Crew...")
    print("---------------------------------")
    
    # kickoff方法开始执行。
    result = financial_crew.kickoff()
    
    print("\n---------------------------------")
    print("## Crew执行完成。")
    print("\n最终结果:\n", result)

if __name__ == "__main__":
    main()

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.genai import types
import nest_asyncio
import asyncio

# 定义会话设置和智能体执行所需的变量
APP_NAME="Google Search_agent"
USER_ID="user1234"
SESSION_ID="1234"

# 定义具有搜索工具访问权限的智能体
root_agent = ADKAgent(
   name="basic_search_agent",
   model="gemini-2.0-flash-exp",
   description="使用Google搜索回答问题的智能体。",
   instruction="我可以通过搜索互联网来回答你的问题。只要问我任何问题！",
   tools=[google_search] # Google搜索是执行Google搜索的预构建工具。
)

# 智能体交互
async def call_agent(query):
   """
   使用查询调用智能体的辅助函数。
   """
   # 会话和运行器
   session_service = InMemorySessionService()
   session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
   runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service)
   
   content = types.Content(role='user', parts=[types.Part(text=query)])
   events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
   
   for event in events:
       if event.is_final_response():
           final_response = event.content.parts[0].text
           print("智能体响应: ", final_response)

nest_asyncio.apply()
asyncio.run(call_agent("最新的AI新闻是什么？"))

import os, getpass
import asyncio
import nest_asyncio
from typing import List
from dotenv import load_dotenv
import logging
from google.adk.agents import Agent as ADKAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.adk.code_executors import BuiltInCodeExecutor
from google.genai import types

# 定义会话设置和智能体执行所需的变量
APP_NAME="calculator"
USER_ID="user1234"
SESSION_ID="session_code_exec_async"

# 智能体定义
code_agent = LlmAgent(
   name="calculator_agent",
   model="gemini-2.0-flash",
   code_executor=BuiltInCodeExecutor(),
   instruction="""你是一个计算器智能体。
   当给定数学表达式时，编写并执行Python代码来计算结果。
   仅以纯文本返回最终数值结果，不带markdown或代码块。
   """,
   description="执行Python代码进行计算。",
)

# 智能体交互（异步）
async def call_agent_async(query):
   # 会话和运行器
   session_service = InMemorySessionService()
   session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
   runner = Runner(agent=code_agent, app_name=APP_NAME, session_service=session_service)
   
   content = types.Content(role='user', parts=[types.Part(text=query)])
   print(f"\n--- 运行查询: {query} ---")
   final_response_text = "未捕获最终文本响应。"
   
   try:
       # 使用run_async
       async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
           print(f"事件ID: {event.id}, 作者: {event.author}")
           
           # --- 首先检查特定部分 ---
           # has_specific_part = False
           if event.content and event.content.parts and event.is_final_response():
               for part in event.content.parts: # 迭代所有部分
                   if part.executable_code:
                       # 通过.code访问实际代码字符串
                       print(f"  调试: 智能体生成代码:\n```python\n{part.executable_code.code}\n```")
                       has_specific_part = True
                   elif part.code_execution_result:
                       # 正确访问结果和输出
                       print(f"  调试: 代码执行结果: {part.code_execution_result.outcome} - 输出:\n{part.code_execution_result.output}")
                       has_specific_part = True
                   # 也为调试打印在任何事件中找到的任何文本部分
                   elif part.text and not part.text.isspace():
                       print(f"  文本: '{part.text.strip()}'")
                       # 不要在这里设置has_specific_part=True，因为我们想要下面的最终响应逻辑
               
               # --- 在特定部分之后检查最终响应 ---
               text_parts = [part.text for part in event.content.parts if part.text]
               final_result = "".join(text_parts)
               print(f"==> 最终智能体响应: {final_result}")
   except Exception as e:
       print(f"智能体运行期间出错: {e}")
   print("-" * 30)

# 运行示例的主异步函数
async def main():
   await call_agent_async("计算(5 + 7) * 3的值")
   await call_agent_async("10的阶乘是什么？")

# 执行主异步函数
try:
   nest_asyncio.apply()
   asyncio.run(main())
except RuntimeError as e:
   # 处理在已运行事件循环（如Jupyter/Colab）中调用asyncio.run时的特定错误
   if "cannot be called from a running event loop" in str(e):
       print("\n在现有事件循环中运行（如Colab/Jupyter）。")
       print("请在笔记本单元格中运行`await main()`。")
       # 如果在交互式环境（如笔记本）中，您可能需要运行:
       # await main()
   else:
       raise e # 重新引发其他运行时错误

import asyncio
from google.genai import types
from google.adk import agents
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
import os

# --- 配置 ---
# 确保您设置了GOOGLE_API_KEY和DATASTORE_ID环境变量
# 例如:
# os.environ["GOOGLE_API_KEY"] = "YOUR_API_KEY"
# os.environ["DATASTORE_ID"] = "YOUR_DATASTORE_ID"
DATASTORE_ID = os.environ.get("DATASTORE_ID")

# --- 应用程序常量 ---
APP_NAME = "vsearch_app"
USER_ID = "user_123"  # 示例用户ID
SESSION_ID = "session_456" # 示例会话ID

# --- 智能体定义（使用指南中较新的模型更新） ---
vsearch_agent = agents.VSearchAgent(
    name="q2_strategy_vsearch_agent",
    description="使用Vertex AI搜索回答Q2策略文档问题。",
    model="gemini-2.0-flash-exp", # 基于指南示例的更新模型
    datastore_id=DATASTORE_ID,
    model_parameters={"temperature": 0.0}
)

# --- 运行器和会话初始化 ---
runner = Runner(
    agent=vsearch_agent,
    app_name=APP_NAME,
    session_service=InMemorySessionService(),
)

# --- 智能体调用逻辑 ---
async def call_vsearch_agent_async(query: str):
    """初始化会话并流式传输智能体的响应。"""
    print(f"用户: {query}")
    print("智能体: ", end="", flush=True)
    
    try:
        # 正确构造消息内容
        content = types.Content(role='user', parts=[types.Part(text=query)])
        
        # 处理从异步运行器到达的事件
        async for event in runner.run_async(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
        ):
            # 用于响应文本的逐令牌流式传输
            if hasattr(event, 'content_part_delta') and event.content_part_delta:
                print(event.content_part_delta.text, end="", flush=True)
            
            # 处理最终响应及其关联元数据
            if event.is_final_response():
                print() # 流式响应后的换行
                if event.grounding_metadata:
                    print(f"  (源归因: 找到{len(event.grounding_metadata.grounding_attributions)}个源)")
                else:
                    print("  (未找到基础元数据)")
                print("-" * 30)
    except Exception as e:
        print(f"\n发生错误: {e}")
        print("请确保您的数据存储ID正确且服务账户具有必要权限。")
        print("-" * 30)

# --- 运行示例 ---
async def run_vsearch_example():
    # 替换为与您的数据存储内容相关的问题
    await call_vsearch_agent_async("总结Q2策略文档的要点。")
    await call_vsearch_agent_async("实验室X提到了哪些安全程序？")

# --- 执行 ---
if __name__ == "__main__":
    if not DATASTORE_ID:
        print("错误: 未设置DATASTORE_ID环境变量。")
    else:
        try:
            asyncio.run(run_vsearch_example())
        except RuntimeError as e:
            # 这处理在已有运行事件循环的环境（如Jupyter笔记本）中调用asyncio.run的情况。
            if "cannot be called from a running event loop" in str(e):
                print("在运行事件循环中跳过执行。请直接运行此脚本。")
            else:
                raise e