flowchart LR
    subgraph 开发环境
        Python[Python 3.11+]
        UV[uv 包管理器]
        VE[虚拟环境]
    end
    subgraph 框架层
        LC[LangChain]
        LG[LangGraph]
    end
    subgraph 模型层
        OpenAI[OpenAI API]
        DS[DeepSeek API]
        Qwen[通义千问 API]
    end
    VE --> LC
    VE --> LG
    LC --> OpenAI
    LC --> DS
    LC --> Qwen

# 安装 uv（macOS / Linux）
curl -LsSf https://astral.sh/uv/install.sh | sh

# 创建新项目
uv init langchain-agent-book
cd langchain-agent-book

# 创建虚拟环境并安装依赖
uv venv
source .venv/bin/activate  # macOS / Linux

# 安装核心依赖
uv add langchain langchain-openai langgraph

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph

# 检查版本
import langchain
import langgraph

print(f"LangChain 版本: {langchain.__version__}")
print(f"LangGraph 版本: {langgraph.__version__}")

# 测试模型调用
model = ChatOpenAI(model="gpt-4o-mini")
response = model.invoke("你好，世界！")
print(response.content)

# 方式 1：环境变量
export OPENAI_API_KEY="sk-your-key-here"

# 方式 2：.env 文件（推荐）
echo "OPENAI_API_KEY=sk-your-key-here" > .env

# 自动加载 .env 文件
from dotenv import load_dotenv
load_dotenv()

# 加载后可以直接使用环境变量
import os
api_key = os.getenv("OPENAI_API_KEY")

flowchart LR
    Input[输入文本] --> Tokenizer[分词器]
    Tokenizer --> Tokens[Token 序列]
    Tokens --> LLM[LLM 模型]
    LLM --> NextToken[预测下一个 Token]
    NextToken --> Append[追加到序列]
    Append --> LLM
    Append --> Output[完整输出]

    style NextToken fill:#f96

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

# 获取 Token 计数
text = "你好，世界！Hello, World!"
tokens = model.get_num_tokens(text)
print(f"Token 数: {tokens}")

# 获取消息的 Token 计数
from langchain_core.messages import HumanMessage, SystemMessage

messages = [
    SystemMessage("你是一个 AI 助手"),
    HumanMessage("给我讲个笑话"),
]
token_count = model.get_num_tokens_from_messages(messages)
print(f"消息总 Token 数: {token_count}")

from langchain_core.messages import SystemMessage, HumanMessage

messages = [
    SystemMessage("你是一个专业的数据分析师。你只使用中文回答。"),
    HumanMessage("分析一下这段数据：销售额 100万，成本 60万，利润 40万"),
]

response = model.invoke(messages)
print(response.content)

# 温度控制输出的随机性
# temperature=0: 最确定，每次输出相同
# temperature=0.7: 平衡创造力和确定性
# temperature=1.5: 高随机性，适合创意写作

creative_model = ChatOpenAI(model="gpt-4o-mini", temperature=0.9)
precise_model = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

# 创意模式：每次都不一样
for i in range(3):
    print(creative_model.invoke("写一句诗").content)

# 精确模式：每次都一样
for i in range(3):
    print(precise_model.invoke("2+2等于多少").content)

flowchart TD
    subgraph 用户层
        UI[用户界面]
    end
    subgraph 框架层
        Agent[Agent]
        Prompt[Prompt 模板]
        Model[Chat Model]
        Parser[输出解析器]
        Tool[工具系统]
        Memory[记忆系统]
    end
    subgraph 外部
        LLM[LLM API]
        API[外部 API]
        DB[数据库]
    end

    UI --> Agent
    Agent --> Prompt
    Prompt --> Model
    Model --> Parser
    Agent --> Tool
    Agent --> Memory
    Tool --> API
    Tool --> DB
    Model --> LLM

# 所有 Chat Model 使用相同的接口
from langchain_openai import ChatOpenAI
from langchain_deepseek import ChatDeepSeek
from langchain_community.chat_models import ChatTongyi

# 创建模型（接口完全一致）
openai_model = ChatOpenAI(model="gpt-4o-mini")
deepseek_model = ChatDeepSeek(model="deepseek-chat")
tongyi_model = ChatTongyi(model="qwen-plus")

# 调用方法完全一样
response = openai_model.invoke("你好")
response = deepseek_model.invoke("你好")
response = tongyi_model.invoke("你好")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. 定义模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 {role}，请用 {language} 回答"),
    ("user", "{question}"),
])

# 2. 创建模型
model = ChatOpenAI(model="gpt-4o-mini")

# 3. 创建解析器
parser = StrOutputParser()

# 4. 串联：模板 -> 模型 -> 解析器
chain = prompt | model | parser

# 5. 调用
result = chain.invoke({
    "role": "历史老师",
    "language": "中文",
    "question": "解释什么是文艺复兴",
})
print(result)

# LCEL 是 LangChain 的声明式管道语法
# | 操作符把组件串联起来，数据自动从前一个传到后一个

chain = prompt | model | parser

# 并行执行
from langchain_core.runnables import RunnableParallel

