viv Blog

构建第一个简单的langgraph项目如下：

.env配置
DEFAULT_LLM_MODEL_NAME：xxx

DEFAULT_LLM_API_KEY:xxx

DEFAULT_LLM_BASE_URL:xxx

import operator
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage, AIMessage
from langchain.tools import tool
from dotenv import load_dotenv
import os

# ----------------------------
# 加载环境变量 & 初始化 LLM
# ----------------------------
load_dotenv()
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model=os.getenv("DEFAULT_LLM_MODEL_NAME"),
    api_key=os.getenv("DEFAULT_LLM_API_KEY"),
    base_url=os.getenv("DEFAULT_LLM_BASE_URL"),
    temperature=0,
)

# ----------------------------
# 工具定义
# ----------------------------
@tool
def analyze_sales_data(query: str) -> str:
    """分析销售数据趋势，返回季度增长、品类表现等关键指标摘要"""
    return "【销售分析】Q4销售额同比增长25%，主要来自电子产品类目。"

@tool
def analyze_sentiment(text: str) -> str:
    """分析用户评论的情感倾向，返回满意度、关键词和整体评价"""
    return "【情感分析】用户评论整体积极，满意度达92%。"

tools = {
    "analyze_sales_data": analyze_sales_data,
    "analyze_sentiment": analyze_sentiment,
}

# ----------------------------
# 状态定义
# ----------------------------
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    sales_result: str | None
    sentiment_result: str | None
    final_report: str | None

# ----------------------------
# 节点函数
# ----------------------------
def route_tasks(state: AgentState) -> Literal["analyze_sales", "analyze_sentiment", "summarize"]:
    """Orchestrator：决定启动哪些分析"""
    query = state["messages"][0].content
    has_sales = "销售" in query or "sales" in query.lower()
    has_sentiment = "评论" in query or "情感" in query or "sentiment" in query.lower()
    
    # 注意：LangGraph 1.0 不支持动态 Send，所以我们用条件边控制
    # 但无法在一个节点返回多个目标 → 改为：先分发到一个“启动器”，再并行
    # 更简单做法：假设最多两个任务，显式建图
    return "start_parallel"

def start_parallel(state: AgentState) -> dict:
    """占位节点，实际并行由图结构实现"""
    return {}

def analyze_sales(state: AgentState) -> dict:
    query = state["messages"][0].content
    result = tools["analyze_sales_data"].invoke(query)
    return {"sales_result": result}

def analyze_sentiment_node(state: AgentState) -> dict:
    query = state["messages"][0].content
    result = tools["analyze_sentiment"].invoke(query)
    return {"sentiment_result": result}

def summarizer(state: AgentState) -> dict:
    parts = []
    if state.get("sales_result"):
        parts.append(state["sales_result"])
    if state.get("sentiment_result"):
        parts.append(state["sentiment_result"])
    
    full_text = "\n".join(parts)
    # 使用工具摘要（或直接拼接）
    summary = f"【综合简报】\n{full_text}\n\n建议：继续保持产品品质，加大营销投入。"
    return {"final_report": summary}

# ----------------------------
# 构建图（LangGraph 1.0 风格）
# ----------------------------
builder = StateGraph(AgentState)

# 添加所有节点
builder.add_node("start_parallel", start_parallel)
builder.add_node("analyze_sales", analyze_sales)
builder.add_node("analyze_sentiment", analyze_sentiment_node)
builder.add_node("summarize", summarizer)

# 入口
builder.add_edge(START, "start_parallel")

# 从 start_parallel 分叉到两个并行节点
builder.add_edge("start_parallel", "analyze_sales")
builder.add_edge("start_parallel", "analyze_sentiment")

# 两个分析完成后汇聚到 summarize
# LangGraph 1.0 会自动等待所有上游完成（因为是 DAG）
builder.add_edge("analyze_sales", "summarize")
builder.add_edge("analyze_sentiment", "summarize")

builder.add_edge("summarize", END)

# 编译
app = builder.compile()

