Agent学习笔记转载

如：用户说：帮我查一下这个客户最近三个月的订单异常情况

后端负责：
1. 校验用户是否有权限查看该客户数据
2. 提供订单查询接口
3. 提供异常订单规则
4. 记录操作日志

Agent 负责：
1. 理解用户想查什么
2. 决定需要调用哪些接口
3. 分析接口返回的数据
4. 总结异常原因并生成自然语言结果

python -m venv .venv

.venv/Scripts/activate
激活成功后，终端前面一般会出现：(.venv)

pip install openai python-dotenv requests notebook jupyterlab ipykernel

请帮我在当前项目目录中搭建 Python + Jupyter Notebook 的学习环境，要求如下：

1. 检查当前电脑是否安装 Python，并确认版本是否为 3.10 或以上。
2. 在当前项目根目录创建 Python 虚拟环境，目录名为 .venv。
3. 激活 .venv 虚拟环境。
4. 在虚拟环境中安装以下依赖：
   - openai
   - python-dotenv
   - requests
   - notebook
   - jupyterlab
   - ipykernel
5. 将当前 .venv 注册为 Jupyter Notebook 内核，内核显示名称为 ".venv"。
6. 在项目根目录创建 .env.example 文件，内容如下：

OPENAI_API_KEY=你的_api_key
OPENAI_BASE_URL=BASE_URL
MODEL_NAME=模型名称

7. 不要创建真实 .env 文件，也不要替我填写 API Key。
8. 安装完成后，请告诉我：
   - Python 版本
   - pip 版本
   - Jupyter 是否安装成功
   - Notebook 内核是否注册成功
   - 下一步我应该如何打开 .ipynb 文件

# 导入 os 模块，用于读取环境变量
import os

# 导入 dotenv，用于从 .env 文件中加载配置
from dotenv import load_dotenv

# 导入 OpenAI 客户端
from openai import OpenAI


# 读取项目根目录下的 .env 文件
load_dotenv(override=True)


# 创建大模型客户端
# api_key：你的 API Key，用于身份认证
# base_url：模型服务地址
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL")
)


# 调用大模型的 Chat Completions 接口
response = client.chat.completions.create(
    # 指定要调用的模型名称
    # .env 配置的模型名称
    model=os.getenv("MODEL_NAME"),

    # messages 是传给大模型的对话上下文
    messages=[
        # system 用来设定模型的角色和行为
        {"role": "system", "content": "你是一个有用的 AI 助手。"},

        # user 表示用户输入的问题
        {"role": "user", "content": "请用一句话解释什么是 Agent。"}
    ]
)


# 取出模型回复的正文内容
answer = response.choices[0].message.content

# 打印模型回复
print(answer)

# 定义一个天气获取工具
# 这里先不调用真实天气 API，只返回固定的北京天气
def get_weather():
    """返回北京天气信息。"""
    return "北京天气：晴，气温 25℃，适合出门。"


# 手动调用工具，查看返回结果
weather = get_weather()
print(weather)

