AI入门实战|Agent骨架与工具调用|Planning规划能力

from dotenv import load_dotenv
import os
from typing import List, TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# 仅保留必须的图组件，删除RetryPolicy废弃导入
from langgraph.graph import StateGraph, START, END

# 加载环境变量
load_dotenv()

# ====================== 1. 自定义工具 ======================
@tool
def calculator(expression: str) -> str:
    """数学四则运算，示例：(10+20)*3"""
    try:
        return f"计算结果：{eval(expression)}"
    except Exception as e:
        return f"计算异常：{str(e)}"

@tool
def get_current_time() -> str:
    """获取当前系统北京时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

tools = [calculator, get_current_time]

# ====================== 2. 定义全局状态（存储任务、步骤、执行记录） ======================
class PlanExecuteState(TypedDict):
    input: str                  # 用户原始复杂任务
    plan: Annotated[List[str], "拆解后的有序步骤列表"]
    completed_steps: Annotated[List[str], "已执行步骤+结果"]
    final_answer: str           # 汇总后的最终回答

# ====================== 3. 初始化大模型并绑定工具 ======================
llm = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    model="deepseek-chat",
    temperature=0
)
llm_with_tools = llm.bind_tools(tools)

# ====================== 4. 规划节点：复杂任务自动拆解步骤 ======================
def plan_task_node(state: PlanExecuteState) -> dict:
    plan_prompt = f"""
你是专业任务规划助手，将用户复杂需求拆解为**有序、可独立执行**的步骤清单。
仅输出步骤，不要多余文字、不要解释。
用户任务：{state['input']}
"""
    resp = llm.invoke(plan_prompt)
    # 按换行切割步骤，过滤空行
    step_list = [line.strip() for line in resp.content.split("\n") if line.strip()]
    print("【自动拆解任务规划】")
    for idx, step in enumerate(step_list, 1):
        print(f"{idx}. {step}")
    return {"plan": step_list, "completed_steps": []}

# ====================== 5. 执行节点：执行单一步骤、调用工具 ======================
def execute_single_step_node(state: PlanExecuteState) -> dict:
    current_step = state["plan"][0]
    print(f"\n【正在执行步骤】{current_step}")
    # 让模型判断是否需要调用工具完成当前步骤
    messages = [("user", current_step)]
    resp = llm_with_tools.invoke(messages)

    step_result = ""
    # 处理工具调用
    if resp.tool_calls:
        for call_info in resp.tool_calls:
            match_tool = next(t for t in tools if t.name == call_info["name"])
            tool_output = match_tool.invoke(call_info["args"])
            step_result = tool_output
    else:
        # 无需工具，直接文本回答
        step_result = resp.content

    # 记录完成步骤，移除当前已执行步骤
    new_completed = state["completed_steps"] + [f"{current_step} → {step_result}"]
    remaining_plan = state["plan"][1:]
    return {"completed_steps": new_completed, "plan": remaining_plan}

# ====================== 6. 汇总节点：整合所有步骤结果生成最终回答 ======================
def summarize_all_node(state: PlanExecuteState) -> dict:
    summary_prompt = f"""
根据用户原始任务和全部执行记录，整理一段通顺完整的最终回答。
用户任务：{state['input']}
执行记录：{state['completed_steps']}
"""
    final_text = llm.invoke(summary_prompt).content
    return {"final_answer": final_text}

# ====================== 7. 搭建状态工作流图 ======================
workflow = StateGraph(PlanExecuteState)
# 注册三个核心节点
workflow.add_node("task_planner", plan_task_node)
workflow.add_node("step_executor", execute_single_step_node)
workflow.add_node("result_summarizer", summarize_all_node)

# 固定流转逻辑
workflow.add_edge(START, "task_planner")
workflow.add_edge("task_planner", "step_executor")

# 条件分支：还有剩余步骤则继续执行，无步骤则汇总结果
def judge_continue(state: PlanExecuteState):
    if len(state["plan"]) > 0:
        return "step_executor"
    else:
        return "result_summarizer"

workflow.add_conditional_edges("step_executor", judge_continue)
workflow.add_edge("result_summarizer", END)

# 编译生成可执行图
agent_graph = workflow.compile()

# ====================== 8. 交互入口 ======================
if __name__ == "__main__":
    complex_task = "先计算 (100+20)*3，再获取当前系统时间，最后把两个结果合并成一段通顺描述"
    output = agent_graph.invoke({"input": complex_task})
    print("\n========== 最终完整回答 ==========")
    print(output["final_answer"])

from dotenv import load_dotenv
import os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# 唯一官方稳定ReAct入口
from langgraph.prebuilt import create_react_agent

load_dotenv()

# 自定义工具
@tool
def calculator(expression: str) -> str:
    """数学四则运算计算器"""
    try:
        return f"结果：{eval(expression)}"
    except Exception as e:
        return f"计算错误：{e}"

@tool
def get_current_time() -> str:
    """获取当前北京时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

tools = [calculator, get_current_time]

# 初始化模型
llm = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    model="deepseek-chat",
    temperature=0
)

# 新版标准ReAct智能体（动态边执行边规划）
react_agent = create_react_agent(model=llm, tools=tools)

if __name__ == "__main__":
    user_task = "计算 (30+70)*2，然后查看当前系统时间"
    res = react_agent.invoke({"messages": [("user", user_task)]})
    print("Agent最终回答：")
    print(res["messages"][-1].content)