LangChain 是一个面向大语言模型（LLM）的开源框架，通过链式任务、代理、工具调用和记忆管理，将自然语言理解与外部知识和功能模块结合，实现复杂多步骤任务和知识增强生成（RAG）应用。

LangChain vs LangGraph

• LangChain：统一接口 + 模块化设计，快速将 LLM 组装成可用的智能体
• LangGraph：状态机 + 图结构，用于长期运行、流程复杂、容错要求高的场景

LangChain 安装

pip

pip install -U langchain
# Requires Python 3.10+

uv

uv add langchain
# Requires Python 3.10+

集成 LLM Provider

pip

pip install -U langchain-openai
pip install -U langchain-anthropic

uv

uv add langchain-openai
uv add langchain-anthropic

基础用法

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

llm = ChatOpenAI(
    model="deepseek-chat",
    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
    openai_api_base=os.getenv("DEEPSEEK_API_BASE"),
    temperature=0.9  # 0-2，值越高输出越随机
)

print(llm.invoke("Hello, how are you?"))

标准化参数

在初始化 ChatModels 时，可设置以下标准化参数：

• model: 模型名称
• temperature: 采样温度（0-2，控制输出随机性）
• timeout: 请求超时时间
• max_tokens: 生成的最大令牌数
  • 中文：约 1.5-2 token/字（因中文字符编码复杂度较高）
  • 英文：约 0.25 token/字符 或 1 token/词
• max_retries: 请求失败时的最大重试次数
• api_key: 大模型供应商的 API 密钥
• base_url: API 请求的端点地址

流式输出

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

# 加载 .env 文件中的环境变量
load_dotenv()

# 使用 GLM (智谱AI) 的配置 - 使用 thinking 模型
llm = ChatOpenAI(
    model="glm-z1-flash",  # GLM-Z1-Flash: 支持深度思考的模型
    openai_api_key=os.getenv("GLM_API_KEY"),
    openai_api_base=os.getenv("GLM_API_BASE"),  # https://open.bigmodel.cn/api/paas/v4
    temperature=0.7
)

# 流式输出示例（包含 thinking）
print("\n💭 思考过程:")
print("-" * 30)

for chunk in llm.stream('派有多少位'):
    # 检查是否有 reasoning_content（思考过程）
    reasoning = chunk.additional_kwargs.get('reasoning_content', '')
    if reasoning:
        print(reasoning, end="", flush=True)

    # 输出最终回答
    if chunk.content:
        # 如果之前有thinking，先换行再输出
        if reasoning:
            print("\n" + "-" * 30)
            print("\n📝 最终回答:")
            print("-" * 30)
        print(chunk.content, end="", flush=True)

print("\n" + "=" * 50)
print("流式输出完成！")

速率限制

# 内存速率限制器：用于控制 API 请求频率，防止超出配额限制
rate_limiter = InMemoryRateLimiter(
    requests_per_second=0.1,  # 每秒允许的请求数（0.1 = 每 10 秒 1 个请求）
    check_every_n_seconds=0.1,  # 检查速率限制的时间间隔（秒）
    max_bucket_size=10  # 令牌桶最大容量，允许短时间内积累的最大请求数
)

model = init_chat_model(
    model="glm-4-flash",
    model_provider="openai",
    api_key=os.getenv("GLM_API_KEY"),
    base_url=os.getenv("GLM_API_BASE"),
    rate_limiter=rate_limiter
)
t1 = time.time()
print(model.invoke("Hello, how are you?"))
t2 = time.time()
print(model.invoke("Hello!"))
t3 = time.time()

print(f"{t3 - t2} ~ 10s")

格式化输出

from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from glm_llm import llm


class Movie(BaseModel):
    title: str = Field(..., description="电影标题")
    year: int = Field(..., description="电影发行年份")
    director: str = Field(..., description="电影导演")
    rating: float = Field(..., description="电影评分（满分10分）")


parser = PydanticOutputParser(pydantic_object=Movie)

prompt = ChatPromptTemplate.from_messages([
    ("system", "{format_instructions}"),
    ("human", "{query}")
])

chain = prompt | llm | parser

response = chain.invoke({
    "query": "提供电影《三傻大闹宝莱坞》的信息",
    "format_instructions": parser.get_format_instructions()
})
print(response)

Agent 开发

LangGraph 提供本地开发服务器，配合 LangGraph Studio 可视化调试 Agent。

LangSmith

LangSmith 是 LangChain 官方的可观察性平台，提供 Tracing、Evaluation、Prompt Management 等功能。开发 LangGraph 应用需要 LangSmith API Key（免费注册）。

LangSmith

1. 安装 LangGraph CLI

Python >= 3.11

pip install --upgrade "langgraph-cli[inmem]"
# 或使用 uv
uv add langgraph-cli --group dev

2. 创建项目

使用模板快速创建：

