俊瑶先森

from mcp.server.fastmcp import FastMCP
from typing import Literal
from tavily import TavilyClient

mcp = FastMCP("Web-Search-Server")

tavily_client = TavilyClient(api_key="tvly-xxxx")  # 修改为你的key

@mcp.tool()
def web_search(query: str, max_results: int = 5,
               topic: Literal["general", "news", "finance"] = "general",
               include_raw_content: bool = False):
    """Run a web search"""
    return tavily_client.search(
        query,
        max_results=max_results,
        include_raw_content=include_raw_content,
        topic=topic,
    )
    
@mcp.tool()
def extract(url: str):
    """Extract web page content from URL."""
    return tavily_client.extract(url)

if __name__ == "__main__":
    mcp.settings.port = 6030
    mcp.run("sse")

import os
from dotenv import load_dotenv
from langchain_core.messages import AIMessageChunk

# 加载 .env 文件中的环境变量
load_dotenv()

# 配置自建模型（必须从 .env 文件中读取）
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

# 检查必需的配置项是否存在
if not BASE_URL:
    raise ValueError("OPENAI_BASE_URL 未在 .env 文件中配置")
if not API_KEY:
    raise ValueError("OPENAI_API_KEY 未在 .env 文件中配置")
if not MODEL_NAME:
    raise ValueError("OPENAI_MODEL 未在 .env 文件中配置")

from deepagents import create_deep_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.chat_models import init_chat_model
import asyncio

system_prompt = """
你是一位研究专家。你的工作是根据用户的要求进行彻底的研究，然后写一份润色的报告。
## 工作流程
1. 理解核心需求，识别关键要素。
2. 制定任务规划，明确待办事项。
3. 逐步研究，记录发现，验证结论。
4. 整体结果和审查, 必要时重新制定解决链路。
5. 输出详细研究报告。
"""

async def main():
    ## 初始化模型
    # 创建模型实例，使用自建模型（兼容 OpenAI API 格式）
    llm = init_chat_model(
        model=MODEL_NAME,
        model_provider="openai",
        temperature=0,
        base_url=BASE_URL,
        api_key=API_KEY,
    )
    
    ## 定义MCP Server, 执行前面创建的
    client = MultiServerMCPClient(
        {
            "web-search": {
                "url": "http://localhost:6030/sse",
                "transport": "sse"
            }
        }
    )
    
    ## 获取 MCP Tools
    tools = await client.get_tools()
    
    ## 创建深度智能体
    agent = create_deep_agent(
        model=llm,
        tools=tools,
        system_prompt=system_prompt
    )
    
    ## 执行深度研究，以流式的方式输出
    async for stream_type, chunk in agent.astream(
            input={
                "messages": [
                    {"role": "user", "content": "埃菲尔铁塔与最高建筑相比有多高？"}
                ]
            },
            stream_mode=["updates", "messages"]
    ):
        if stream_type == "messages" and type(chunk[0]) is AIMessageChunk:
            content = chunk[0].content
            if not content:
                continue
            print(content, end="", flush=True)
        elif stream_type == "updates":
            if "model" in chunk:
                model = chunk["model"]
                if "messages" in model:
                    messages = model["messages"]
                    for message in messages:
                        tool_calls = message.tool_calls
                        for tool_call in tool_calls:
                            name = tool_call["name"]
                            args = tool_call["args"]
                            if name == "write_todos":
                                todos = args["todos"]
                                print_todos = []
                                for todo in todos:
                                    todo_content = todo["content"]
                                    status = todo["status"]
                                    print_todos.append(f"> {todo_content}  -->  {status}")
                                if print_todos:
                                    print_todos = "\n".join(print_todos)
                                    print(f"\n> TODO: \n{print_todos}\n\n")
                            else:
                                print(f"\n> Call MCP: {name}, args: {args}\n")

            if "tools" in chunk:
                tools = chunk["tools"]
                if "messages" in tools:
                    messages = tools["messages"]
                    for message in messages:
                        content = message.content
                        if "Updated todo list" in content:
                            continue
                        print("\nMCP Result: \n", content, "\n")

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