# ----------------------------
# 运行
# ----------------------------
if __name__ == "__main__":
    user_input = "请同时分析销售数据和用户评论情感，并汇总成一份简报。"
    inputs = {"messages": [HumanMessage(content=user_input)]}
    
    result = app.invoke(inputs)
    
    print("\n=== 最终报告 ===")
    print(result.get("final_report", "未生成报告"))

模块化拆分

project/
├── main.py
├── schemas.py
├── agents/
│ ├── init.py
│ ├── tools.py
│ └── nodes.py
└── graph/
├── init.py
└── workflow.py

schemas.py

是整个系统的唯一数据源（Single Source of Truth）
使用 Annotated 声明合并策略（如 operator.add 实现消息累积）

from typing import TypedDict, Annotated, Optional
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    sales_result: Optional[str]
    sentiment_result: Optional[str]
    final_report: Optional[str]

agent/tools.py

原子能力单元：
使用 LangChain 的 @tool 装饰器封装业务逻辑
每个工具必须有 docstring（供 LLM 理解用途）

from langchain.tools import tool

@tool
def analyze_sales_data(query: str) -> str:
    """分析销售数据趋势..."""
    return "【销售分析】Q4销售额同比增长25%..."

@tool
def analyze_sentiment(text: str) -> str:
    """分析用户评论的情感倾向..."""
    return "【情感分析】用户评论整体积极..."

TOOLS = {
    "analyze_sales_data": analyze_sales_data,
    "analyze_sentiment": analyze_sentiment,
}

agent/nodes.py

Agent 行为节点：
每个函数是一个状态转换器：state → partial_update
从全局 state 读取输入，返回要更新的字段

from .tools import TOOLS
from typing import Dict, Any

def start_parallel(state: Dict[str, Any]) -> Dict[str, Any]:
    return {}

def analyze_sales(state: Dict[str, Any]) -> Dict[str, Any]:
    query = state["messages"][0].content
    result = TOOLS["analyze_sales_data"].invoke(query)
    return {"sales_result": result}

def analyze_sentiment_node(state: Dict[str, Any]) -> Dict[str, Any]:
    query = state["messages"][0].content
    result = TOOLS["analyze_sentiment"].invoke(query)
    return {"sentiment_result": result}

def summarizer(state: Dict[str, Any]) -> Dict[str, Any]:
    parts = []
    if state.get("sales_result"):
        parts.append(state["sales_result"])
    if state.get("sentiment_result"):
        parts.append(state["sentiment_result"])
    full_text = "\n".join(parts)
    summary = f"【综合简报】\n{full_text}\n\n建议：..."
    return {"final_report": summary}

graph/workflow.py 流程编排

from .tools import TOOLS
from typing import Dict, Any

def start_parallel(state: Dict[str, Any]) -> Dict[str, Any]:
    return {}

def analyze_sales(state: Dict[str, Any]) -> Dict[str, Any]:
    query = state["messages"][0].content
    result = TOOLS["analyze_sales_data"].invoke(query)
    return {"sales_result": result}

def analyze_sentiment_node(state: Dict[str, Any]) -> Dict[str, Any]:
    query = state["messages"][0].content
    result = TOOLS["analyze_sentiment"].invoke(query)
    return {"sentiment_result": result}

def summarizer(state: Dict[str, Any]) -> Dict[str, Any]:
    parts = []
    if state.get("sales_result"):
        parts.append(state["sales_result"])
    if state.get("sentiment_result"):
        parts.append(state["sentiment_result"])
    full_text = "\n".join(parts)
    summary = f"【综合简报】\n{full_text}\n\n建议：..."
    return {"final_report": summary}

main.py

from dotenv import load_dotenv
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from graph.workflow import create_workflow

# ----------------------------
# 加载环境变量 & 初始化 LLM
# ----------------------------
load_dotenv()
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model=os.getenv("DEFAULT_LLM_MODEL_NAME"),
    api_key=os.getenv("DEFAULT_LLM_API_KEY"),
    base_url=os.getenv("DEFAULT_LLM_BASE_URL"),
    temperature=0,
)


