智能体系统快速搭建案例先行，通俗易懂

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing import TypedDict

class State(TypedDict):
    question: str
    answer: str

def answer_node(state: State):
    llm = ChatOpenAI(model="gpt-5-nano")
    response = llm.invoke(state["question"])
    return {"answer": response.content}

graph = StateGraph(State)
graph.add_node("answer_node", answer_node)  # ✅ 改名
graph.add_edge(START, "answer_node")
graph.add_edge("answer_node", END)

app = graph.compile()
result = app.invoke({"question": "什么是维基百科？一句话回答"})
print(result["answer"])

# ---------- 可视化 ----------
# 使用 Mermaid 语法画出图，并以图片形式显示
# 注意：需要在 app.compile() 之后才能获取图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

# 从 langgraph.graph 模块中导入 StateGraph、START、END
# StateGraph 是 LangGraph 中用于构建“状态图”的核心类
# START 和 END 分别代表图的起点与终点，用于定义流程的开始与结束
from langgraph.graph import StateGraph, START, END

# 从 langchain_openai 导入 ChatOpenAI
# 这是一个封装好的 OpenAI 接口类，可以直接调用 GPT 模型
from langchain_openai import ChatOpenAI

# 从 typing 模块导入 TypedDict，用于定义强类型字典（类似结构体）
## 另外一个常用的是类型定义工具是 pydantic.BaseModel
from typing import TypedDict

# 定义一个 State 类型，用来描述每个节点处理的数据结构
# 在这里，每个“状态”（state）是一个字典，包含两个字段：
# question（问题）和 answer（回答）
class State(TypedDict):
    question: str  # 用户输入的问题
    answer: str    # 模型生成的答案
    ## 简单结构，只能保存最新一轮对话的内容


# 定义一个节点函数 answer_node，用于处理输入状态并生成输出状态
def answer_node(state: State): # 这里 State 是类的名字，state 是一个实例化的变量名字
    # 创建一个 ChatOpenAI 实例，指定使用的模型是 gpt-5-nano
    llm = ChatOpenAI(model="gpt-5-nano")
    
    # 调用模型的 invoke 方法，将当前状态中的问题作为输入
    response = llm.invoke(state["question"])
    
    # 返回一个新的状态字典，只包含 "answer" 字段
    # response.content 是模型生成的文本内容
    return {"answer": response.content}


# 创建一个状态图对象，并指定状态类型为上面定义的 State
graph = StateGraph(State)

# 向图中添加一个节点，名称为 "answer_node"，对应的执行函数是 answer_node
graph.add_node("answer_node", answer_node)

# 添加边：定义图中节点之间的执行顺序
# 从 START 节点（起点）连到 "answer_node"
graph.add_edge(START, "answer_node")

# 从 "answer_node" 节点连到 END（终点）
graph.add_edge("answer_node", END)

# 编译图，将其转换为可执行的“应用”对象
app = graph.compile()

# 调用应用，传入初始状态（包含一个问题）
result = app.invoke({"question": "什么是维基百科？一句话回答"})

# 打印出模型生成的回答内容
print(result["answer"])


# 传入一个新的问题。AI 无法记忆上一个问题，因为我们这个简单图没有记忆机制。
# 后面我们会详细介绍 LangGraph 的记忆机制
result = app.invoke({"question": "刚才我问了个什么问题？"})
# 打印出模型生成的回答内容
print()
print("第二个问题的回答： ")
print(result["answer"])

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI

# 定义工具 1：搜索工具
@tool
def search_web(query: str) -> str:
    """搜索网络信息。当需要查找最新信息或事实时使用此工具。"""
    # 模拟搜索结果
    search_results = {
        "天气": "今天北京天气晴朗，温度 20°C",
        "新闻": "AI 技术持续突破，LangGraph 1.0 正式发布",
        "股票": "科技股今日上涨 2.5%"
    }
    for key in search_results:
        if key in query:
            return search_results[key]
    return f"搜索结果：关于 '{query}' 的最新信息"

# 定义工具 2：计算器
@tool
def calculator(expression: str) -> str:
    """执行数学计算。输入格式：数字1 运算符 数字2，例如：'123 * 456'"""
    try:
        # 安全的计算方法：手动解析表达式
        parts = expression.strip().split()
        if len(parts) != 3:
            return "错误：请使用格式 '数字1 运算符 数字2'，如 '123 * 456'"

        num1, operator, num2 = parts
        num1, num2 = float(num1), float(num2)

        if operator == '+':
            result = num1 + num2
        elif operator == '-':
            result = num1 - num2
        elif operator == '*':
            result = num1 * num2
        elif operator == '/':
            result = num1 / num2 if num2 != 0 else "错误：除数不能为零"
        else:
            return f"错误：不支持的运算符 '{operator}'，仅支持 +, -, *, /"

        return f"计算结果：{expression} = {result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