# 1. 定义工具说明书
# 这份说明书是给大模型看的，不是给 Python 执行的
tools = [
    {
        "type": "function",
        "function": {
            # 工具名称，需要和后面的工具注册表保持一致
            "name": "get_weather",

            # 工具描述：告诉大模型这个工具什么时候该用
            "description": "获取北京当前天气。当用户询问北京天气时使用这个工具。",

            # 工具参数：这个工具暂时不需要参数，所以 properties 为空
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    }
]

# 2. 建立工具注册表
# 注册表的作用：把大模型返回的工具名称，映射到真正的 Python 函数
tool_registry = {
    "get_weather": get_weather
}

# 3. 准备用户问题
messages = [
    {"role": "system", "content": "你是一个有用的 AI 助手。"},
    {"role": "user", "content": "北京今天天气怎么样？"}
]

# 4. 第一次调用大模型
# 这一次调用会把 tools 一起传给大模型，让它自己判断是否需要调用工具
response = client.chat.completions.create(
    model=os.getenv("MODEL_NAME"),
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

# 5. 取出大模型的回复
assistant_message = response.choices[0].message

# 6. 判断大模型是否想调用工具
if assistant_message.tool_calls:
    print("大模型决定调用工具")

    # 先把大模型的工具调用请求加入消息历史
    messages.append(assistant_message)

    # 遍历大模型请求调用的工具
    for tool_call in assistant_message.tool_calls:
        tool_name = tool_call.function.name
        print("工具名称：", tool_name)

        # 从注册表中找到真正的 Python 函数
        tool_func = tool_registry[tool_name]

        # 执行工具，拿到结果
        tool_result = tool_func()
        print("工具返回：", tool_result)

        # 把工具结果加入消息历史，交还给大模型
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": tool_result
        })

    # 7. 第二次调用大模型
    # 这一次大模型会根据工具结果，组织成自然语言回答
    final_response = client.chat.completions.create(
        model=os.getenv("MODEL_NAME"),
        messages=messages
    )

    print("最终回答：")
    print(final_response.choices[0].message.content)

else:
    # 如果大模型认为不需要工具，就直接输出它的回答
    print("大模型没有调用工具")
    print(assistant_message.content)

1. 写一个 Python 工具函数 get_weather()
2. 写 tools 工具说明书，让大模型知道有这个工具
3. 大模型返回 tool_calls，表示“我想调用 get_weather”
4. Python 执行 get_weather，再把结果交还给大模型生成最终回答

用户问题
  ↓
LLM 读取 tools 说明书
  ↓
LLM 生成 tool_calls
  ↓
Python 执行真实函数
  ↓
工具结果写回 messages
  ↓
LLM 生成最终回答

我要调用 get_weather

LLM 如何决定调用工具
Python 如何执行工具
工具结果如何回到 LLM
LLM 如何生成最终回答

pip install -U langchain langchain-core python-dotenv langchain-openai

本教程使用的 langchain 版本为 1.2x；各版本可能会有差异

from dotenv import load_dotenv
import os

from langchain_openai import ChatOpenAI

load_dotenv(override=True)

llm = ChatOpenAI(
    model="gpt-5.5",
    temperature=0,
    # 在根目录建一个.env，把base_url和apikay放到环境变量中
    base_url=os.getenv("OPENAI_BASE_URL"),
    api_key=os.getenv("OPENAI_API_KEY")
  )

# invoke()：执行一次调用
response = llm.invoke("请用一句话介绍 LangChain")
print(response)

from langchain.messages import SystemMessage, HumanMessage

messages = [
    SystemMessage("你是一名耐心的 Python 老师，回答要适合新手。"),
    HumanMessage("什么是函数？")
]

response = llm.invoke(messages)

ai_message = response.content

print(response.content)

from langchain.messages import AIMessage

messages.append(AIMessage(ai_message))
messages.append(HumanMessage("能不能再举一个例子？"))

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

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template(
    "请用新手能听懂的方式解释：{topic}"
)

formatted_prompt = prompt.invoke({"topic": "LangChain"})
print(formatted_prompt.text)

response = llm.invoke(formatted_prompt.text)
print(response.content)

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名 AI 应用开发老师，回答要清晰、简洁。"),
    ("human", "请解释这个概念：{topic}")
])

messages = prompt.invoke({"topic": "Prompt 模板"})
response = llm.invoke(messages)

print(response.content)

from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名新手教程作者。"),
    ("human", "请用三句话解释：{topic}")
])

parser = StrOutputParser()

chain = prompt | llm | parser

result = chain.invoke({"topic": "LangChain 的 Runnable"})
print(result)

输入变量 -> Prompt 生成消息 -> 模型生成回复 -> 解析成字符串

我们可以理解为语法糖

from pydantic import BaseModel, Field

class Person(BaseModel):
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    city: str = Field(description="所在城市")

structured_model = llm.with_structured_output(
    Person,
    method="function_calling"
)

result = structured_model.invoke(
    "张三今年 18 岁，住在北京。"
)