if __name__ == "__main__":
    app = create_workflow()
    ## 运行调用
    inputs = {"messages": [HumanMessage(content="请同时分析销售和评论...")]}
    result = app.invoke(inputs)
    print("\n=== 最终报告 ===")
    print(result.get("final_report", "未生成报告"))

总结

langgraph框架下，简单的实现了一个分析报告agent,主要模块包括：

环境与LLM初始化
工具定义
状态模型定义
节点函数设计
图构建
运行调用

下一个项目中，我们会增加记忆存储，多agent协同等功能；

算法题

【最长公共子串】 vs 【最长公共子序列】

考察动态规划知识；

区分：如果题目强调 “连续” → 子串；如果说 “顺序不变，可删除字符” → 子序列

最长公共子串（Substring）
使用动态规划，dp[i][j] 表示以 A[i-1] 和 B[j-1] 结尾的公共子串长度
只有当 A[i-1] == B[j-1] 时，才能延续之前的匹配：
dp[i][j] = dp[i-1][j-1] + 1
否则 dp[i][j] = 0（因为不连续就断了）

dp = [[0]*(n+1) for _ in range(m+1)]
max_len = 0
end_pos = 0
for i in range(1, m+1):
    for j in range(1, n+1):
        if A[i-1] == B[j-1]:
            dp[i][j] = dp[i-1][j-1] + 1
            if dp[i][j] > max_len:
                max_len = dp[i][j]
                end_pos = i
        else:
            dp[i][j] = 0  # 关键！不连续就归零

最长公共子序列（Subsequence）
✅ 核心思想：
dp[i][j] 表示 A[:i] 和 B[:j] 的 LCS 长度
转移方程：

if A[i-1] == B[j-1]:
    dp[i][j] = dp[i-1][j-1] + 1
else:
    dp[i][j] = max(dp[i-1][j], dp[i][j-1])

维度	最长公共子串	最长公共子序列
连续性	必须连续	不必连续
DP 状态含义	以 i,j 结尾的连续匹配长度	前 i,j 个字符的最优解
不匹配时处理	`dp[i][j] = 0`（断开）	`dp[i][j] = max(上, 左)`（继承）
结果位置	需记录结束位置以回溯子串	通常只需长度，回溯较复杂
最大长度	≤ min(len(A), len(B))	可接近 min(len(A), len(B))，甚至等于
应用场景	文本相似片段检测、DNA 序列局部匹配	编辑距离、版本 diff、生物序列比对

夜航星

发表于 2026-02-01

执念的鱼
提着灯闯过远洋的甄选
继续下潜
无需誓言
我的心像自沉的旧母舰
没入深渊

永别啊
我曾凝望
曾是航向的日出
末路残烛
容我吹熄
藏起火种
向宇宙远渡

点燃星
亲手点燃目光尽头的准星
让夜降临
让陨石重启如萤火虫闪烁的飞行
破灭魂灵

点燃星
亲手点燃黑暗森林的火星
蒙昧初醒
而我却轻声告别这新生的黎明

执念的鱼
孤独着闯过自然的甄选
文明岸边
无需誓言
我的心像重启的旧母舰
去星云巅
永别啊
我曾凝望
曾是航向的日出
末路残烛
容我吹熄
藏起火种
向宇宙远渡

我是星
利剑开刃寒光锋芒的银星
绝不消隐
不回顾永难再折返的故园的光阴
决意前进

我是星
我愿投身前途未卜的群星
为梦长明
让希望做我无声永存的墓志铭

我是星
利剑开刃寒光锋芒的银星
绝不消隐
不回顾永难再折返的故园的光阴
决意前进

点燃星
亲手点燃黑暗森林的火星
蒙昧初醒
而我却轻声告别这新生的黎明

预训练

从算法的角度来看，大模型预训练（Large Model Pre-training）是指在大规模无标注或弱标注数据集上，通过自监督学习（Self-supervised Learning）等算法策略，对一个参数量巨大的神经网络模型进行初步参数优化的过程。

其核心目标是让模型学习到通用的数据表征能力（如语言的统计规律、语义结构、上下文关系等），为后续的微调（Fine-tuning）或下游任务（如分类、问答、生成等）提供良好的初始化参数。