parallel_chain = RunnableParallel(
    answer=prompt | model | parser,
    length=lambda x: len(x["question"]),
)

result = parallel_chain.invoke({
    "role": "老师",
    "language": "中文",
    "question": "什么是 AI？",
})
print(result)
# {'answer': '...', 'length': 9}

chain = prompt | model | parser  # 构建管道，不执行
result = chain.invoke(inputs)     # 这行才真正执行

flowchart LR
    subgraph 你的代码
        API["model.invoke(messages)"]
    end
    subgraph ChatModel
        Auth[认证模块]
        Retry[重试逻辑]
        Format[请求格式化]
        Parse[响应解析]
    end
    subgraph 模型 API
        O[OpenAI]
        DS[DeepSeek]
        QW[通义千问]
    end

    API --> Auth
    Auth --> Retry
    Retry --> Format
    Format --> O
    Format --> DS
    Format --> QW
    O --> Parse
    DS --> Parse
    QW --> Parse
    Parse --> Response["response.content"]

from langchain_openai import ChatOpenAI

# 最基本的用法
model = ChatOpenAI(model="gpt-4o-mini")
response = model.invoke("谁是世界上最聪明的人？")
print(response.content)

# 带消息列表的调用
from langchain_core.messages import HumanMessage, SystemMessage

messages = [
    SystemMessage("你是苏格拉底，用提问的方式回答问题。"),
    HumanMessage("人生的意义是什么？"),
]
response = model.invoke(messages)

from langchain_core.tools import tool

# 定义工具
@tool
def get_weather(city: str) -> str:
    """查询城市的天气"""
    return f"{city} 今天 25°C，晴"

# 把工具绑定到模型
model_with_tools = model.bind_tools([get_weather])

# 现在当模型认为需要天气信息时，会自动请求调用工具
response = model_with_tools.invoke("北京天气怎么样？")
print(response.tool_calls)
# [{'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'call_xxx'}]

from pydantic import BaseModel

# 定义输出结构
class MovieReview(BaseModel):
    title: str
    rating: int
    summary: str
    pros: list[str]
    cons: list[str]

# 使用 with_structured_output 绑定结构
structured_model = model.with_structured_output(MovieReview)

# 模型会直接返回 Pydantic 对象
review = structured_model.invoke("评价一下电影《盗梦空间》")
print(f"评分: {review.rating}/10")
print(f"优点: {review.pros}")

# 流式输出：一个字一个字地接收
for chunk in model.stream("给我讲一个关于 AI 的故事"):
    print(chunk.content, end="", flush=True)

# 异步调用：不阻塞主线程
import asyncio

async def main():
    model = ChatOpenAI(model="gpt-4o-mini")

    # 并发调用多个请求
    tasks = [
        model.ainvoke("写一首关于春天的诗"),
        model.ainvoke("讲一个笑话"),
        model.ainvoke("推荐三本书"),
    ]

    results = await asyncio.gather(*tasks)
    for r in results:
        print(r.content[:50] + "...")
        print("---")

asyncio.run(main())

flowchart TD
    subgraph 模板系统
        PT[PromptTemplate<br/>单条消息]
        CT[ChatPromptTemplate<br/>多条消息]
        MP[MessagesPlaceholder<br/>动态消息插入]
    end
    subgraph 消息结构
        S[System Prompt<br/>系统指令]
        H[Human Message<br/>用户输入]
        AI[AI Message<br/>模型回复]
    end

    PT -->|参数化| S
    CT -->|组合| S
    CT -->|组合| H
    MP -->|插入历史| H

    S --> Model
    H --> Model

from langchain_core.prompts import PromptTemplate

# 定义模板
template = PromptTemplate.from_template(
    "请用 {language} 回答以下问题：{question}"
)

# 填充参数（方式 1）
prompt = template.format(language="中文", question="什么是量子计算？")
print(prompt)

# 填充参数（方式 2）
prompt = template.invoke({"language": "中文", "question": "什么是量子计算？"})
print(prompt.text)

from langchain_core.prompts import ChatPromptTemplate

# 方式 1：使用元组列表
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是 {role}，用 {language} 回答"),
    ("user", "{question}"),
])

formatted = prompt.invoke({
    "role": "物理学家",
    "language": "中文",
    "question": "解释相对论",
})

# 查看生成的消息列表
for msg in formatted.to_messages():
    print(f"{msg.type}: {msg.content}")

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# MessagesPlaceholder 用于在固定位置插入动态消息列表
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 AI 助手，请用中文回答"),
    MessagesPlaceholder(variable_name="history"),  # 对话历史插入到这里
    ("user", "{input}"),
])

# 使用时传入 history 参数
formatted = prompt.invoke({
    "history": [
        ("user", "你好"),
        ("assistant", "你好！有什么可以帮助你的？"),
    ],
    "input": "我刚才说了什么？",
})

for msg in formatted.to_messages():
    print(f"{msg.type}: {msg.content[:50]}")

from langchain_core.prompts import FewShotChatMessagePromptTemplate

# 示例数据
examples = [
    {"input": "2+2=?", "output": "4"},
    {"input": "3*5=?", "output": "15"},
    {"input": "10/2=?", "output": "5"},
]

