框架专题

langchain

"""
LangChain 1.2 LCEL / Runnable 核心特性示例

这个文件尽量用“一个例子”覆盖：

1. Prompt
2. LLM
3. Runnable
4. LCEL Pipe
5. Parallel 并行
6. Router 路由
7. State 传递
8. Retrieval 思想
9. Stream 流式
10. Batch 批量
11. Runnable Graph 思想

安装：

pip install langchain langchain-openai

环境变量：

export OPENAI_API_KEY=xxx
"""

from typing import TypedDict

from langchain_openai import ChatOpenAI

from langchain_core.prompts import ChatPromptTemplate

from langchain_core.output_parsers import (
    StrOutputParser,
)

from langchain_core.runnables import (
    RunnableLambda,
    RunnableParallel,
    RunnableBranch,
)

# =========================================================
# 1. 初始化 LLM
# =========================================================

# LangChain 里：
#
# ChatOpenAI 本身就是 Runnable
#
# 所以：
#
# prompt | llm | parser
#
# 本质上是：
#
# Runnable -> Runnable -> Runnable
#

llm = ChatOpenAI(
    model="gpt-4.1-mini",
    temperature=0,
)

# =========================================================
# 2. Prompt Template
# =========================================================

# PromptTemplate 也是 Runnable
#
# 输入：
# {
#   "input": "xxx"
# }
#
# 输出：
# ChatPromptValue
#

base_prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        "You are a helpful assistant."
    ),
    (
        "human",
        "{input}"
    )
])

# =========================================================
# 3. 定义 State
# =========================================================

# LangChain LCEL 虽然不像 LangGraph
# 强制要求 state schema
#
# 但实际工程里：
#
# 仍然推荐明确 state 结构
#

class State(TypedDict):

    # 用户输入
    input: str

# =========================================================
# 4. RunnableLambda
# =========================================================

# RunnableLambda：
#
# 最常见 Runnable
#
# 用来：
#
# - 数据清洗
# - state 转换
# - retrieval
# - tool 包装
# - 路由逻辑
#
# 本质：
#
# 普通 python 函数
# -> Runnable
#

clean_input = RunnableLambda(

    # x 就是 state
    lambda x: {

        # strip 去掉前后空格
        "input": x["input"].strip()

    }
)

# =========================================================
# 5. RunnableParallel（并行 Fan-out）
# =========================================================

# RunnableParallel：
#
# 会并行执行多个 Runnable
#
# 类似：
#
#            /-> A
# input ----|
#            \-> B
#
# 非常适合：
#
# - 多路检索
# - 多 Agent
# - 多 Prompt
# - 多模型
#

parallel = RunnableParallel({

    # -----------------------------------------------------
    # 分支1：原始问题
    # -----------------------------------------------------

    "question":

        RunnableLambda(
            lambda x: x["input"]
        ),

    # -----------------------------------------------------
    # 分支2：分类器
    # -----------------------------------------------------

    # 这里演示：
    #
    # Prompt
    #   -> LLM
    #   -> Parser
    #   -> Lambda
    #
    # 的完整 LCEL 链式调用
    #

    "category":

        base_prompt
        | llm
        | StrOutputParser()

        # 简单分类
        | RunnableLambda(
            lambda text: (

                "code"

                if "python" in text.lower()

                else "general"
            )
        ),

    # -----------------------------------------------------
    # 分支3：模拟 Retrieval
    # -----------------------------------------------------

    # 实际生产里：
    #
    # 这里通常会：
    #
    # query
    # -> embedding
    # -> vector db
    # -> rerank
    #
    # 这里简化一下
    #

    "context":

        RunnableLambda(
            lambda x:
                f"mock retrieval result for: {x['input']}"
        )
})

# =========================================================
# 6. 两个不同的下游 Chain
# =========================================================

# ---------------------------------------------------------
# code chain
# ---------------------------------------------------------

# 如果是代码问题：
#
# 使用“工程师风格”回答
#

code_chain = (

    ChatPromptTemplate.from_template(
        """
You are a senior engineer.

Question:
{question}
"""
    )

    | llm

    | StrOutputParser()
)

# ---------------------------------------------------------
# general chain
# ---------------------------------------------------------

general_chain = (

    ChatPromptTemplate.from_template(
        """
Answer normally.

Question:
{question}
"""
    )

    | llm

    | StrOutputParser()
)

# =========================================================
# 7. RunnableBranch（Router）
# =========================================================

# RunnableBranch：
#
# LangChain 的“路由器”
#
# 类似：
#
# if else
#
# 或：
#
# AI Router
#

router = RunnableBranch(

    # -----------------------------------------------------
    # 条件1
    # -----------------------------------------------------

    (
        # 条件函数
        lambda x:

            x["category"] == "code",

        # 满足条件走这里
        code_chain,
    ),

    # -----------------------------------------------------
    # 默认分支
    # -----------------------------------------------------

    general_chain
)

# =========================================================
# 8. 最终 Chain
# =========================================================

# LCEL 核心：
#
# 使用 | 组合 Runnable
#
# 本质：
#
# DAG（有向图）
#