print(result)
print(result.name)
print(result.age)
print(result.city)

for chunk in llm.stream("请用 100 字介绍 LangChain"):
    print(chunk.content, end="", flush=True)

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询城市天气的本地示例工具。"""
    weather_map = {
        "北京": "晴，25 摄氏度",
        "上海": "多云，27 摄氏度",
        "广州": "小雨，29 摄氏度",
    }
    return weather_map.get(city, "暂未查询到该城市天气")

from langchain.agents import create_agent

agent = create_agent(
    model=llm,
    tools=[get_weather],
    system_prompt="你是一个会使用工具解决问题的助手。"
)

result = agent.invoke({
    "messages": [
        {"role": "user", "content": "上海的天气如何？"}
    ]
})

print(result["messages"][-1].content)

用户提出问题 -> 模型判断是否需要工具 -> 调用工具 -> 读取工具结果 -> 生成最终回答

[HumanMessage(content='上海的天气如何？'), AIMessage(content='', , tool_calls=[{'name': 'get_weather', 'args': {'city': '上海'}}), ToolMessage(content='多云，27 摄氏度', name='get_weather'), AIMessage(content='上海现在多云，气温约 27°C。')]

1: HumanMessage: 用户输入
2: AIMessage: ai回答
3: tool_calls: ai抛出要调用工具请求
4: ToolMessage: 工具调用结果
5: AIMessage: 调用完工具后，AI的输出

# 简单任务可以直接传字符串：
model.invoke("解释一下 Python 变量")

# 需要角色设定、多轮历史时，使用 messages：
model.invoke([
    SystemMessage("你是一名 Python 老师"),
    HumanMessage("什么是变量？")
])

如果只是问答、总结、改写，不一定需要 Agent。

如果任务需要模型自己判断是否调用工具，例如计算、搜索、查数据库、调用 API，就可以考虑 Agent。

LangChain 负责提供大模型应用开发的常用组件
LangGraph 负责管理更复杂的 Agent / 工作流执行过程。

在 LangChain 1.0 之后，很多 Agent 能力底层其实是基于 LangGraph 构建的。

例如:
你用 create_agent() 创建 Agent 时
表面上是在使用 LangChain 的高级接口；
但底层执行流程、状态流转、工具调用等能力，很多是由 LangGraph 提供支撑的。

在 Agent 和复杂工作流这部分，可以理解为 LangChain 是对 LangGraph 的高层封装。

分场景：
1：如果你只是做一个很简单的大模型调用：没必要

2：如果你要做的是一个真正的 AI 应用，LangChain 就很有价值了
对新手来说，学习 LangChain 的好处是：

能理解大模型应用的标准开发流程
能快速接触 Prompt、Message、Tool、Agent、RAG 等核心概念
能少写很多重复代码
能更容易从简单 Demo 过渡到真实项目

总结：
如果你只是玩 API，可以不学 LangChain；
如果你想系统开发 AI 应用，LangChain 很值得学。

入门阶段，先学本教程这些知识点就够了。

from dotenv import load_dotenv
import os

load_dotenv(override=True)

api_key = os.getenv("OPENAI_API_KEY")
base_url = os.getenv("OPENAI_BASE_URL")
model_name = os.getenv("MODEL_NAME", "gpt-5.4-mini")

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model=model_name,
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

print("模型初始化完成")

from langchain.agents import create_agent

agent = create_agent(
    model=llm,
    tools=[],
    system_prompt=(
        "你叫小智，是一名乐于助人的智能助手。"
        "请在对话中保持温和、有耐心的语气。"
        "回答新手问题时，要尽量解释清楚，并适当举例。"
    ),
)

print("Agent 创建完成")

result = agent.invoke({
    "messages": [
        {"role": "user", "content": "你好，你是谁？"}
    ]
})

final_message = result["messages"][-1]
print(final_message.content)

messages = []

messages.append({
  "role": "user", "content": "我叫小明"
})

result = agent.invoke({"messages": messages})
reply = result["messages"][-1].content

print("小智:", reply)

messages.append({"role": "assistant", "content": reply})

stream_messages = [
    {"role": "user", "content": "请用三句话解释什么是 LangChain。"}
]

full_reply = ""

print("小智：", end="", flush=True)

for part in agent.stream(
    {"messages": stream_messages},
    stream_mode="messages",
    version="v2",
):
    if part["type"] != "messages":
        continue

    token, metadata = part["data"]

    if metadata.get("langgraph_node") != "model":
        continue

    if token.text:
        print(token.text, end="", flush=True)
        full_reply += token.text

print("\n\n完整回复：")
print(full_reply)

def chat_once(user_input: str, messages: list[dict]) -> str:
    messages.append({"role": "user", "content": user_input})

    print("小智：", end="", flush=True)
    full_reply = ""

    for part in agent.stream(
        {"messages": messages},
        stream_mode="messages",
        version="v2",
    ):
        if part["type"] != "messages":
            continue

        token, metadata = part["data"]

        if metadata.get("langgraph_node") != "model":
            continue

        if token.text:
            print(token.text, end="", flush=True)
            full_reply += token.text

    print()

    messages.append({"role": "assistant", "content": full_reply})
    return full_reply

messages = []

print("输入 exit 或 quit 退出对话\n")

while True:
    user_input = input("你：")

    if user_input.lower() in {"exit", "quit"}:
        print("对话结束，再见！")
        break

    if not user_input.strip():
        print("请输入内容。")
        continue

    print(f"你：{user_input}")
    chat_once(user_input, messages)

    print("-" * 40)

    # 避免历史消息无限增长。这里保留最近 100 条消息。
    messages = messages[-100:]

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent/graph.py:graph"
  },
  "env": ".env"
}

from dotenv import load_dotenv
import os
from langchain_openai import ChatOpenAI

from langchain.agents import create_agent

load_dotenv(override=True)


def get_weather(city: str) -> str:
    """查询指定城市的天气。"""
    return f"{city} 现在晴,25°C。"


llm = ChatOpenAI(
    model="gpt-5.5",
    temperature=0,
    base_url=os.getenv("OPENAI_BASE_URL"),
    api_key=os.getenv("OPENAI_API_KEY")
)


graph = create_agent(
    model=llm,
    tools=[get_weather],
    system_prompt="你是一个乐于助人的助手,需要时调用工具来回答问题。",
)

langgraph dev

问：北京天气怎样？

__start__
   │
   ▼
 model      ← 模型看到问题,决定"得调 get_weather"
   │ (虚线,选了 tools 这条岔路)
   ▼
 tools      ← 执行 get_weather("北京") → "北京 现在晴,25°C"
   │ (虚线,把结果送回去)
   ▼
 model      ← 模型拿到工具结果,组织成人话:"北京现在晴,25度"
   │ (虚线,这次没工具要调了,选 __end__)
   ▼
__end__     ← 返回最终答复

# 导入
from langchain.agents.middleware import AgentMiddleware

# 自定义中间件，所有中间件钩子都写在这里
# 继承AgentMiddleware
# 没实现的话，就不会触发，也不会在 studio 图中画出
class MyMiddleware(AgentMiddleware):

# 把定义好的中间件传到 graph 中
graph = create_agent(
    model=llm,
    tools=[get_weather],
    system_prompt="你是一个乐于助人的助手,需要时调用工具来回答问题。",
    middleware=[MyMiddleware()],
)

class MyMiddleware(AgentMiddleware):
  
  def before_model(self, state, runtime):
    print("agent开始前")
    messages = state["messages"]
    print("消息列表：", messages)
    return None

def after_model(self, state, runtime):
  last = state["messages"][-1]
  print(f"🟢 [after_model] 模型已返回{last.content!r}")
  return None

from langchain.agents.middleware import ModelCallLimitMiddleware

graph = create_agent(
    model=llm,
    tools=[get_weather],
    middleware=[
        MyMiddleware(),                          # 自定义
        ModelCallLimitMiddleware(max_calls=3),   # 封装好的,直接配参数
    ],
)