# 示例模板
example_prompt = ChatPromptTemplate.from_messages([
    ("user", "{input}"),
    ("assistant", "{output}"),
])

# 创建 Few-shot 模板
few_shot_prompt = FewShotChatMessagePromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
)

# 最终的完整 prompt
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个数学计算助手"),
    few_shot_prompt,
    ("user", "{input}"),
])

result = final_prompt.invoke({"input": "8*7=?"})
for msg in result.to_messages():
    print(f"{msg.type}: {msg.content}")

# 部分填充
template = PromptTemplate.from_template("{greeting}，{name}")
partial = template.partial(greeting="你好")
result = partial.invoke({"name": "小明"})
print(result.text)  # "你好，小明"

flowchart LR
    LLM[LLM<br/>输出文本] --> Parser[输出解析器]
    Parser --> Str[StrOutputParser --> str]
    Parser --> Json[JsonOutputParser --> dict]
    Parser --> Pyd[PydanticOutputParser --> BaseModel]
    Parser --> List[ListOutputParser --> list]

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# 直接调用模型返回的是 AIMessage
response = model.invoke("你好")
print(type(response))  # <class 'langchain_core.messages.AIMessage'>

# 经过解析器后得到纯文本
text = parser.invoke(response)
print(type(text))  # <class 'str'>

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate

parser = JsonOutputParser()