import os
from dotenv import load_dotenv
from langchain_core.messages import AIMessageChunk

# 加载 .env 文件中的环境变量
load_dotenv()

# 配置自建模型（必须从 .env 文件中读取）
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

# 检查必需的配置项是否存在
if not BASE_URL:
    raise ValueError("OPENAI_BASE_URL 未在 .env 文件中配置")
if not API_KEY:
    raise ValueError("OPENAI_API_KEY 未在 .env 文件中配置")
if not MODEL_NAME:
    raise ValueError("OPENAI_MODEL 未在 .env 文件中配置")

from fastapi import FastAPI, HTTPException, Header, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import List, Dict, Any, Optional, Generator
from langchain.chat_models import init_chat_model
from langchain_mcp_adapters.client import MultiServerMCPClient
from deepagents import create_deep_agent
from datetime import datetime
import time, uuid

app = FastAPI(title="OpenAI Compatible Chat API")
api_key = "sk-da4b6cb4a41e4cascascasc9508deb556942"
agent = None

system_prompt = """
你是一位研究专家。你的工作是根据用户的要求进行彻底的研究，然后写一份润色的报告。
## 工作流程
1. 理解核心需求，识别关键要素。
2. 制定任务规划，明确待办事项。
3. 逐步研究，记录发现，验证结论。
4. 整体结果和审查, 必要时重新制定解决链路。
5. 输出详细研究报告。
"""

# 创建模型实例，使用自建模型（兼容 OpenAI API 格式）
llm = init_chat_model(
    model=MODEL_NAME,
    model_provider="openai",
    temperature=0,
    base_url=BASE_URL,
    api_key=API_KEY,
)

**关键点说明：**
- 模型配置从 `.env` 文件读取，便于不同环境使用不同配置
- `init_chat_model()` 支持自建模型，只需提供兼容 OpenAI API 格式的服务端点
- 全局变量 `llm` 在应用启动时初始化，避免重复创建


class ChatCompletionRequest(BaseModel):
    model: str
    messages: List[Dict[str, Any]]
    temperature: Optional[float] = 1.0
    max_tokens: Optional[int] = None
    stream: Optional[bool] = False


class ChatCompletionChoice(BaseModel):
    index: int
    message: List[Dict[str, Any]]
    finish_reason: str


class ChatCompletionResponse(BaseModel):
    id: str
    object: str = "chat.completion"
    created: int
    model: str
    choices: List[ChatCompletionChoice]
    usage: Dict[str, int]


class ChatCompletionChunk(BaseModel):
    id: str
    object: str = "chat.completion.chunk"
    created: int
    model: str
    choices: List[Dict[str, Any]]


async def create_agent():
    '''创建MCP Agent'''
    client = MultiServerMCPClient(
        {
            "web-search": {
                "url": "http://localhost:6030/sse",
                "transport": "sse"
            }
        }
    )
    tools = await client.get_tools()
    return create_deep_agent(
        model=llm,
        tools=tools,
        system_prompt=system_prompt
    )