langgraph new my-agent --template new-langgraph-project-python

或手动创建项目结构：

my-agent/
├── src/
│   └── agent/
│       └── my_agent.py
├── langgraph.json
├── pyproject.toml
└── .env

3. 配置 .env

在 LangSmith Settings 获取 API Key：

LANGSMITH_API_KEY=lsv2_pt_xxx
LANGCHAIN_TRACING_V2=true

4. 创建 langgraph.json

{
  "$schema": "https://langchain-ai.github.io/langgraph/schemas/langgraph.schema.json",
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent/my_agent.py:agent"
  },
  "env": ".env"
}

• dependencies: 项目依赖路径，. 表示当前目录
• graphs: Agent 入口，格式为 文件路径:变量名
• env: 环境变量文件路径

5. 编写 Agent

# src/agent/my_agent.py
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="glm-4-flash",
    api_key=os.getenv("GLM_API_KEY"),
    base_url=os.getenv("GLM_API_BASE"),
)


def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件到指定地址

    Args:
        to: 收件人邮箱地址
        subject: 邮件主题
        body: 邮件正文内容
    """
    # 邮件发送逻辑
    return f"邮件已发送至 {to}"


agent = create_react_agent(
    model=llm,
    tools=[send_email],
    prompt="你是一个邮件助手。请使用 send_email 工具发送邮件。",
)

6. 配置 pyproject.toml

[project]
name = "my-agent"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "langgraph>=0.2.0",
    "langchain-openai>=0.3.0",
    "python-dotenv>=1.0.0",
]

[dependency-groups]
dev = [
    "langgraph-cli[inmem]>=0.1.0",
]

安装依赖：

pip install -e .

7. 启动本地服务

langgraph dev

启动成功后输出：

> Ready!
> - API: http://localhost:2024
> - Docs: http://localhost:2024/docs
> - LangGraph Studio: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024

访问 LangGraph Studio 链接即可在浏览器中调试 Agent。

Define Tools

"""
网页搜索工具模块
提供两种实现方式：@tool 装饰器 和 BaseTool 类
"""
from typing import Type

from langchain_core.tools import tool, BaseTool
from pydantic import BaseModel, Field

from src.agent.llm import zhipu_client


# ==================== 方式一：@tool 装饰器 ====================

class WebSearchInput(BaseModel):
    """网页搜索工具的输入参数"""
    query: str = Field(description="要搜索的关键词或问题")


@tool(description="搜索互联网的工具，可以搜索所有公开信息", args_schema=WebSearchInput)
def web_search(query: str) -> str:
    """执行网页搜索并返回结果"""
    try:
        res = zhipu_client.web_search.web_search(
            search_engine="search_pro",
            search_query=query
        )
        if res.search_result:
            return "\n\n".join([chunk.content for chunk in res.search_result])
        else:
            return "未找到相关结果"
    except Exception as e:
        return f"搜索失败: {str(e)}"


# ==================== 方式二：BaseTool 类 ====================

class WebSearchTool(BaseTool):
    """基于 BaseTool 的网页搜索工具"""
    name: str = "web_search"
    description: str = "搜索互联网的工具，可以搜索所有公开信息"
    args_schema: Type[BaseModel] = WebSearchInput

    def _run(self, query: str) -> str:
        """同步执行网页搜索"""
        try:
            res = zhipu_client.web_search.web_search(
                search_engine="search_pro",
                search_query=query
            )
            if res.search_result:
                return "\n\n".join([chunk.content for chunk in res.search_result])
            else:
                return "未找到相关结果"
        except Exception as e:
            return f"搜索失败: {str(e)}"

    async def _arun(self, query: str) -> str:
        """异步执行（可选实现）"""
        # 如果有异步版本的 API，可以在这里实现
        return self._run(query)


# 创建 BaseTool 实例
web_search_tool = WebSearchTool()


if __name__ == '__main__':
    print("=== @tool 装饰器方式 ===")
    print(f"name: {web_search.name}")
    print(f"description: {web_search.description}")
    print(f"args_schema: {web_search.args_schema.model_json_schema()}")

    print("\n=== BaseTool 类方式 ===")
    print(f"name: {web_search_tool.name}")
    print(f"description: {web_search_tool.description}")
    print(f"args_schema: {web_search_tool.args_schema.model_json_schema()}")

    # 测试调用
    print("\n=== 测试调用 ===")
    res = web_search_tool.invoke({'query': '如何使用LangChain？'})
    print(res)

Text ->SQL

1. 用户输入自然语言
2. 问题重写和关键词提取
3. LLM 查询数据库，和已有的表结构和表关系比对
4. SQL 查询生成
5. SQL 校验->语义一致性校验
6. 是否通过，通过则开始 SQL 查询；反之，重新->Step 4
7. 将返回的 sql 结果转换为自然语言
8. 返回 ANS

贡献者: paiad