# 定义工具 3：天气查询
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气信息。"""
    weather_data = {
        "北京": "晴朗，20°C，空气质量良好",
        "上海": "多云，18°C，有轻微雾霾",
        "深圳": "阴天，25°C，湿度较高"
    }
    return weather_data.get(city, f"{city}：暂无天气数据")

# 创建 Agent（LangChain 1.0 新 API）
agent = create_agent(
    model=ChatOpenAI(model="gpt-5-nano", temperature=0),
    tools=[search_web, calculator, get_weather],
    system_prompt="你是一个智能助手，可以使用工具帮助用户解决问题。请先思考需要使用哪些工具，然后逐步执行。"
)

# 执行任务：需要多次工具调用
result = agent.invoke({
    "messages": [{"role": "user", "content": "北京今天天气怎么样？顺便帮我算一下 123 * 456 等于多少？"}]
})

print("=" * 50)
print("Agent 最终回答：")
print(result["messages"][-1].content)
print("=" * 50)

# 查看完整的推理过程（包括工具调用）
print("\n完整对话历史：")
for msg in result["messages"]:
    if hasattr(msg, 'content') and msg.content:
        print(f"\n[{msg.__class__.__name__}]: {msg.content}")
    elif hasattr(msg, 'tool_calls') and msg.tool_calls:
        print(f"\n[调用工具]: {[tc['name'] for tc in msg.tool_calls]}")

# ==============================================================
# ReAct Agent 联系人信息提取（无 stop 参数 + 稳定输出）
# ==============================================================
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
import json

# --------------------------------------------------------------
# 一、定义输出模型
# --------------------------------------------------------------
class ContactInfo(BaseModel):
    name: str = Field(description="联系人姓名")
    email: str = Field(description="电子邮件地址")
    phone: str = Field(description="电话号码")
    company: str = Field(description="公司名称")
    completed: bool = Field(description="是否提取完成", default=False)

# --------------------------------------------------------------
# 二、定义工具
# --------------------------------------------------------------
@tool
def extract_contact(text: str) -> str:
    """从文本中提取联系人信息"""
    result = {}
    if "张三" in text:
        result["name"] = "张三"
    if "zhangsan@example.com" in text:
        result["email"] = "zhangsan@example.com"
    if "138-0000-0001" in text:
        result["phone"] = "138-0000-0001"
    if "科技公司" in text:
        result["company"] = "科技公司"
    status = "已完成" if len(result) == 4 else "部分完成"
    return json.dumps({"partial_result": result, "status": status})

@tool
def check_completion(result_text: str) -> str:
    """检查联系人信息是否完整"""
    try:
        data = json.loads(result_text)
        res = data.get("partial_result", {})
        if all(k in res for k in ["name", "email", "phone", "company"]):
            return "完成"
        return "未完成"
    except Exception:
        return "未完成"

# --------------------------------------------------------------
# 三、创建 Agent
# --------------------------------------------------------------
llm = ChatOpenAI(model="gpt-5-nano", temperature=0)

agent_contact = create_agent(
    model=llm,
    tools=[extract_contact, check_completion],
    system_prompt=(
        "你是一个联系人信息提取助手。\n"
        "任务：提取文本中的联系人信息（name, email, phone, company）。\n"
        "执行逻辑：\n"
        "1️⃣ 调用 extract_contact 提取信息。\n"
        "2️⃣ 调用 check_completion 检查是否完成。\n"
        "3️⃣ 如果返回“完成”，输出最终 JSON："
        "{\"name\":..., \"email\":..., \"phone\":..., \"company\":..., \"completed\": true}\n"
        "4️⃣ 如果三次检查后仍未完成，输出当前最完整结果，completed=false。\n"
        "不要调用工具超过 3 次。输出后立即结束。"
    )
)

# --------------------------------------------------------------
# 四、执行调用
# --------------------------------------------------------------
user_input = {
    "messages": [
        {
            "role": "user",
            "content": (
                "请帮我提取这段文本的联系人信息："
                "张三是我们公司的技术总监，他的邮箱是 zhangsan@example.com，"
                "电话是 138-0000-0001，在科技公司工作。"
            ),
        }
    ]
}

result = agent_contact.invoke(user_input, config={"recursion_limit": 10})

# --------------------------------------------------------------
# 五、解析输出
# --------------------------------------------------------------
# 👇 获取最后一条模型输出消息
final_message = result["messages"][-1].content

print("\n🧾 原始输出：")
print(final_message)

# 尝试解析为 JSON
try:
    structured = json.loads(final_message)
    contact = ContactInfo(**structured)
    print("\n✅ 提取结果（结构化）：")
    print(f"  姓名：{contact.name}")
    print(f"  邮箱：{contact.email}")
    print(f"  电话：{contact.phone}")
    print(f"  公司：{contact.company}")
    print(f"  是否完成：{contact.completed}")
except Exception:
    print("\n⚠️ 模型输出不是标准 JSON，请检查：")
    print(final_message)