async def invoke_agent(messages: []):
    global agent
    if not agent:
        agent = await create_agent()
    async for stream_type, chunk in agent.astream({"messages": messages}, stream_mode=["updates", "messages"]):
        if stream_type == "messages" and type(chunk[0]) is AIMessageChunk:
            content = chunk[0].content
            if not content:
                continue
            yield content
        elif stream_type == "updates":
            if "model" in chunk:
                model = chunk["model"]
                if "messages" in model:
                    messages = model["messages"]
                    for message in messages:
                        tool_calls = message.tool_calls
                        for tool_call in tool_calls:
                            name = tool_call["name"]
                            args = tool_call["args"]
                            if name == "write_todos":
                                todos = args["todos"]
                                print_todos = []
                                for todo in todos:
                                    todo_content = todo["content"]
                                    status = todo["status"]
                                    print_todos.append(f"> {todo_content}  -->  {status}")
                                if print_todos:
                                    print_todos = "\n".join(print_todos)
                                    yield f"\n> TODO: \n{print_todos}\n\n"
                            else:
                                yield f"\n> Call MCP Server: {name}, args: {args}\n\n"
            # 客户端暂时不输出工具的输出，如果需要，放开注释即可
            # if "tools" in chunk:
            #     tools = chunk["tools"]
            #     if "messages" in tools:
            #         messages = tools["messages"]
            #         for message in messages:
            #             content = message.content
            #             if "Updated todo list" in content:
            #                 continue
            #             yield f"> MCP Response: \n\n```json\n{content}\n```\n"


async def handle_stream_response(messages: [],
                                 model: str, request_id: str) -> Generator[str, None, None]:
    # 执行 agent
    async for msg in invoke_agent(messages):
        chunk_data = ChatCompletionChunk(
            id=request_id,
            created=int(time.time()),
            model=model,
            choices=[{
                "index": 0,
                "delta": {
                    "content": msg
                },
                "finish_reason": None
            }]
        )
        yield f"data: {chunk_data.model_dump_json()}\n\n"

    final_chunk = ChatCompletionChunk(
        id=request_id,
        created=int(time.time()),
        model=model,
        choices=[{
            "index": 0,
            "delta": {},
            "finish_reason": "stop"
        }]
    )
    yield f"data: {final_chunk.model_dump_json()}\n\n"
    yield "data: [DONE]\n\n"


async def verify_auth(authorization: Optional[str] = Header(None)) -> bool:
    '''验证token'''
    if not authorization:
        return False
    if authorization.startswith("Bearer "):
        token = authorization[7:]
    else:
        token = authorization
    return token == api_key


@app.post("/v1/chat/completions")
async def chat_completions(request: ChatCompletionRequest, auth_result: bool = Depends(verify_auth)):
    # 检查身份验证结果
    if not auth_result:
        raise HTTPException(
            status_code=401,
            detail={
                "error": {
                    "message": "Invalid authentication credentials",
                    "type": "invalid_request_error",
                    "param": None,
                    "code": "invalid_api_key"
                }
            },
            headers={"WWW-Authenticate": "Bearer"}
        )
    ## 暂不支持非流式返回
    if not request.stream:
        raise HTTPException(
            status_code=400,
            detail={
                "error": {
                    "message": "Streaming responses are not implemented in this mock API",
                    "type": "invalid_request_error",
                    "param": "stream",
                    "code": "invalid_parameter"
                }
            }
        )
    try:
        # 触发 agent 并流式返回
        request_id = f"chatcmpl-{uuid.uuid4().hex[:8]}"
        return StreamingResponse(
            handle_stream_response(request.messages, request.model, request_id),
            media_type="text/plain",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "Content-Type": "text/event-stream"
            }
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/v1/models")
async def list_models():
    """列出可用模型"""
    return {
        "object": "list",
        "data": [
            {
                "id": "agent_model",
                "object": "model",
                "created": int(time.time()),
                "owned_by": "agent"
            }
        ]
    }


@app.get("/health")
async def health_check():
    """健康检查"""
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from langchain.tools import tool

@tool
def get_exchange_rate(currency: str) -> str:
    """
    获取货币汇率信息
    
    这个工具用于查询指定货币对人民币的汇率。
    在实际应用中，这里可以对接真实的汇率 API（如 ExchangeRate-API）。
    
    Args:
        currency: 货币代码，例如"USD"（美元）、"EUR"（欧元）
    
    Returns:
        返回格式化的汇率信息字符串
    """
    # 简单的汇率数据映射表（实际应用中应该从汇率API获取）
    if currency == "USD" or currency == "美元":
        return "美元对人民币汇率：1 USD = 7.25 CNY"
    if currency == "EUR" or currency == "欧元":
        return "欧元对人民币汇率：1 EUR = 7.85 CNY"
    return f"没有找到{currency}的汇率信息"