chain = (

    # step1
    clean_input

    # step2
    | parallel

    # step3
    #
    # 合并 state
    #
    | RunnableLambda(

        lambda x: {

            # 保留原字段
            **x,

            # 拼接 context
            "question":

                f"""
User Question:
{x['question']}

Retrieved Context:
{x['context']}
"""
        }
    )

    # step4
    | router
)

# =========================================================
# 9. invoke（普通调用）
# =========================================================

print("\n================ invoke ================\n")

result = chain.invoke({

    "input":
        "How to write python retry logic?"

})

print(result)

# =========================================================
# 10. stream（流式）
# =========================================================

# stream：
#
# token 流式输出
#
# 对聊天 UI 很重要
#

print("\n================ stream ================\n")

for chunk in chain.stream({

    "input":
        "Explain python async"

}):

    print(chunk, end="", flush=True)

# =========================================================
# 11. batch（批量）
# =========================================================

# batch：
#
# 批量并发执行
#
# 适合：
#
# - embedding
# - evaluation
# - dataset processing
#

print("\n\n================ batch ================\n")

results = chain.batch([

    {"input": "What is Rust?"},

    {"input": "Python multiprocessing"},

    {"input": "Explain PostgreSQL index"},
])

for i, r in enumerate(results):

    print(f"\n--- result {i} ---\n")

    print(r)

问题：怎么用 langchain 的链式调用简化流程

langchain 1.2+ 有 lcel，有 Runnable

先定义组件，如 prompt、message、llmClient、输出解析器

通过管道连接这些组件

langchain 解决了大部分边界代码，不用再写错误处理、输出解析、流式处理等同时他做了消息格式化，不用关心模型间的区别

问题：有哪些核心组件

model retrieval chain memery agent callbacks LCEL

问题：langchain agent 有哪几类

toolcall ReAct plan router

langgraph

问题：与 chain 的区别

langchain 是构造链路的，解决链接问题

langgraph 是构造图，解决控制问题，他有state的概念，模型可维护state来控制执行链路，提升执行效果

autogen

# pip install pyautogen

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# =========================================
# 1. LLM 配置（所有 agent 共用）
# =========================================
llm_config = {
    "model": "gpt-4o-mini",
    "api_key": "YOUR_API_KEY"
}

# =========================================
# 2. 定义 LLM Agent（负责思考）
# =========================================
planner = AssistantAgent(
    name="planner",
    llm_config=llm_config,
    system_message="你是任务规划者，负责拆解任务。"
)

coder = AssistantAgent(
    name="coder",
    llm_config=llm_config,
    system_message="你是程序员，只负责编写代码。"
)

reviewer = AssistantAgent(
    name="reviewer",
    llm_config=llm_config,
    system_message="你是代码审查者，负责发现问题并改进。"
)

# =========================================
# 3. UserProxyAgent（执行器 + 工具人）
#    👉 可以执行代码 / 充当人类 / sandbox
# =========================================
executor = UserProxyAgent(
    name="executor",
    human_input_mode="NEVER",  # 自动执行
    code_execution_config={"work_dir": "workdir"}
)

# =========================================
# 4. 多 Agent 群聊（核心：协作机制）
# =========================================
groupchat = GroupChat(
    agents=[executor, planner, coder, reviewer],
    messages=[],
    max_round=8  # 防止无限循环
)

manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)

# =========================================
# 5. 启动任务
# =========================================
executor.initiate_chat(
    manager,
    message="写一个 Python 快排，并解释复杂度"
)

问题：是什么，解决什么问题

多智能体框架通过“对话”驱动 agent 协作用于复杂任务拆解、协作、工具调用

问题：如何实现 multi-agent 协作？

首先会定义不同角色的 agent，比如一个负责规划，一个负责执行，一个负责写代码，一个负责检查结果。

然后把它们放到一个 GroupChat 里，让它们像在一个群聊里一样互相发消息。

关键点不是“谁调用谁”，而是：

谁说完话，下一步由 manager 决定该轮到哪个 agent 回应。

问题：如何避免无限对话循环？

第一层是硬限制，比如 max_turns，直接限制最多对话多少轮，到点就结束。

第二层是 termination condition，比如某个 agent 明确说“任务完成”或者输出了结束信号，就主动停止。

第三层是策略控制，比如 manager 做判断，如果发现重复循环、或者没有新信息产生，就强制终止或者跳出循环。

crewai

问题：是什么

CrewAI 本质上就是一个多智能体框架，它的设计思路特别像“一个小团队在做项目”。

CrewAI 本质是一个“基于角色分工的 AI 工作流框架”，通过 agent + task + process，把复杂任务拆成类似团队协作的流水线执行。

问题：CrewAI 和 AutoGen 有什么区别？

AutoGen 更像“群聊”，所有 agent 在一个聊天室里自由对话，manager 决定谁说话。

CrewAI 更像“公司组织结构”，是很明确的流程 and 分工。

简单说：

AutoGen：对话驱动（比较自由） CrewAI：任务驱动（结构更清晰）

CrewAI 更偏工程化 workflow。

问题：CrewAI 如何实现 agent 协作？

它不是像 AutoGen 那种聊天式协作。

而是：

每个 agent 都有明确任务，然后通过 Task 来传递结果。

上一个 agent 的 output 会成为下一个 agent 的 input。

langchain​

langgraph​

autogen​

crewai​

langchain

langgraph

autogen

crewai