LangChain agent 开发框架学习笔记
·
1. 简介
LLM 大模型应用开发框架, 用来构建 agent.
LangChain 提供统一接口,可连接:大模型、Prompt、向量数据库、工具调用、记忆系统以及 Agent 工作流.
它是 python 语言, 虽然也有 js 版本叫 LangChain.js, 但生态和更新速度都不如 Python.
2. 核心组件
LLM:连接 OpenAI、Claude、Gemini 等大模型
PromptTemplate:管理 Prompt 模板
Chains:构建多步骤 AI 工作流
Memory:实现多轮对话记忆
Tools:调用搜索、数据库、API 等工具
Agents:让 AI 自动决策与执行任务
Vector Store:连接向量数据库实现 RAG
3. 极简 demo
最简单的 LangChain + 阿里云大模型 Demo
import sys
sys.stdout.reconfigure(encoding="utf-8")
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompt_values import ChatPromptValue
from langchain_core.messages import AIMessage
load_dotenv()
llm = ChatOpenAI(model="qwen-plus")
def demo_lcel():
"""写法一:LCEL 管道风格(推荐)"""
chain = (
ChatPromptTemplate.from_messages([("human", "{question}")])
| llm
| StrOutputParser()
)
result = chain.invoke({"question": "你好,请用一句话介绍你自己"})
print(result)
def demo_step_by_step() -> None:
"""写法二:普通逐步调用"""
prompt: ChatPromptTemplate = ChatPromptTemplate.from_messages([("human", "{question}")])
parser: StrOutputParser = StrOutputParser()
# 第一步:模板填充变量,生成提示词
prompt_value: ChatPromptValue = prompt.invoke({"question": "你好,请用一句话介绍你自己"})
# 第二步:调用 LLM,返回 AIMessage 对象
ai_message: AIMessage = llm.invoke(prompt_value)
# 第三步:解析器提取纯文本
result: str = parser.invoke(ai_message)
print(result)
# demo_lcel()
demo_step_by_step()
4. llm.invoke() 调用链
4.1 http 请求
背后是 httpx 框架的 http post 请求.
ChatOpenAI.invoke() ← langchain-core 入口
└→ _generate() ← langchain_openai/chat_models/base.py:1695
└→ self.client.with_raw_response.create(**payload) ← 进入 openai SDK
└→ SyncAPIClient.request() ← openai/_base_client.py:995
└→ _send_request() ← openai/_base_client.py:957
└→ self._client.send(request) ← httpx.Client.send() 🔥 真正的 HTTP 请求
4.2 LLM 传参
调用 LLM 的相关参数见下.
# 位于 site-packages/openai/resources/chat/completions/completions.py
class Completions(SyncAPIResource):
@required_args(["messages", "model"], ["messages", "model", "stream"])
def create(...):
return self._post(
"/chat/completions",
body=maybe_transform(
{
"messages": messages,
"model": model,
"audio": audio,
"frequency_penalty": frequency_penalty,
"function_call": function_call,
"functions": functions,
"logit_bias": logit_bias,
"logprobs": logprobs,
"max_completion_tokens": max_completion_tokens,
"max_tokens": max_tokens,
"metadata": metadata,
"modalities": modalities,
"moderation": moderation,
"n": n,
"parallel_tool_calls": parallel_tool_calls,
"prediction": prediction,
"presence_penalty": presence_penalty,
"prompt_cache_key": prompt_cache_key,
"prompt_cache_options": prompt_cache_options,
"prompt_cache_retention": prompt_cache_retention,
"reasoning_effort": reasoning_effort,
"response_format": response_format,
"safety_identifier": safety_identifier,
"seed": seed,
"service_tier": service_tier,
"stop": stop,
"store": store,
"stream": stream,
"stream_options": stream_options,
"temperature": temperature,
"tool_choice": tool_choice,
"tools": tools,
"top_logprobs": top_logprobs,
"top_p": top_p,
"user": user,
"verbosity": verbosity,
"web_search_options": web_search_options,
},
completion_create_params.CompletionCreateParamsStreaming
if stream
else completion_create_params.CompletionCreateParamsNonStreaming,
),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
security={"bearer_auth": True},
),
cast_to=ChatCompletion,
stream=stream or False,
stream_cls=Stream[ChatCompletionChunk],
)
附录
更多推荐

所有评论(0)