@tool
def calculate(expression: str) -> str:
    """
    计算表达式
    
    这个工具用于执行数学表达式计算。
    注意：实际生产环境中应该使用更安全的表达式解析库（如 numexpr），
    而不是直接使用 eval()，这里仅用于演示。
    
    Args:
        expression: 数学表达式字符串，例如"100 * 7.25"
    
    Returns:
        计算结果字符串
    """
    # 使用 eval 计算表达式（仅用于演示，生产环境需更安全的方案）
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误: {str(e)}"

# 定义工具列表，用于绑定到 LLM
# LLM 可以根据工具描述自动决定何时调用哪个工具
TOOLS = [get_exchange_rate, calculate]

# 创建工具字典，key 为工具名称，value 为工具对象
# 在执行工具调用时，通过工具名称快速查找对应的工具对象
TOOL_DICT = {tool.name: tool for tool in TOOLS}

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# 加载 .env 文件中的环境变量
load_dotenv()

# 配置自建模型（必须从 .env 文件中读取）
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

# 检查必需的配置项是否存在
if not BASE_URL:
    raise ValueError("OPENAI_BASE_URL 未在 .env 文件中配置")
if not API_KEY:
    raise ValueError("OPENAI_API_KEY 未在 .env 文件中配置")
if not MODEL_NAME:
    raise ValueError("OPENAI_MODEL 未在 .env 文件中配置")

# 创建模型实例
LLM = init_chat_model(
    model=MODEL_NAME,
    model_provider="openai",
    temperature=0.7,
    base_url=BASE_URL,
    api_key=API_KEY,
)

# 将 LLM 与工具列表绑定
# bind_tools() 方法会将工具的描述信息注入到模型的上下文中
# 这样模型在生成回复时，可以识别何时需要调用工具，并生成相应的工具调用请求
LLM_WITH_TOOLS = LLM.bind_tools(TOOLS)

from langgraph.graph.message import MessagesState

class PlanAndExecuteState(MessagesState):
    """
    计划和执行状态定义
    
    这个状态类用于在整个 Agent 执行过程中保存和传递信息。
    
    继承自 MessagesState 的属性：
        messages: List[BaseMessage] - 消息列表，包含对话历史、工具调用等
    
    自定义属性：
        plan: str - 生成的执行计划文本
        step: int - 当前执行的步骤编号（从 0 开始）
    """

    plan: str  # 生成的计划，由 plan_node 生成并保存
    step: int  # 当前步骤，每次执行后递增，用于跟踪执行进度

from langchain.messages import SystemMessage

# 生成计划的 System Prompt
# 这个提示词指导 LLM 如何生成执行计划
SYSTEM_PROMPT_PLAN = """
你是一位智能助手，擅长解决用户提出的各种问题。请为用户提出的问题创建分析方案步骤。

如果有需要，可以调用工具。

你可以调用工具的列表如下：
get_exchange_rate:
    获取指定货币对人民币的汇率信息
    Parameters:
    -----------
    currency : str
        货币代码，例如"USD"（美元）、"EUR"（欧元）
    Returns:
    --------
    str
        指定货币的汇率信息

calculate:
    计算表达式
    Parameters:
    -----------
    expression : str
        表达式内容
    Returns:
    --------
    str
        表达式的计算结果

要求：
1.用中文列出清晰步骤
2.每个步骤标记序号
3.明确说明需要分析和执行的内容
4.只需输出计划内容，不要做任何额外的解释和说明
5.设计的方案步骤要紧紧贴合我的工具所能返回的内容，不要超出工具返回的内容
"""