下面从几个关键算法维度来诠释大模型预训练：

1. 目标函数设计：自监督学习

大模型预训练通常不依赖人工标注标签，而是通过构造代理任务（pretext tasks）从原始数据中自动生成监督信号。典型例子包括：
语言建模（Language Modeling）：
自回归语言模型（如 GPT 系列）：最大化序列的联合概率: ，使用下一个词预测作为目标。
掩码语言建模（Masked Language Modeling, MLM，如 BERT）：随机掩盖输入中的一些 token，让模型预测被掩盖的内容

对比学习（Contrastive Learning）：
在某些多模态大模型（如 CLIP）中，通过最大化正样本对（如图文对）的相似度，同时最小化负样本对的相似度，构建 InfoNCE 等损失函数。
这些目标函数本质上是在没有人工标签的情况下，从数据本身挖掘结构信息，从而驱动模型学习有意义的表示。

2. 模型架构：Transformer 为核心

当前主流的大模型几乎都基于 Transformer 架构，因其具备以下优势：
并行计算能力强：相比 RNN，Transformer 可以高效利用 GPU/TPU 进行大规模并行训练。
长距离依赖建模：通过自注意力机制（Self-Attention），模型能捕捉任意两个 token 之间的依赖关系。
可扩展性好：通过堆叠更多层、扩大隐藏维度、增加注意力头数等方式，可以轻松扩展至百亿甚至万亿参数规模。

3. 优化算法与训练策略

由于模型规模巨大、数据量庞大，预训练对优化算法和工程实现提出极高要求：
优化器：常用 AdamW（带权重衰减的 Adam），因其在大规模训练中表现稳定。
学习率调度：采用 warmup + decay 策略（如线性 warmup 后余弦退火），避免初期训练不稳定。
梯度裁剪（Gradient Clipping）：防止梯度爆炸。
混合精度训练（Mixed Precision）：使用 FP16/FP8 加速计算并节省显存。
分布式训练：结合数据并行、模型并行、流水线并行（如 ZeRO、Megatron-LM、DeepSpeed）等技术，在数千 GPU 上协同训练。

4. 数据规模与泛化能力

预训练的一个关键假设是：更大的数据 + 更大的模型 → 更强的泛化能力。算法上体现为：
模型在海量文本（如 Common Crawl、The Pile、WuDao 等）上学习到丰富的语言知识（语法、事实、推理模式等）。
遵循 scaling law（缩放律）：在一定范围内，验证损失随模型参数量、数据量、计算量的幂律下降。

5. 预训练 vs 微调的算法角色

预训练阶段：学习通用表示（representation learning），属于“基础能力”构建。
微调阶段：在特定任务的小规模标注数据上，通过有监督学习调整部分或全部参数，实现任务适配。
近年来也出现无需微调的范式（如 Prompting、In-context Learning），但其有效性仍依赖于高质量的预训练。
总结（算法视角）
大模型预训练 = 自监督目标函数 + Transformer 架构 + 大规模优化算法 + 海量数据驱动的表示学习
它本质上是一种无监督表示学习算法，通过在超大规模数据上优化精心设计的代理任务，使模型内化世界知识与结构规律，从而为各类下游任务提供强大的先验能力。

大模型发展史

发表于 2025-06-01 更新于 2026-02-01 分类于 LLM

大模型（Large Language Models, LLMs）的发展史是人工智能从规则驱动走向数据驱动、从专用智能迈向通用智能的关键历程。以下按时间脉络系统梳理其发展过程，涵盖技术演进、代表性模型与关键转折点：

一、前奏阶段：

符号主义 → 统计方法 → 神经网络萌芽（1950s–2017）

1. 符号主义时代（1950s–1980s）
  核心思想：用人工编写的逻辑规则（如 if-then）模拟人类推理。
  代表：专家系统（如DENDRAL化学分析系统）、知识图谱雏形。
  局限：无法处理现实世界的模糊性与复杂性；规则难以穷尽。