prompt = PromptTemplate.from_template(
    "提取以下文本中的信息，输出 JSON 格式：\n{input}\n\n{format_instructions}",
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | model | parser

result = chain.invoke({"input": "张三，28岁，住在北京，是一名工程师"})
print(result)
# {'name': '张三', 'age': 28, 'city': '北京', 'job': '工程师'}

from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser

# 定义输出结构
class Person(BaseModel):
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    city: str = Field(description="所在城市")
    job: str = Field(description="职业")

parser = PydanticOutputParser(pydantic_object=Person)

prompt = PromptTemplate.from_template(
    "从以下文本中提取人物信息：\n{input}\n\n{format_instructions}",
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | model | parser

person = chain.invoke({"input": "李四，35岁，上海，产品经理"})
print(f"姓名: {person.name}")
print(f"年龄: {person.age}")
print(type(person))  # <class '__main__.Person'>

from langchain.output_parsers import OutputFixingParser

# 如果解析失败，OutputFixingParser 会自动让模型修正输出
fixing_parser = OutputFixingParser.from_llm(
    parser=parser,
    llm=ChatOpenAI(model="gpt-4o-mini"),
)

# 即使原始输出格式有误，也能自动修复
result = fixing_parser.parse('{"name": "王五", "age": "二十八", "city": "广州", "job": "设计师"}')

sequenceDiagram
    participant User as 用户
    participant LLM as LLM
    participant Tool as 工具系统
    participant API as 外部 API

    User->>LLM: "北京天气怎么样？"
    Note over LLM: 我需要天气信息
    LLM->>User: tool_calls=[get_weather(city="北京")]
    User->>Tool: 执行 get_weather("北京")
    Tool->>API: 调用天气 API
    API->>Tool: 返回数据
    Tool->>LLM: ToolMessage("北京 22°C 晴")
    Note over LLM: 基于工具结果生成回答
    LLM->>User: "北京今天 22°C，天气晴朗"

from langchain_core.tools import tool

# 最简单的工具定义
@tool
def get_current_time() -> str:
    """返回当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 带参数的工具
@tool
def search_web(query: str) -> str:
    """搜索网络信息（模拟）"""
    return f"关于 '{query}' 的搜索结果..."

# 复杂参数的工具
@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送电子邮件"""
    return f"邮件已发送至 {to}，主题：{subject}"

# 查看工具的信息
print(search_web.name)        # "search_web"
print(search_web.description) # "搜索网络信息（模拟）"
print(search_web.args)        # {"query": {"title": "Query", "type": "string"}}

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

# 绑定多个工具
tools = [get_current_time, search_web, send_email]
model_with_tools = model.bind_tools(tools)

# 当模型认为需要工具时，会返回 tool_calls
response = model_with_tools.invoke("现在几点了？")
if response.tool_calls:
    print(response.tool_calls)
    # [{'name': 'get_current_time', 'args': {}, 'id': 'call_xxx'}]

# 如果有 tool_calls，执行对应的工具
def execute_tool_calls(response, tools_map: dict):
    """执行模型请求的工具调用"""
    messages = []
    for tc in response.tool_calls:
        tool_name = tc["name"]
        tool_args = tc["args"]
        tool_id = tc["id"]

        tool_fn = tools_map[tool_name]
        result = tool_fn.invoke(tool_args)

        messages.append(ToolMessage(content=result, tool_call_id=tool_id))

    return messages

from langgraph.prebuilt import ToolNode

# ToolNode 自动解析 tool_calls 并执行对应的工具
tool_node = ToolNode([get_current_time, search_web, send_email])

# 传入包含 tool_calls 的消息
response = model_with_tools.invoke("现在几点了？")
tool_node.invoke({"messages": [response]})
# 返回包含 ToolMessage 的更新字典

flowchart TD
    subgraph 短期记忆
        MSG1[消息 1] --> MSG2[消息 2]
        MSG2 --> MSG3[消息 3]
        MSG3 --> MSG4[...最近 N 条]
    end
    subgraph 长期记忆
        Facts[用户画像<br/>偏好设置<br/>关键事实]
    end

    MSG4 --> Agent[Agent]
    Facts --> Agent
    Agent --> Response[回复]

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": response}

# 构建图 + Checkpointer
graph = StateGraph(MessagesState)
graph.add_node("model", call_model)
graph.add_edge(START, "model")
graph.add_edge("model", END)

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# 多轮对话：同一个 thread_id 延续上下文
config = {"configurable": {"thread_id": "session-1"}}

app.invoke({"messages": [("user", "你好，我叫小明")]}, config=config)
app.invoke({"messages": [("user", "我叫什么名字？")]}, config=config)
# Agent 记得你叫小明

from langgraph.store.memory import InMemoryStore

# 创建 Store
store = InMemoryStore()

# 存储用户偏好
store.put(
    ("users", "user_123"),  # namespace
    "preferences",          # key
    {
        "language": "中文",
        "tone": "正式",
        "topics": ["AI", "Python", "Agent"],
    },
)

# 读取长期记忆
item = store.get(("users", "user_123"), "preferences")
print(item.value)  # {'language': '中文', ...}

# 将记忆注入 System Prompt
user_prefs = store.get(("users", "user_123"), "preferences")
system_prompt = f"""用户偏好:
- 语言: {user_prefs.value['language']}
- 语气: {user_prefs.value['tone']}
- 感兴趣的话题: {', '.join(user_prefs.value['topics'])}
请根据上述偏好回答。"""

messages = [
    ("system", system_prompt),
    ("user", "给我推荐一些学习资料"),
]

response = model.invoke(messages)

from langgraph.store.postgres import PostgresStore

# 生产环境用 PostgreSQL 持久化存储
store = PostgresStore.from_conn_string(
    "postgresql://user:pass@localhost:5432/langgraph"
)
store.setup()  # 建表

# 使用方式和 InMemoryStore 一样
store.put(("users", "user_456"), "preferences", {"language": "English"})

def extract_and_store_memory(state: MessagesState, store, user_id: str):
    """从对话中提取关键信息并存入长期记忆"""
    extraction_prompt = f"""从以下对话中提取用户的个人信息和偏好：
    {state["messages"]}

    输出 JSON 格式的关键信息列表。如果没有新信息，输出 []。"""

    extraction = model.invoke(extraction_prompt)

    import json
    new_memories = json.loads(extraction.content)

    for memory in new_memories:
        store.put(
            ("users", user_id),
            memory["key"],
            memory["value"],
        )

Agent = Model + Tools + System Prompt + Checkpointer + Middleware

flowchart TD
    Input[用户输入] --> Agent
    subgraph Agent[create_agent 内部]
        Graph[StateGraph]
        SystemP[System Prompt 注入]
        LLM[LLM 节点]
        Tools[ToolNode]
        Router[条件路由]
    end
    Graph --> SystemP
    SystemP --> LLM
    LLM --> Router
    Router -->|需要工具| Tools
    Tools --> LLM
    Router -->|完成| Output[最终输出]

from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver

# 1. 定义工具
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气"""
    return f"{city} 今天 22°C 晴"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    return str(eval(expression))

tools = [get_weather, calculate]

# 2. 创建模型
model = ChatOpenAI(model="gpt-4o-mini")

# 3. 一行创建 Agent
agent = create_agent(
    model=model,
    tools=tools,
    system_prompt="你是一个智能助手。请使用工具获取实时信息，不要编造数据。",
    checkpointer=MemorySaver(),
)

# 4. 调用 Agent
config = {"configurable": {"thread_id": "session-1"}}

response = agent.invoke(
    {"messages": [("user", "北京天气怎么样？")]},
    config=config,
)
print(response["messages"][-1].content)

# 多轮对话
response = agent.invoke(
    {"messages": [("user", "那上海呢？")]},
    config=config,
)

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

system_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 {role} 助手。请用 {language} 回答。"),
    MessagesPlaceholder(variable_name="messages"),
])

agent = create_agent(
    model=model,
    tools=tools,
    system_prompt=system_prompt,
    checkpointer=MemorySaver(),
)

response = agent.invoke(
    {
        "messages": [("user", "解释什么是 Agent")],
        "role": "技术",
        "language": "中文",
    },
    config=config,
)

agent = create_agent(
    model=model,
    tools=tools,
    system_prompt="你是一个严谨的数据分析师",
    checkpointer=MemorySaver(),
    recursion_limit=50,
    return_intermediate_steps=True,
)

response = agent.invoke(
    {"messages": [("user", "计算 1234 x 5678")]},
    config=config,
)
if "intermediate_steps" in response:
    for step in response["intermediate_steps"]:
        print(f"步骤: {step}")

from langgraph.prebuilt import create_react_agent

# 更灵活的底层版本
react_agent = create_react_agent(
    model=model,
    tools=tools,
    state_schema=MessagesState,
    state_modifier="你是一个助手",
    checkpointer=MemorySaver(),
)

response = react_agent.invoke(
    {"messages": [("user", "北京天气")]},
    config=config,
)

# 查看 Agent 的图结构
print(agent.get_graph())
agent.get_graph().print_ascii()

def create_agent_simplified(model, tools, system_prompt, checkpointer):
    """简化版的 create_agent 内部实现"""

    prompt_template = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        MessagesPlaceholder(variable_name="messages"),
    ])

    model_with_tools = model.bind_tools(tools)

    def call_agent(state: MessagesState):
        messages = prompt_template.invoke(state).to_messages()
        response = model_with_tools.invoke(messages)
        return {"messages": response}

    def should_continue(state: MessagesState):
        last = state["messages"][-1]
        if hasattr(last, "tool_calls") and last.tool_calls:
            return "tools"
        return END

    graph = StateGraph(MessagesState)
    graph.add_node("agent", call_agent)
    graph.add_node("tools", ToolNode(tools))
    graph.add_edge(START, "agent")
    graph.add_conditional_edges("agent", should_continue)
    graph.add_edge("tools", "agent")

    return graph.compile(checkpointer=checkpointer)

from typing import Annotated, Sequence
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph.message import add_messages

# MessagesState 的内部实现大致是这样一个 TypedDict
class MyMessagesState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    # 你还可以添加其他字段
    user_name: str
    language: str

flowchart LR
    START --> node1[LLM 节点]
    node1 --> END

sequenceDiagram
    participant Input as 输入 State
    participant Node as LLM 节点
    participant Output as 输出 State

    Input->>Node: messages=[HumanMessage("你好")]
    Note over Node: model.invoke(messages) 返回 AIMessage("你好！有什么可以帮助你的？")
    Node->>Output: return {"messages": AIMessage("你好！")}
    Note over Output: add_messages 将新消息追加：<br/>[HumanMessage("你好"), AIMessage("你好！")]

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END

# 1. 初始化模型
model = ChatOpenAI(model="gpt-4o-mini")

# 2. 定义节点函数
#    节点函数接收 state，返回一个字典，字典中的字段会合并到 state 中
def call_model(state: MessagesState):
    """简单的 LLM 节点：把当前消息列表传给模型，返回模型响应"""
    response = model.invoke(state["messages"])
    # 因为 MessagesState 的 messages 字段用了 add_messages reducer，
    # 所以返回 {"messages": response} 会将 response 追加到消息列表末尾
    return {"messages": response}

# 3. 构建图
graph = StateGraph(MessagesState)  # 使用 MessagesState，不需要自定义 State

# 添加节点：第一个参数是节点名称，第二个参数是节点函数
graph.add_node("llm", call_model)

# 添加边：定义数据流向
graph.add_edge(START, "llm")   # 从 START 到 llm 节点
graph.add_edge("llm", END)     # 从 llm 节点到 END（结束）

# 4. 编译图
app = graph.compile()

# 5. 执行图
result = app.invoke({"messages": [("user", "你好！")]})
print(result["messages"])

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 自定义 State：在 MessagesState 基础上增加字段
class CustomState(TypedDict):
    messages: Annotated[list, add_messages]
    language: str          # 用户语言偏好
    processing_time: float # 处理耗时

def call_model_with_lang(state: CustomState):
    """根据用户语言偏好，注入提示词"""
    lang = state.get("language", "中文")
    system_prompt = f"请用{lang}回答用户问题。"
    messages = [("system", system_prompt)] + state["messages"]
    response = model.invoke(messages)
    return {"messages": response}

graph = StateGraph(CustomState)
graph.add_node("llm", call_model_with_lang)
graph.add_edge(START, "llm")
graph.add_edge("llm", END)

app = graph.compile()
result = app.invoke({
    "messages": [("user", "介绍一下自己")],
    "language": "中文",
    "processing_time": 0.0,
})

# add_messages 的简化行为
old_messages = [HumanMessage(content="你好", id="1")]
new_message = AIMessage(content="你好！", id="2")

# add_messages 不会返回 [old + new] 这种两层列表
# 它返回扁平的合并列表
merged = add_messages(old_messages, new_message)
# 结果: [HumanMessage("你好", id="1"), AIMessage("你好！", id="2")]

def my_node(state: MessagesState):
    model.invoke(state["messages"])  # 忘了 return！
    # 错误：这个节点执行了但没有返回结果，状态不变

def my_node_fixed(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": response}  # 正确：返回更新

def call_model(state): ...

graph.add_node("llm_node", call_model)  # 节点名称是 "llm_node"
graph.add_edge("llm_node", END)  # ✅ 正确：用字符串名称
graph.add_edge(call_model, END)  # ❌ 错误：不能用函数对象

def router(state: MessagesState) -> str:
    """根据状态决定下一步去哪"""
    if condition:
        return "node_a"
    else:
        return "node_b"

LLM → 判断是否需要调用工具？
  ├─ 是 → 调用工具 → 回到 LLM（循环）
  └─ 否 → 输出最终回答 → END

flowchart TD
    START --> agent[Agent / LLM]
    agent --> decision{需要调用工具？}
    decision -->|是| tools[执行工具]
    tools --> agent
    decision -->|否| END

flowchart TD
    START --> agent[Agent / LLM]
    agent --> decision{调用哪个工具？}
    decision -->|search_tool| search[搜索工具]
    decision -->|calculator_tool| calc[计算工具]
    decision -->|no tool| END
    search --> agent
    calc --> agent

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode

# 1. 定义工具
def search_weather(city: str) -> str:
    """查询指定城市的天气"""
    # 实际项目中这里会调用真实的天气 API
    return f"{city}今天晴，温度 22-28°C"

def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except:
        return "表达式错误"

tools = [search_weather, calculate]

# 2. 初始化模型并绑定工具
model = ChatOpenAI(model="gpt-4o-mini")
model_with_tools = model.bind_tools(tools)

# 3. 定义节点

def call_agent(state: MessagesState):
    """Agent 核心节点：调用 LLM 进行推理和决策"""
    messages = state["messages"]
    response = model_with_tools.invoke(messages)
    # 如果 LLM 决定调用工具，响应中会包含 tool_calls 字段
    return {"messages": response}

def should_continue(state: MessagesState) -> Literal["tools", END]:
    """条件边路由函数：判断是继续调用工具还是结束"""
    last_message = state["messages"][-1]

    # 如果 LLM 发出了工具调用请求，就去执行工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"

    # 否则结束
    return END

# 4. 构建图
graph = StateGraph(MessagesState)

# 注册节点
graph.add_node("agent", call_agent)
graph.add_node("tools", ToolNode(tools))  # ToolNode 是预置的工具执行节点

# 添加边
graph.add_edge(START, "agent")                       # 从 START 进入 agent
graph.add_conditional_edges("agent", should_continue) # agent 之后条件路由
graph.add_edge("tools", "agent")                     # 工具执行完后回到 agent

# 5. 编译
app = graph.compile()

# 6. 执行
def run_agent(query: str):
    """运行 Agent"""
    result = app.invoke({"messages": [("user", query)]})
    return result["messages"][-1].content

# 测试
print(run_agent("北京天气怎么样？"))
print(run_agent("计算 12345 * 6789"))

# ToolNode 大致相当于下面的代码
def manual_tool_node(state: MessagesState, tools: dict):
    """手动实现工具调用节点"""
    last_message = state["messages"][-1]
    tool_messages = []
    for tc in last_message.tool_calls:
        tool = tools[tc["name"]]
        result = tool.invoke(tc["args"])
        tool_messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))
    return {"messages": tool_messages}

# 基本用法：路由函数返回节点名称
graph.add_conditional_edges("agent", should_continue)

# 如果用字典映射，可以把函数输出映射到不同节点名
graph.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", END: END}  # 函数返回 "tools" → 映射到 "tools" 节点
)

# 路由函数可以返回列表，实现并行分发
def parallel_router(state) -> list[str]:
    tasks = state.get("pending_tasks", [])
    return ["task_worker"] * len(tasks)  # 返回多个相同节点名

# 这是 LangGraph 内部条件边路由的简化逻辑
def handle_conditional_edge(state, router_func, edge_map):
    next_nodes = router_func(state)
    if isinstance(next_nodes, str):
        next_nodes = [next_nodes]
    for node_name in next_nodes:
        if node_name == END or node_name == "__end__":
            return  # 结束
        if node_name not in registered_nodes:
            raise ValueError(f"节点 '{node_name}' 未注册")
        execute_node(node_name, state)

# 设置自定义递归限制
app = graph.compile()
# 在调用时设置
result = app.invoke(
    {"messages": [("user", "复杂任务")]},
    {"recursion_limit": 50}  # 允许最多 50 步
)

def complex_router(state) -> Literal["node_a", "node_b", "node_c", END]:
    """处理多种分支情况"""
    if state.get("requires_human", False):
        return "human_approval"
    if state.get("needs_tools", False):
        return "node_b"
    if state.get("is_done", False):
        return END
    return "node_c"

# 正确
graph.add_conditional_edges("agent", should_continue)

# 也正确（显式映射）
graph.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", END: END}
)

def router(state) -> str:
    if condition:
        return "not_registered_node"  # ❌ 这个节点没注册！
    return END

graph.add_node("agent", call_agent)  # 只注册了 agent
graph.add_conditional_edges("agent", router)  # 运行时会报错

system_prompt = """你是一个智能助手。你可以使用工具来获取信息。
但请注意：
1. 只有需要实时数据时才调用工具。
2. 如果用户只是闲聊，直接回答即可。
3. 如果工具已经返回了足够的信息，基于这些信息回答即可，不需要反复调工具。"""

# 不同流模式的效果对比
app.stream(input, stream_mode="values")      # 每步输出完整 State（最粗粒度）
app.stream(input, stream_mode="updates")     # 每步输出节点更新值（调试友好）
app.stream(input, stream_mode="messages")    # Token 级流式（最细粒度）
app.stream(input, stream_mode="custom")      # 仅自定义事件
app.stream(input, stream_mode=["updates", "messages"])  # 组合模式

sequenceDiagram
    participant Client as 客户端
    participant Server as LangGraph 服务
    participant LLM as LLM API

    Client->>Server: invoke({"messages": ["讲个故事"]})
    Server->>LLM: model.invoke(messages, stream=True)

    Note over LLM: 开始逐 Token 生成

    loop 每个 Token
        LLM->>Server: Token "从前"
        Server->>Client: event: token, content="从前"

        LLM->>Server: Token "有"
        Server->>Client: event: token, content="有"

        LLM->>Server: Token "座"
        Server->>Client: event: token, content="座"

        LLM->>Server: Token "山"
        Server->>Client: event: token, content="山"
    end

    Note over LLM: 生成完毕
    Server->>Client: event: end, final message

sequenceDiagram
    participant Client as 客户端
    participant Graph as LangGraph 图

    Client->>Graph: stream(input)
    Note over Graph: 执行 agent 节点
    Graph->>Client: {"agent": {"messages": [AIMessage(...)]}}

    Note over Graph: 条件路由 → 决定走 tools
    Note over Graph: 执行 tools 节点
    Graph->>Client: {"tools": {"messages": [ToolMessage(...)]}}

    Note over Graph: 回到 agent 节点
    Graph->>Client: {"agent": {"messages": [AIMessage(...)]}}

    Note over Graph: 条件路由 → 决定结束
    Graph->>Client: end of stream

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode

model = ChatOpenAI(model="gpt-4o-mini")
tools = [...]  # 你的工具列表

def call_agent(state: MessagesState):
    response = model.bind_tools(tools).invoke(state["messages"])
    return {"messages": response}

def should_continue(state):
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

graph = StateGraph(MessagesState)
graph.add_node("agent", call_agent)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

app = graph.compile()

# 节点级流式：每执行完一个节点，拿到该节点的输出
for event in app.stream(
    {"messages": [("user", "北京天气怎么样？")]},
    stream_mode="updates"  # 每次返回一个节点的更新
):
    for node_name, update in event.items():
        print(f"[{node_name}] 输出: {update}")
        # 输出可能是：
        # [agent] 输出: {'messages': AIMessage(...tool_calls...)}
        # [tools] 输出: {'messages': ToolMessage(...)}
        # [agent] 输出: {'messages': AIMessage("北京今天晴...")}

# Token 级流式：逐 Token 输出 AI 消息内容
for event in app.stream(
    {"messages": [("user", "讲个笑话")]},
    stream_mode="messages"  # 开启 Token 级流式
):
    # 在 messages 模式下，每个 event 是一个 (chunk, metadata) 元组
    chunk, metadata = event
    if hasattr(chunk, "content"):
        print(chunk.content, end="", flush=True)
    # chunk 可能是 AIMessageChunk，包含部分内容

# 同时订阅多个流模式
for event_type, data in app.stream(
    {"messages": [("user", "讲个故事")]},
    stream_mode=["updates", "messages"]
):
    if event_type == "updates":
        # 节点更新事件
        for node_name, update in data.items():
            print(f"\n[节点完成] {node_name}")

    elif event_type == "messages":
        # Token 事件
        chunk, metadata = data
        if hasattr(chunk, "content"):
            print(chunk.content, end="", flush=True)

from langgraph.callbacks import dispatch_custom_event

def long_running_node(state: MessagesState):
    """一个耗时较长的节点，发射进度事件给前端"""
    total_steps = 5

    for i in range(total_steps):
        # 执行子任务
        result = do_some_work(state, i)

        # 发射自定义事件（需要配置 stream_mode="custom"）
        dispatch_custom_event(
            "progress",  # 事件名称
            {
                "step": i + 1,
                "total": total_steps,
                "partial_result": result,
            }
        )

    return {"final_result": aggregate_results(state)}

# 前端接收自定义事件
for event in app.stream(
    input_data,
    stream_mode="custom"
):
    if event["type"] == "progress":
        progress = event["data"]
        print(f"进度: {progress['step']}/{progress['total']}")

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

# 流式 API 端点
@app.post("/chat/stream")
async def chat_stream(query: dict):
    """流式聊天 API"""
    async def event_generator():
        async for event in app.astream(
            {"messages": [("user", query["message"])]},
            stream_mode=["messages", "updates"]
        ):
            yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )

data: {"event": "messages", "data": {"chunk": "从前", "metadata": {...}}}

data: {"event": "messages", "data": {"chunk": "有座", "metadata": {...}}}

data: {"event": "updates", "data": {"agent": {"messages": [...]}}}

data: [DONE]

try:
    for event in app.stream(input, stream_mode="messages"):
        chunk, metadata = event
        print(chunk.content, end="", flush=True)
except ConnectionError:
    print("\n[连接中断]")
except Exception as e:
    print(f"\n[流错误]: {e}")

# interrupt 的基本使用模式
def approval_node(state: MessagesState):
    """需要人类审批的节点"""
    decision = interrupt({
        "question": "是否批准这个操作？",
        "details": state["pending_action"],
    })
    # 当人类通过 Command(resume="approve") 恢复时，
    # decision 的值就是 "approve"
    return {"decision": decision}

sequenceDiagram
    participant User as 用户（人类）
    participant Agent as LangGraph Agent
    participant LLM as LLM

    User->>Agent: 发送消息："给我发一封邮件"
    Agent->>LLM: 调用 LLM 生成邮件内容
    LLM->>Agent: 返回邮件草稿
    Note over Agent: Agent 决定需要审批
    Agent->>Agent: interrupt({question: "是否发送？", draft: "..."})
    Agent-->>User: ❗ 暂停，等待你的决定

    Note over User: 人类查看邮件内容
    User->>Agent: Command(resume="approve")
    Note over Agent: interrupt 返回 "approve"
    Agent->>LLM: 调用发送邮件工具
    LLM->>Agent: 发送成功
    Agent->>User: "邮件已发送"

flowchart TD
    START --> agent[Agent]
    agent --> approve1{初级审批}
    approve1 -->|拒绝| END
    approve1 -->|批准| agent2[执行操作]
    agent2 --> approve2{高级审批<br/>金额>10000}
    approve2 -->|拒绝| END
    approve2 -->|批准| final[最终执行]
    final --> END

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

def call_agent(state: MessagesState):
    """Agent 节点：生成回复"""
    response = model.invoke(state["messages"])
    return {"messages": response}

def human_approval_node(state: MessagesState):
    """审批节点：暂停执行，等待人类确认"""
    last_msg = state["messages"][-1]

    # Interrupt 暂停图执行，返回数据给调用方
    # 调用方通过 Command(resume=...) 恢复执行
    decision = interrupt({
        "type": "approval",
        "question": "是否发送这条消息？",
        "content": last_msg.content,
    })

    # 当 Command(resume=...) 被调用后，decision 就是 resume 的值
    return {"approved": decision == "approve"}

def after_approval(state: MessagesState):
    """审批后节点：根据审批结果执行"""
    if state.get("approved", False):
        return {"messages": [("system", "[消息已发送]")]}
    else:
        return {"messages": [("system", "[消息已取消]")]}

# 构建图
graph = StateGraph(MessagesState)
graph.add_node("agent", call_agent)
graph.add_node("approval", human_approval_node)
graph.add_node("result", after_approval)

graph.add_edge(START, "agent")
graph.add_edge("agent", "approval")
graph.add_edge("approval", "result")
graph.add_edge("result", END)

# ⚠️ 重要：interrupt 依赖 checkpoint，必须配置 checkpointer
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# 第一次执行：在 approval 节点处暂停
thread_config = {"configurable": {"thread_id": "thread-001"}}

# 这会执行到 approval 节点，然后暂停
result = app.invoke(
    {"messages": [("user", "帮我发一条消息给团队：明天下午开会")]},
    config=thread_config
)
# 此时 result 包含 interrupt 之前的所有状态
# 调用方可以展示给人类看

# 人类做出决定后，通过 Command(resume=...) 恢复执行
result = app.invoke(
    None,  # 不需要再传入 input
    config=thread_config,
    command=Command(resume="approve")  # 注入人类的决定
)
# interrupt 返回 "approve"，继续执行 result 节点
print(result["messages"][-1].content)

def first_approval(state: MessagesState):
    """初级审批"""
    decision = interrupt({"level": "1", "question": "初级审批：是否继续？"})
    return {"level1_approved": decision == "approve"}

def second_approval(state: MessagesState):
    """高级审批（高金额场景）"""
    amount = state.get("amount", 0)
    if amount > 10000:
        decision = interrupt({"level": "2", "question": f"金额 {amount} 超过 10000，需要高级审批"})
        return {"level2_approved": decision == "approve"}
    return {"level2_approved": True}

graph = StateGraph(MessagesState)
graph.add_node("agent", call_agent)
graph.add_node("approval1", first_approval)
graph.add_node("approval2", second_approval)
graph.add_edge(START, "agent")
graph.add_edge("agent", "approval1")
graph.add_edge("approval1", "approval2")
graph.add_edge("approval2", END)

app = graph.compile(checkpointer=MemorySaver())

# 逐步恢复每个中断点
thread_config = {"configurable": {"thread_id": "thread-002"}}

# 第一次执行：停在 approval1
app.invoke({"messages": [("user", "支付 50000 元")]}, config=thread_config)

# 审批 level1
app.invoke(None, config=thread_config, command=Command(resume="approve"))

# 第二次恢复后停在 approval2（因为金额 > 10000）
app.invoke(None, config=thread_config, command=Command(resume="approve"))

# 执行完毕

import time

def approval_with_timeout(state: MessagesState):
    """带超时的审批节点"""
    # 记录开始时间
    start_time = time.time()
    max_wait = 30  # 最多等 30 秒

    while time.time() - start_time < max_wait:
        try:
            decision = interrupt({
                "type": "approval",
                "question": "是否批准？30 秒内未响应将自动拒绝。",
                "timeout": max_wait,
                "remaining": int(max_wait - (time.time() - start_time)),
            })
            return {"approved": decision == "approve"}
        except TimeoutError:
            # 如果中断超时，自动拒绝
            pass

    return {"approved": False}