def plan_node(state: PlanAndExecuteState) -> PlanAndExecuteState:
    """
    生成计划节点
    
    这个节点的职责：
    1. 接收用户的问题（在 state["messages"] 中）
    2. 调用 LLM 生成执行计划
    3. 将计划保存到状态中，供后续节点使用
    
    Args:
        state: 当前状态，包含用户消息
    
    Returns:
        更新后的状态，包含生成的计划
    """
    
    # 构造消息列表
    # SystemMessage 放在最前面，用于设置 LLM 的角色和行为
    # state["messages"] 包含用户的问题
    messages = [SystemMessage(content=SYSTEM_PROMPT_PLAN)] + state["messages"]
    
    # 调用 LLM 生成计划
    # invoke() 方法会发送消息到 LLM，并返回回复
    # .content 获取回复的文本内容（计划文本）
    response = LLM_WITH_TOOLS.invoke(input=messages)
    plan = response.content
    print(f"\n【生成的计划】\n{plan}\n")
    
    # 更新状态，保存生成的计划
    # 这个计划将在 execute_node 中被使用
    state["plan"] = plan
    
    # 返回更新后的状态
    # LangGraph 会将这个状态传递给下一个节点
    return state

from langchain.messages import SystemMessage

# 执行计划的 System Prompt
# 这个提示词指导 LLM 如何执行计划中的步骤
SYSTEM_PROMPT_EXECUTE = """你是一位思路清晰、有条理的智能助手，你必须严格按照以下计划执行任务：

当前计划：
{plan}

如果你认为计划已经执行到最后一步了，请在内容的末尾加上 Final Answer 字样

示例：
这个问题的最终答案是99.99。Final Answer"""

def execute_node(state: PlanAndExecuteState) -> PlanAndExecuteState:
    """
    执行计划节点
    
    这个节点的职责：
    1. 读取计划节点生成的计划
    2. 根据计划、对话历史和工具调用结果，执行当前步骤
    3. LLM 可能直接回复，也可能生成工具调用请求
    4. 更新执行步骤计数器
    
    Args:
        state: 当前状态，包含计划、消息历史和步骤编号
    
    Returns:
        更新后的状态，包含执行结果
    """
    
    # 获取当前步骤编号
    # 用于跟踪执行进度和日志输出
    step = state["step"]
    
    # 构造消息列表
    # SystemMessage 包含当前计划，指导 LLM 执行
    # state["messages"] 包含完整的对话历史（用户消息、AI回复、工具调用结果等）
    system_message = SystemMessage(content=SYSTEM_PROMPT_EXECUTE.format(plan=state["plan"]))
    messages = [system_message] + state["messages"]
    
    # 调用 LLM 执行计划
    # LLM 会分析当前状态，决定下一步操作
    # 如果 LLM 认为需要调用工具，会在 resp.tool_calls 中返回工具调用信息
    # 如果 LLM 认为可以直接回复，会在 resp.content 中返回文本内容
    resp = LLM_WITH_TOOLS.invoke(input=messages)
    print(f"\n【第 {step + 1} 步】{resp.content}")
    
    # 更新状态
    # 1. 将 LLM 的回复添加到消息历史中
    # 2. 递增步骤计数器
    state["messages"].append(resp)
    state["step"] = step + 1
    
    # 返回更新后的状态
    # 后续的条件边会根据 resp 的内容决定下一步流向
    return state

from langchain.messages import ToolMessage