1. 统计学习时代（1990s–2012）
  引入概率模型（如n-gram、隐马尔可夫模型 HMM）处理语言。
  2003年：Bengio 提出神经网络语言模型（NNLM），首次将词嵌入（word embedding）引入NLP。
  2013年：Google 发布 Word2Vec，高效学习词向量，奠定语义表征基础。
1. 深度学习兴起（2012–2017）
  2012年：AlexNet 在 ImageNet 夺冠，引爆深度学习浪潮。
  RNN/LSTM 成为主流序列模型，但存在长程依赖弱、训练慢、难并行等问题。
  2014年：Bahdanau 等提出注意力机制（Attention），改善机器翻译对齐问题。
  ✅ 此阶段为大模型奠定了数据（如ImageNet）、算力（GPU）、算法（词嵌入、注意力）三大基础。

二、奠基时刻：

Transformer 诞生（2017）
2017年：Google 发表划时代论文《Attention Is All You Need》，提出 Transformer 架构。
完全基于自注意力机制（Self-Attention），抛弃循环结构。
支持并行训练，高效处理长文本。
成为所有现代大模型的统一架构基础。
🔑 Transformer 是大模型发展的“奇点”——没有它，就没有GPT、BERT或ChatGPT。

三、预训练范式确立（2018–2019）

1. 2018年：双雄并起
  OpenAI 发布 GPT-1（1.17亿参数）：
  基于 Transformer 解码器，采用自回归语言建模。
  验证“预训练 + 微调”范式在多任务上的有效性。
  Google 发布 BERT：
  基于 Transformer 编码器，采用双向掩码语言建模（MLM）。
  在理解类任务（如问答、分类）上大幅刷新SOTA。
  📌 形成两大技术路线：生成式（GPT） vs 判别式（BERT）。
1. 2019年：规模初显
  GPT-2（15亿参数）发布，展示零样本生成能力，因安全顾虑未完全开源。
  XLNet、RoBERTa 等改进模型涌现，优化训练策略。

四、规模化爆发期：能力涌现（2020–2022）

1. 2020年：GPT-3 开启千亿时代
  1750亿参数，训练数据达570GB。
  核心突破：能力涌现（Emergent Abilities） —— 模型在达到一定规模后，自发掌握推理、代码生成、少样本学习等能力。
  引入 Prompt Engineering（提示工程），用户通过自然语言指令引导模型，无需微调。
1. 2021–2022：全球竞速
  Google 发布 PaLM（5400亿参数），验证万亿级潜力。
  Meta 开源 OPT，推动开放生态。
  中国启动大模型战略：百度文心、阿里通义、智谱GLM、深度求索DeepSeek等相继发布。

五、破圈普及期：

ChatGPT 与多模态（2022.11–2023）

1. 2022年11月：ChatGPT 爆发
  基于 GPT-3.5 + RLHF（人类反馈强化学习）。
  实现流畅对话、逻辑连贯、风格可控，月活破亿，引爆全球AI热潮。
  标志：大模型从技术圈走向大众。
1. 2023年：多模态与国产崛起
  GPT-4：支持图文输入，专业考试表现超人类。
  Gemini 1.0（Google）：原生支持文本、图像、音频、视频。
  国产模型集中发布：文心一言、通义千问、豆包、GLM-3 等。
  Llama 2 开源（Meta），推动私有化部署。

六、产业深化期：

降本增效与垂直落地（2024–）

核心趋势：
架构优化：采用 MoE（混合专家）架构（如 Qwen3、GLM-4），在性能与成本间取得平衡。
开源普惠：DeepSeek-R1、Qwen2.5（Apache 2.0）、Llama 3.1 等推动中小企业接入。
行业适配：
医疗：辅助诊断、病历生成
金融：风控、投研报告
政务/工业：智能客服、质检
超长上下文：Gemini 3、GPT-5 支持 1M+ tokens，处理整本书或长会议记录。

七、关键技术演进总结

Transformer 并行化、长程依赖建模的基础架构
预训练 + 微调范式革命，实现知识迁移
Prompt Engineering 降低使用门槛，激活零样本能力
RLHF 对齐人类价值观，提升对话质量
LoRA / QLoRA 高效微调，降低算力需求
MoE 动态激活专家子网络，提升推理效率