AI入门实战|Agent骨架与工具调用|LangGraph流程编排

from dotenv import load_dotenv
import os
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# 官方唯一稳定预构建Agent入口
from langgraph.prebuilt import create_react_agent

# 加载环境变量
load_dotenv()

# ---------------------- 1. 定义工具（Agent扩展能力） ----------------------
@tool
def calculator(expression: str) -> str:
    """
    数学四则运算工具
    参数expression：数学表达式，支持括号，示例：(10+20)*3
    """
    try:
        return f"计算结果：{eval(expression)}"
    except Exception as err:
        return f"计算失败：{str(err)}"

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

# 工具集合，全部注入Agent
tool_list = [calculator, get_current_time]

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

# ---------------------- 3. 生成完整Agent骨架 ----------------------
# 内部自动封装：状态管理 + 流程循环 + 工具解析执行
agent = create_react_agent(model=llm, tools=tool_list)

# ---------------------- 4. 交互测试 ----------------------
if __name__ == "__main__":
    print("===== ReAct入门Agent｜工具调用演示 =====")
    print("输入exit退出对话\n")
    while True:
        user_input = input("你：").strip()
        if user_input.lower() == "exit":
            print("对话结束")
            break
        # 调用Agent，传入用户消息
        response = agent.invoke({"messages": [("user", user_input)]})
        # 提取最终回答
        final_answer = response["messages"][-1].content
        print(f"助手：{final_answer}\n")

from dotenv import load_dotenv
import os
from typing import List, TypedDict, Annotated
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END

load_dotenv()

# ===================== 一、定义工具（Agent外部能力） =====================
@tool
def calculator(expression: str) -> str:
    """四则运算计算器，支持括号混合运算"""
    try:
        return f"计算结果：{eval(expression)}"
    except Exception as e:
        return f"计算错误：{str(e)}"

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

tools = [calculator, get_current_time]

# ===================== 二、定义全局状态State（Agent内存骨架） =====================
# TypedDict：统一存储整个流程所有数据，所有节点共享
class AgentState(TypedDict):
    user_task: str                      # 用户原始任务
    task_plan: Annotated[List[str], "拆解后的步骤列表"]
    finished_records: Annotated[List[str], "已执行步骤与结果"]
    final_output: str                   # 最终汇总回答

# ===================== 三、初始化大模型 =====================
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)

# ===================== 四、定义流程节点（Graph最小执行单元） =====================
# 节点1：规划节点 —— 复杂任务自动拆解步骤（任务规划能力）
def plan_node(state: AgentState) -> dict:
    prompt = f"""
你是任务规划专家，将用户复杂需求拆分为有序、可执行的步骤，仅输出步骤，无多余文字。
用户需求：{state['user_task']}
"""
    res = llm.invoke(prompt)
    step_list = [line.strip() for line in res.content.split("\n") if line.strip()]
    print("【任务自动拆解规划】")
    for i, step in enumerate(step_list, 1):
        print(f"{i}. {step}")
    return {"task_plan": step_list, "finished_records": []}

# 节点2：执行节点 —— 执行单一步骤、调用对应工具
def execute_node(state: AgentState) -> dict:
    current_step = state["task_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 in resp.tool_calls:
            match_tool = next(t for t in tools if t.name == call["name"])
            step_result = match_tool.invoke(call["args"])
    else:
        step_result = resp.content
    # 更新状态：已完成步骤入库，移除当前执行步骤
    new_finish = state["finished_records"] + [f"{current_step} → {step_result}"]
    remaining_plan = state["task_plan"][1:]
    return {"finished_records": new_finish, "task_plan": remaining_plan}

# 节点3：汇总节点 —— 整合所有步骤结果，生成最终回答
def summary_node(state: AgentState) -> dict:
    prompt = f"""
根据用户原始任务和全部执行记录，生成通顺完整的最终回答。
用户任务：{state['user_task']}
执行记录：{state['finished_records']}
"""
    answer = llm.invoke(prompt).content
    return {"final_output": answer}

# ===================== 五、流程编排：搭建状态图（LangGraph核心） =====================
# 1. 创建状态图，绑定自定义状态
graph = StateGraph(AgentState)
# 2. 注册全部自定义节点
graph.add_node("task_planner", plan_node)
graph.add_node("step_executor", execute_node)
graph.add_node("result_summary", summary_node)

# 3. 配置流转边，定义执行顺序
graph.add_edge(START, "task_planner")          # 程序入口 → 任务规划
graph.add_edge("task_planner", "step_executor")# 规划完成 → 执行第一步

# 4. 条件分支：判断是否还有剩余步骤，实现循环执行
def judge_loop(state: AgentState):
    if len(state["task_plan"]) > 0:
        # 还有步骤，继续执行
        return "step_executor"
    else:
        # 全部执行完毕，进入汇总
        return "result_summary"

graph.add_conditional_edges("step_executor", judge_loop)
graph.add_edge("result_summary", END) # 汇总完成，流程结束

# 5. 编译生成可运行Agent
agent_workflow = graph.compile()

# ===================== 六、运行入口 =====================
if __name__ == "__main__":
    # 复合多步骤任务，测试自动拆解能力
    task = "先计算 (100+20)*3，再获取当前系统时间，最后整合两段信息形成完整描述"
    result = agent_workflow.invoke({"user_task": task})
    print("\n========== Agent最终回答 ==========")
    print(result["final_output"])