def tool_node(state: PlanAndExecuteState) -> PlanAndExecuteState:
    """
    工具调用节点
    
    这个节点的职责：
    1. 从最后一条消息中提取工具调用信息
    2. 根据工具名称查找对应的工具对象
    3. 使用工具参数执行工具调用
    4. 将工具调用结果封装为 ToolMessage，添加到消息历史中
    
    Args:
        state: 当前状态，最后一条消息应包含 tool_calls
    
    Returns:
        更新后的状态，包含工具调用结果
    """
    
    # 获取最后一条消息中的工具调用信息
    # 最后一条消息应该是 execute_node 中 LLM 生成的回复
    # 如果 LLM 决定调用工具，这条消息会包含 tool_calls 字段
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls
    
    # 如果没有工具调用，直接返回（理论上不应该发生，因为只有有工具调用才会进入此节点）
    if tool_calls is None:
        return state
    
    # 遍历所有工具调用（LLM 可能同时调用多个工具）
    for tool_call in tool_calls:
        # 提取工具名称和参数
        # tool_call 是一个字典，包含：
        #   - "name": 工具名称，如 "get_exchange_rate"
        #   - "args": 工具参数字典，如 {"currency": "USD"}
        #   - "id": 工具调用 ID，用于关联 ToolMessage
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        
        # 从工具字典中查找对应的工具对象
        tool = TOOL_DICT.get(tool_name)
        if tool is None:
            # 如果找不到工具，跳过（实际应用中应该记录错误）
            continue
        
        # 执行工具调用
        # invoke() 方法会使用 tool_args 调用工具函数
        # 例如：get_exchange_rate.invoke({"currency": "USD"}) -> "美元对人民币汇率：1 USD = 7.25 CNY"
        tool_call_result = tool.invoke(input=tool_args)
        print(f"\n调用工具: {tool_name}")
        print(f"参数: {tool_args}")
        print(f"执行结果: {tool_call_result}")
        
        # 将工具调用结果封装为 ToolMessage，添加到消息历史中
        # ToolMessage 需要包含 tool_call_id，用于关联对应的工具调用请求
        # 这样 LLM 在后续回复时，可以知道哪个工具调用产生了哪个结果
        state["messages"].append(
            ToolMessage(
                content=f"工具调用结果: {tool_call_result}",
                tool_call_id=tool_call["id"],  # 关联工具调用请求
            )
        )
    
    # 返回更新后的状态
    # 状态中现在包含了工具调用结果，流程会回到 execute_node 继续执行
    return state

from langgraph.graph import END

def conditional_edge(state: PlanAndExecuteState) -> str:
    """
    条件边函数
    
    这个函数决定 execute_node 执行后的下一步流向：
    - 如果 LLM 回复中包含 "Final Answer"，返回 END，表示任务完成
    - 否则返回 "tool_node"，表示需要调用工具
    
    Args:
        state: 当前状态，包含执行结果
    
    Returns:
        下一个节点的名称，或 END 表示结束
    """
    
    # 获取最后一条消息（应该是 execute_node 中 LLM 的回复）
    last_message = state["messages"][-1]
    
    # 检查回复中是否包含 "Final Answer" 标识
    # 这是我们在 execute_node 的 System Prompt 中要求的终止条件
    if "Final Answer" in last_message.content:
        # 如果包含，返回 END，表示任务完成，流程结束
        return END
    else:
        # 如果不包含，返回 "tool_node"，表示需要调用工具
        # 注意：这里假设 LLM 的回复一定包含 tool_calls
        # 实际应用中可能需要更复杂的判断逻辑
        return "tool_node"

from langgraph.graph import StateGraph, START, END
from langchain.messages import HumanMessage
from typing_extensions import TypedDict, Annotated

def build_agent():
    """
    构建 Plan-and-Execute Agent
    
    这个函数负责：
    1. 创建状态图（StateGraph）
    2. 添加所有节点
    3. 添加节点之间的边（普通边和条件边）
    4. 编译成可执行的 Agent
    
    Returns:
        编译后的 Agent，可以用于执行任务
    """
    
    # 创建状态图，指定状态类型为 PlanAndExecuteState
    # StateGraph 是 LangGraph 的核心类，用于构建有状态的工作流
    graph = StateGraph(PlanAndExecuteState)
    
    # 添加节点
    # add_node() 方法将节点函数注册到图中
    # 第一个参数是节点名称（字符串），第二个参数是节点函数
    graph.add_node("plan_node", plan_node)      # 计划节点
    graph.add_node("execute_node", execute_node)  # 执行节点
    graph.add_node("tool_node", tool_node)      # 工具节点
    
    # 添加普通边（无条件流转）
    # START 是 LangGraph 的特殊节点，表示工作流的起始点
    graph.add_edge(START, "plan_node")          # 从起始点进入计划节点
    graph.add_edge("plan_node", "execute_node")  # 计划节点执行完后进入执行节点
    
    # 添加条件边（根据条件决定流转方向）
    # add_conditional_edges() 用于添加条件分支
    graph.add_conditional_edges(
        source="execute_node",              # 源节点：执行节点
        path=conditional_edge,              # 条件判断函数
        path_map={                          # 路径映射：条件函数的返回值 -> 目标节点
            "tool_node": "tool_node",      # 如果返回 "tool_node"，进入工具节点
            END: END,                       # 如果返回 END，流程结束
        },
    )
    
    # 添加普通边：工具节点执行完后，回到执行节点继续执行
    # 这样就形成了一个循环：execute_node -> tool_node -> execute_node
    graph.add_edge("tool_node", "execute_node")
    
    # 编译 Agent
    # compile() 方法会将图编译成可执行的状态机
    # 编译后的 Agent 可以接收初始状态并执行整个工作流
    agent = graph.compile()
    
    # 返回编译后的 Agent
    return agent

def run_agent(agent, user_query: str) -> str:  
    """
    运行 Agent
    
    这个函数负责：
    1. 初始化 Agent 状态
    2. 执行 Agent 工作流
    3. 返回最终结果
    
    Args:
        agent: 编译后的 Agent 实例
        user_query: 用户的问题
    
    Returns:
        最终的回答文本
    """
    
    # 初始化状态
    # 创建用户消息，包含用户的问题
    messages = [HumanMessage(content=user_query)]
    
    # 创建初始状态对象
    # plan 初始化为空字符串，将在 plan_node 中生成
    # step 初始化为 0，表示还未开始执行
    # messages 包含用户的问题
    init_state = PlanAndExecuteState(
        plan="",           # 计划初始为空
        step=0,            # 步骤从 0 开始
        messages=messages, # 包含用户消息
    )
    
    # 运行 Agent
    # invoke() 方法会从 START 节点开始执行，直到遇到 END
    # 执行过程中，状态会在节点之间传递和更新
    result = agent.invoke(init_state)
    
    # 返回最终结果
    # 最后一条消息应该是包含 "Final Answer" 的回复
    return result["messages"][-1].content

if __name__ == "__main__":
    print("=" * 60)
    print("Plan-and-Execute Agent Demo")
    print("=" * 60)
    
    # 构建 Agent
    agent = build_agent()
    
    # 测试用例：汇率兑换问题
    user_query = "我想用100美元和50欧元兑换人民币，总共能兑换多少人民币？"
    print(f"\n【用户问题】\n{user_query}\n")
    print("-" * 60)
    
    # 运行 Agent
    result = run_agent(agent, user_query)
    
    print("\n" + "=" * 60)
    print("【最终答案】")
    print("=" * 60)
    print(result)
    print("=" * 60)

import os
import operator
from typing import Literal
from dotenv import load_dotenv

from langchain.tools import tool
from langchain.chat_models import init_chat_model
from langchain.messages import AnyMessage, SystemMessage, ToolMessage, HumanMessage
from typing_extensions import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

# 加载 .env 文件中的环境变量
load_dotenv()

# 配置自建模型（必须从 .env 文件中读取）
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

# 检查必需的配置项是否存在
if not BASE_URL:
    raise ValueError("OPENAI_BASE_URL 未在 .env 文件中配置")
if not API_KEY:
    raise ValueError("OPENAI_API_KEY 未在 .env 文件中配置")
if not MODEL_NAME:
    raise ValueError("OPENAI_MODEL 未在 .env 文件中配置")

model = init_chat_model(
    model=MODEL_NAME,  # 自建模型直接使用模型名称，不需要 openai: 前缀
    model_provider="openai",
    temperature=0.7,
    base_url=BASE_URL,
    api_key=API_KEY,
)


# 定义工具
@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a * b


@tool
def add(a: int, b: int) -> int:
    """Adds `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """Divide `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a / b


# 为LLM添加工具
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)


# 步骤2：定义状态
class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int


# 步骤3：定义模型节点
def llm_call(state: dict):
    """LLM decides whether to call a tool or not"""
    return {
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
                    )
                ]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }


# 步骤4：定义工具节点
def tool_node(state: dict):
    """Performs the tool call"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=str(observation), tool_call_id=tool_call["id"]))
    return {"messages": result}


# 步骤5：定义结束逻辑
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""
    messages = state["messages"]
    last_message = messages[-1]

    # If the LLM makes a tool call, then perform an action
    if last_message.tool_calls:
        return "tool_node"

    # Otherwise, we stop (reply to the user)
    return END


# 步骤6：构建并编译代理
# 构建工作流
agent_builder = StateGraph(MessagesState)

# 添加节点
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# 添加边以连接节点
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tool_node", "llm_call")

# 编译代理
agent = agent_builder.compile()


# 示例用法
if __name__ == "__main__":
    print("=== LangGraph Calculator Agent Demo ===\n")
    
    # 测试用例1：加法
    print("Test 1: Add 3 and 4")
    messages = [HumanMessage(content="Add 3 and 4.")]
    result = agent.invoke({"messages": messages})
    print(f"LLM calls: {result.get('llm_calls', 0)}")
    for m in result["messages"]:
        print(f"- {m.__class__.__name__}: {m.content}")
    print("\n")
    
    # 测试用例2：乘法
    print("Test 2: Multiply 5 and 6")
    messages = [HumanMessage(content="Multiply 5 and 6.")]
    result = agent.invoke({"messages": messages})
    print(f"LLM calls: {result.get('llm_calls', 0)}")
    for m in result["messages"]:
        print(f"- {m.__class__.__name__}: {m.content}")
    print("\n")
    
    # 测试用例3：除法
    print("Test 3: Divide 20 by 4")
    messages = [HumanMessage(content="Divide 20 by 4.")]
    result = agent.invoke({"messages": messages})
    print(f"LLM calls: {result.get('llm_calls', 0)}")
    for m in result["messages"]:
        print(f"- {m.__class__.__name__}: {m.content}")
    print("\n")
    
    # 测试用例4：复杂计算
    print("Test 4: Add 10 and 5, then multiply the result by 3")
    messages = [HumanMessage(content="Add 10 and 5, then multiply the result by 3.")]
    result = agent.invoke({"messages": messages})
    print(f"LLM calls: {result.get('llm_calls', 0)}")
    for m in result["messages"]:
        print(f"- {m.__class__.__name__}: {m.content}")

load_dotenv()  # 加载 .env 文件中的环境变量
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`."""
    return a * b

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

def llm_call(state: dict):
    return {
        "messages": [
            model_with_tools.invoke([
                SystemMessage(content="...")] + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

def tool_node(state: dict):
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(...))
    return {"messages": result}

def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tool_node"  # 有工具调用，去执行工具
    return END  # 没有工具调用，结束流程

agent_builder = StateGraph(MessagesState)
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
agent_builder.add_edge("tool_node", "llm_call")
agent = agent_builder.compile()

load_dotenv()  # 加载 .env 文件中的环境变量
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`."""
    return a * b

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

def llm_call(state: dict):
    return {
        "messages": [
            model_with_tools.invoke([
                SystemMessage(content="...")] + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

def tool_node(state: dict):
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(...))
    return {"messages": result}

def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tool_node"  # 有工具调用，去执行工具
    return END  # 没有工具调用，结束流程

agent_builder = StateGraph(MessagesState)
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
agent_builder.add_edge("tool_node", "llm_call")
agent = agent_builder.compile()