Python LangChain模块深度解析:AI应用开发框架,简化大模型应用开发

LangChain生态系统（2026年）
├── LangChain（基础框架）
│   ├── 模型抽象：统一接口支持100+模型供应商
│   ├── 提示工程：模板化管理与优化
│   ├── 工具集成：数百种外部服务连接
│   └── 记忆管理：多种策略解决LLM无状态性
├── LangGraph（运行时编排）
│   ├── 图结构建模：有向图表示智能体系统
│   ├── 多智能体协作：研究→分析→报告全流程
│   └── 可视化调试：直观呈现执行路径
├── DeepAgents（智能体外骨骼）
│   ├── 任务规划：分解长任务为子目标
│   ├── 虚拟文件系统：高效管理大输出
│   └── 子智能体委托：动态创建专项Agent
└── LangSmith（监控平台）
    ├── 追踪：记录每个步骤和工具调用
    ├── 评估：自动化质量检查
    └── 部署：生产环境管理

# 安装依赖：pip install langchain langchain-openai
import os
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain

# 设置OpenAI API密钥（实际使用时请替换为您的密钥）
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# 1. 创建大语言模型实例
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# 2. 定义提示模板
prompt_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的Python技术专家"),
    ("user", "请用简洁的语言解释{concept}的概念，并给出一个应用示例")
])

# 3. 创建链
chain = LLMChain(llm=llm, prompt=prompt_template)

# 4. 运行链
result = chain.invoke({"concept": "Python装饰器"})
print(f"AI回复：{result['text']}")

# 输出示例：
# AI回复：Python装饰器是一种特殊函数，用于修改其他函数的行为...

from langchain.chains import SequentialChain, LLMChain
from langchain.prompts import PromptTemplate

# 第一步：生成问题
problem_template = PromptTemplate(
    input_variables=["topic"],
    template="请生成一个关于{topic}的编程问题，难度适中"
)

# 第二步：生成解决方案
solution_template = PromptTemplate(
    input_variables=["problem"],
    template="针对以下编程问题，提供Python解决方案：{problem}"
)

# 第三步：代码优化建议
optimize_template = PromptTemplate(
    input_variables=["solution"],
    template="分析以下Python代码，提出3条优化建议：{solution}"
)

# 创建三个链
problem_chain = LLMChain(llm=llm, prompt=problem_template, output_key="problem")
solution_chain = LLMChain(llm=llm, prompt=solution_template, output_key="solution")
optimize_chain = LLMChain(llm=llm, prompt=optimize_template, output_key="optimizations")

# 组合成顺序链
multi_step_chain = SequentialChain(
    chains=[problem_chain, solution_chain, optimize_chain],
    input_variables=["topic"],
    output_variables=["problem", "solution", "optimizations"],
    verbose=True# 显示执行过程
)

# 运行多步推理
result = multi_step_chain.invoke({"topic": "数据可视化"})
print(f"生成的问题：{result['problem']}")
print(f"解决方案：{result['solution']}")
print(f"优化建议：{result['optimizations']}")

from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain.tools import DuckDuckGoSearchRun, PythonREPLTool
from langchain import hub

# 定义可用工具
search = DuckDuckGoSearchRun()
python_repl = PythonREPLTool()

tools = [
    Tool(
        name="Search",
        func=search.run,
        description="用于搜索最新信息，适用于事实性问题和最新动态"
    ),
    Tool(
        name="PythonREPL",
        func=python_repl.run,
        description="用于执行Python代码，适用于计算、数据处理和代码验证"
    )
]

# 从LangChain Hub获取ReAct提示模板
prompt = hub.pull("hwchase17/react")

# 创建代理
agent = create_react_agent(llm, tools, prompt)

# 创建执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=5# 限制最大迭代次数，避免无限循环
)

# 运行代理处理复杂任务
task = "分析最近一周Python在AI领域的发展趋势，并生成一个简单的数据可视化代码"
result = agent_executor.invoke({"input": task})
print(f"最终结果：{result['output']}")

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 创建记忆实例
memory = ConversationBufferMemory(
    return_messages=True,  # 返回消息对象而非字符串
    memory_key="chat_history"# 记忆在上下文中的键名
)

# 创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 多轮对话演示
print("第一轮对话：")
response1 = conversation.invoke("介绍一下Python的异步编程")
print(f"AI：{response1['response']}")

print("\n第二轮对话（基于历史）：")
response2 = conversation.invoke("刚才提到的asyncio有什么优缺点？")
print(f"AI：{response2['response']}")

print("\n第三轮对话（继续深入）：")
response3 = conversation.invoke("如何避免异步编程中的常见陷阱？")
print(f"AI：{response3['response']}")

# 查看完整对话历史
print("\n完整对话历史：")
for i, msg in enumerate(memory.chat_history.messages):
    print(f"{i+1}. {msg.type}: {msg.content[:100]}...")

from langchain.tools import BaseTool
from typing import Optional
import requests

# 自定义工具：天气查询
classWeatherTool(BaseTool):
    name = "weather_checker"
    description = "查询指定城市的当前天气情况"

def_run(self, city: str) -> str:
"""实际执行逻辑"""
try:
# 模拟API调用（实际使用时请替换为真实天气API）
# 这里使用模拟数据演示
            weather_data = {
"北京": {"temp": "18°C", "condition": "晴朗", "humidity": "45%"},
"上海": {"temp": "22°C", "condition": "多云", "humidity": "65%"},
"深圳": {"temp": "26°C", "condition": "小雨", "humidity": "85%"}
            }

if city in weather_data:
                data = weather_data[city]
returnf"{city}当前天气：温度{data['temp']}，{data['condition']}，湿度{data['humidity']}"
else:
returnf"抱歉，未找到{city}的天气信息"

except Exception as e:
returnf"查询天气时出错：{str(e)}"

def_arun(self, city: str):
"""异步版本（可选）"""
raise NotImplementedError("此工具不支持异步")

# 使用自定义工具
weather_tool = WeatherTool()

# 创建包含自定义工具的代理
from langchain.agents import initialize_agent

custom_tools = [weather_tool] + tools  # 组合自定义工具和内置工具

custom_agent = initialize_agent(
    tools=custom_tools,
    llm=llm,
    agent="zero-shot-react-description",
    verbose=True
)

# 执行包含天气查询的任务
weather_task = "查询北京的天气，并根据天气建议是否适合户外运动"
result = custom_agent.invoke(weather_task)
print(f"任务结果：{result['output']}")

# 多模型配置示例（需要相应API密钥）
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

# 初始化不同模型（实际使用时需要相应API密钥）
models = {
"openai": ChatOpenAI(model="gpt-4o", temperature=0.7),
"anthropic": ChatAnthropic(model="claude-3-opus-20240229", temperature=0.5),
"google": ChatGoogleGenerativeAI(model="gemini-pro", temperature=0.6)
}

# 统一接口调用不同模型
defask_model(model_name, question):
    model = models.get(model_name)
ifnot model:
return"模型未配置"

try:
        response = model.invoke(question)
return response.content
except Exception as e:
returnf"调用{model_name}失败：{str(e)}"

# 测试不同模型
test_question = "用Python实现快速排序算法，并解释其工作原理"

for model_name in ["openai", "anthropic", "google"]:
    print(f"\n{model_name.upper()}模型回答：")
    answer = ask_model(model_name, test_question)
    print(answer[:200] + "...")  # 只显示前200字符

from langchain.prompts import ChatPromptTemplate

chat_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的Python开发助手"),
    ("user", "请解释{concept}的概念"),
    ("assistant", "好的，我来解释{concept}..."),
    ("user", "{question}")
])

formatted_prompt = chat_template.format_messages(
    concept="Python装饰器",
    question="装饰器在Web开发中有哪些应用？"
)

from langchain.prompts import MessagesPlaceholder

dynamic_template = ChatPromptTemplate.from_messages([
    MessagesPlaceholder(variable_name="history"),
    ("user", "{input}")
])

from langchain.prompts import PromptTemplate

simple_template = PromptTemplate(
    input_variables=["topic", "difficulty"],
    template="生成一个关于{topic}的{difficulty}难度编程问题"
)

from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List

classCodeExample(BaseModel):
    language: str = Field(description="编程语言")
    code: str = Field(description="完整代码")
    explanation: str = Field(description="代码解释")
    complexity: str = Field(description="时间复杂度")

parser = PydanticOutputParser(pydantic_object=CodeExample)

from langchain.output_parsers import JsonOutputParser

json_parser = JsonOutputParser()

# 使用时指定输出格式
from langchain.prompts import PromptTemplate

template = """请以JSON格式返回以下信息：
{format_instructions}

问题：{question}
"""

prompt = PromptTemplate(
    template=template,
    input_variables=["question"],
    partial_variables={"format_instructions": json_parser.get_format_instructions()}
)

from langchain.output_parsers import StructuredOutputParser, ResponseSchema

response_schemas = [
    ResponseSchema(name="function_name", description="函数名称"),
    ResponseSchema(name="code", description="完整代码"),
    ResponseSchema(name="time_complexity", description="时间复杂度")
]

output_parser = StructuredOutputParser.from_response_schemas(response_schemas)

from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate
from langchain.output_parsers import PydanticOutputParser

# 1. 定义输出模型
classAlgorithmExplanation(BaseModel):
    name: str = Field(description="算法名称")
    python_code: str = Field(description="Python实现代码")
    explanation: str = Field(description="算法原理解释")
    use_cases: List[str] = Field(description="适用场景列表")

# 2. 创建解析器
parser = PydanticOutputParser(pydantic_object=AlgorithmExplanation)

# 3. 构建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个算法专家，请按照指定格式回答。"),
    ("user", "请解释{algorithm}算法：\n{format_instructions}")
])

# 4. 创建链
chain = LLMChain(
    llm=llm,
    prompt=prompt,
    output_parser=parser
)

# 5. 运行
result = chain.invoke({
"algorithm": "快速排序",
"format_instructions": parser.get_format_instructions()
})

print(f"算法名称：{result['text'].name}")
print(f"代码：\n{result['text'].python_code}")

from langchain.document_loaders import PyPDFLoader

loader = PyPDFLoader("example.pdf")
documents = loader.load()
print(f"加载了{len(documents)}页PDF文档")

from langchain.document_loaders import Docx2txtLoader

loader = Docx2txtLoader("report.docx")
documents = loader.load()

from langchain.document_loaders import WebBaseLoader

loader = WebBaseLoader(["https://example.com"])
documents = loader.load()

from langchain.document_loaders import UnstructuredMarkdownLoader

loader = UnstructuredMarkdownLoader("README.md")
documents = loader.load()

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "，", " ", ""]
)

chunks = text_splitter.split_documents(documents)
print(f"原始文档分割为{len(chunks)}个文本块")

from langchain.text_splitter import TokenTextSplitter

token_splitter = TokenTextSplitter(
    chunk_size=1000,
    chunk_overlap=100
)

token_chunks = token_splitter.split_documents(documents)

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

# 创建嵌入
embeddings = OpenAIEmbeddings()

# 创建向量存储
vectorstore = FAISS.from_documents(chunks, embeddings)

# 相似性检索
query = "Python异步编程的优势"
docs = vectorstore.similarity_search(query, k=3)

for i, doc in enumerate(docs):
    print(f"结果{i+1}: {doc.page_content[:100]}...")

from langchain.vectorstores import Chroma

chroma_store = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 持久化存储
chroma_store.persist()

# 加载已有存储
loaded_store = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

from langchain.chains import RetrievalQA
from langchain.memory import ConversationBufferMemory

classRAGSystem:
"""完整的RAG系统实现"""

def__init__(self, document_paths, embedding_model="text-embedding-ada-002"):
        self.embeddings = OpenAIEmbeddings(model=embedding_model)
        self.documents = self.load_documents(document_paths)
        self.vectorstore = self.create_vectorstore()
        self.memory = ConversationBufferMemory(
            return_messages=True,
            memory_key="chat_history"
        )

# 创建RAG链
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=ChatOpenAI(model="gpt-4o", temperature=0.3),
            chain_type="stuff",
            retriever=self.vectorstore.as_retriever(search_kwargs={"k": 4}),
            memory=self.memory,
            return_source_documents=True
        )

defload_documents(self, paths):
"""加载多种格式文档"""
from langchain.document_loaders import (
            PyPDFLoader, Docx2txtLoader, 
            TextLoader, CSVLoader
        )

        all_docs = []
for path in paths:
if path.endswith('.pdf'):
                loader = PyPDFLoader(path)
elif path.endswith('.docx'):
                loader = Docx2txtLoader(path)
elif path.endswith('.txt'):
                loader = TextLoader(path)
elif path.endswith('.csv'):
                loader = CSVLoader(path)
else:
continue

            docs = loader.load()
            all_docs.extend(docs)

return all_docs

defcreate_vectorstore(self):
"""创建向量存储"""
# 分割文本
from langchain.text_splitter import RecursiveCharacterTextSplitter

        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )

        chunks = text_splitter.split_documents(self.documents)

# 创建向量存储
        vectorstore = FAISS.from_documents(chunks, self.embeddings)
return vectorstore

defquery(self, question):
"""查询知识库"""
        result = self.qa_chain.invoke({"query": question})

return {
"answer": result["result"],
"sources": [
f"{doc.metadata.get('source', '未知')} (第{doc.metadata.get('page', 'N/A')}页)"
for doc in result["source_documents"]
            ]
        }

# 使用示例
rag_system = RAGSystem([
"docs/project_guide.pdf",
"docs/api_reference.docx",
"docs/faq.txt"
])

response = rag_system.query("项目部署有哪些注意事项？")
print(f"回答：{response['answer']}")
print(f"来源：{response['sources']}")
## 六、高级用法与性能优化

### 4.1 流式输出与实时交互

LangChain支持流式输出，显著提升用户体验，减少等待时间。

#### 示例7：流式输出实现

```python
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# 创建支持流式输出的模型
streaming_llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.7,
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

# 创建对话链（流式版本）
streaming_conversation = ConversationChain(
    llm=streaming_llm,
    memory=ConversationBufferMemory(),
    verbose=False
)

print("开始流式对话（输入'退出'结束）：")
whileTrue:
    user_input = input("\n你：")
if user_input.lower() == '退出':
break

    print("AI：", end="")
    response = streaming_conversation.invoke(user_input)
    print()  # 换行

from langchain.memory import ConversationSummaryMemory
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# 创建总结记忆
summary_memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="chat_history",
    return_messages=True
)

# 定义提示模板
summary_prompt = PromptTemplate(
    input_variables=["chat_history", "input"],
    template="基于以下对话历史：{chat_history}\n回答用户问题：{input}"
)

# 创建链
summary_chain = LLMChain(
    llm=llm,
    prompt=summary_prompt,
    memory=summary_memory,
    verbose=True
)

# 模拟长对话
long_conversation = [
"Python有哪些主要的数据类型？",
"列表和元组有什么区别？",
"字典的键有什么要求？",
"集合在什么场景下使用？",
"这些数据类型的内存占用如何？"
]

for i, question in enumerate(long_conversation):
    print(f"\n第{i+1}轮提问：{question}")
    response = summary_chain.invoke({"input": question})
    print(f"回答：{response['text'][:100]}...")

# 查看压缩后的记忆
print("\n压缩后的记忆摘要：")
print(summary_memory.buffer[:500])

from langchain.callbacks import CallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from tenacity import retry, stop_after_attempt, wait_exponential

# 使用tenacity实现重试机制
@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=4, max=10)  # 指数退避
)
defrobust_chain_invoke(chain, input_data):
"""增强的链调用，包含重试机制"""
try:
return chain.invoke(input_data)
except Exception as e:
        print(f"调用失败：{str(e)}，正在重试...")
raise# 触发重试

# 创建带重试的链
callback_manager = CallbackManager([StreamingStdOutCallbackHandler()])

robust_llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.7,
    callback_manager=callback_manager
)

# 实际使用
try:
    result = robust_chain_invoke(
        LLMChain(llm=robust_llm, prompt=prompt_template),
        {"concept": "机器学习"}
    )
    print(f"最终结果：{result['text']}")
except Exception as e:
    print(f"所有重试均失败：{str(e)}")

# 完整实现：智能客服系统
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter

classSmartCustomerService:
"""智能客服系统"""

def__init__(self, documents, api_key):
        self.api_key = api_key
        os.environ["OPENAI_API_KEY"] = api_key

# 1. 文本分割
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )
        texts = text_splitter.split_documents(documents)

# 2. 创建向量数据库
        embeddings = OpenAIEmbeddings()
        self.vectorstore = FAISS.from_documents(texts, embeddings)

# 3. 创建检索链
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=ChatOpenAI(model="gpt-4o", temperature=0.3),
            chain_type="stuff",
            retriever=self.vectorstore.as_retriever(search_kwargs={"k": 3}),
            return_source_documents=True
        )

defask(self, question):
"""回答用户问题"""
        result = self.qa_chain.invoke({"query": question})

return {
"answer": result["result"],
"sources": [doc.metadata.get("source", "未知") 
for doc in result["source_documents"]]
        }

# 模拟企业知识库文档
from langchain.docstore.document import Document

documents = [
    Document(
        page_content="公司产品A支持自动数据分析和可视化，适用于中小型企业。",
        metadata={"source": "产品手册", "page": 1}
    ),
    Document(
        page_content="技术支持服务时间为工作日9:00-18:00，紧急问题可拨打400-xxx-xxxx。",
        metadata={"source": "服务协议", "page": 3}
    ),
    Document(
        page_content="数据导出支持CSV、Excel和JSON格式，单次最多导出10万条记录。",
        metadata={"source": "用户指南", "page": 5}
    )
]

# 创建客服实例
customer_service = SmartCustomerService(
    documents=documents,
    api_key="your-api-key"# 实际使用时替换
)

# 测试客服系统
questions = [
"公司有哪些产品？",
"技术支持时间是什么时候？",
"可以导出哪些格式的数据？"
]

for q in questions:
    print(f"\n用户问：{q}")
    response = customer_service.ask(q)
    print(f"客服答：{response['answer']}")
    print(f"来源：{response['sources']}")

# 完整实现：文档分析助手
import PyPDF2
import docx
from langchain.chains.summarize import load_summarize_chain
from langchain.text_splitter import CharacterTextSplitter

classDocumentAnalyzer:
"""文档分析助手"""

def__init__(self, llm):
        self.llm = llm

defread_pdf(self, file_path):
"""读取PDF文件"""
with open(file_path, 'rb') as file:
            pdf_reader = PyPDF2.PdfReader(file)
            text = ""
for page in pdf_reader.pages:
                text += page.extract_text()
return text

defread_docx(self, file_path):
"""读取Word文档"""
        doc = docx.Document(file_path)
return"\n".join([para.text for para in doc.paragraphs])

defsummarize(self, text, summary_type="brief"):
"""生成摘要"""
# 分割文本
        text_splitter = CharacterTextSplitter(
            separator="\n",
            chunk_size=1000,
            chunk_overlap=200
        )
        texts = text_splitter.split_text(text)
        docs = [Document(page_content=t) for t in texts]

# 选择摘要类型
if summary_type == "brief":
            chain_type = "map_reduce"
elif summary_type == "detailed":
            chain_type = "refine"
else:
            chain_type = "stuff"

# 加载摘要链
        chain = load_summarize_chain(
            self.llm,
            chain_type=chain_type,
            verbose=False
        )

return chain.invoke(docs)["output_text"]

defextract_key_points(self, text, num_points=5):
"""提取关键点"""
        prompt = f"""请从以下文本中提取{num_points}个关键点：

{text[:2000]}...

        以列表形式返回关键点，每个关键点用简短的一句话描述。"""

        response = self.llm.invoke(prompt)
return response.content

# 使用示例
analyzer = DocumentAnalyzer(llm)

# 模拟文档内容（实际使用时从文件读取）
sample_text = """Python是一种高级编程语言，由Guido van Rossum于1991年创建。
它具有简单易学的语法，强调代码的可读性。
Python支持多种编程范式，包括面向对象、命令式、函数式和过程式编程。
它拥有丰富的标准库和第三方库，广泛应用于Web开发、数据分析、人工智能等领域。
Python的解释器可以在多种操作系统上运行，包括Windows、Linux和macOS。"""

# 生成摘要
summary = analyzer.summarize(sample_text, "brief")
print(f"文档摘要：\n{summary}")

# 提取关键点
key_points = analyzer.extract_key_points(sample_text, 3)
print(f"\n关键点：\n{key_points}")

# 完整实现：代码生成与审查工具
import ast
import subprocess
from typing import List, Dict

classCodeAssistant:
"""代码生成与审查助手"""

def__init__(self, llm):
        self.llm = llm

defgenerate_code(self, requirement, language="python"):
"""根据需求生成代码"""
        prompt = f"""根据以下需求，生成{language}代码：

        需求：{requirement}

        要求：
        1. 代码完整可运行
        2. 包含必要的注释
        3. 考虑异常处理
        4. 遵循最佳实践

        只返回代码，不要有其他说明。"""

        response = self.llm.invoke(prompt)
return response.content

defreview_code(self, code):
"""代码审查"""
        prompt = f"""审查以下Python代码，提供：

        <!-- list-split -->

        1. 潜在Bug（3个）
        2. 性能优化建议（3条）
        3. 代码风格改进（3条）

        代码：
{code}

        以结构化方式返回结果。"""

        response = self.llm.invoke(prompt)
return response.content

defexplain_code(self, code):
"""代码解释"""
        prompt = f"""用通俗易懂的语言解释以下代码的功能和工作原理：

{code}

        解释内容包括：

        <!-- list-split -->

        1. 代码的整体功能
        2. 关键函数的作用
        3. 主要逻辑流程
        4. 可能的输入输出示例"""

        response = self.llm.invoke(prompt)
return response.content

defdebug_code(self, code, error_message):
"""调试代码"""
        prompt = f"""帮助调试以下代码：

        代码：
{code}

        错误信息：
{error_message}

        请：

        <!-- list-split -->

        1. 分析错误原因
        2. 提供修复方案
        3. 给出修复后的完整代码"""

        response = self.llm.invoke(prompt)
return response.content

# 使用示例
assistant = CodeAssistant(llm)

# 生成代码
requirement = "实现一个函数，计算斐波那契数列的第n项，要求使用缓存优化性能"
generated_code = assistant.generate_code(requirement)
print(f"生成的代码：\n{generated_code}")

# 审查代码
review = assistant.review_code(generated_code)
print(f"\n代码审查结果：\n{review}")

# 解释代码
explanation = assistant.explain_code(generated_code)
print(f"\n代码解释：\n{explanation[:500]}...")

llm = ChatOpenAI(model="gpt-4o", streaming=True, temperature=0.7)

memory = ConversationSummaryMemory(llm=llm, memory_key="chat_history")

import asyncio

asyncdefparallel_invoke(chains, inputs):
    tasks = [chain.ainvoke(input) for chain in chains]
returnawait asyncio.gather(*tasks)

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

set_llm_cache(InMemoryCache())

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    return_source_documents=True
)

from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

classCodeResponse(BaseModel):
    function_name: str = Field(description="函数名称")
    code: str = Field(description="完整代码")
    explanation: str = Field(description="代码说明")

parser = PydanticOutputParser(pydantic_object=CodeResponse)

deffact_check(answer, reference_sources):
"""使用参考源校验答案准确性"""
    prompt = f"""验证以下答案是否与参考源一致：

    答案：{answer}
    参考源：{reference_sources}

    返回：'一致'或'不一致'及具体差异"""

return llm.invoke(prompt).content

# 配置LangSmith（需要API密钥）
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# 在链中记录关键信息
chain = LLMChain(llm=llm, prompt=prompt, verbose=True)

from langchain import hub

# 拉取最新提示模板
prompt = hub.pull("hwchase17/react")

# 推送更新
# hub.push("your-username/your-prompt", updated_prompt)

from langchain.memory import ConversationTokenBufferMemory

memory = ConversationTokenBufferMemory(
    llm=llm,
    max_token_limit=4000,  # 根据模型调整
    return_messages=True
)

defhierarchical_compression(messages, llm):
"""分层压缩对话历史"""
# 第一层：提取关键实体
# 第二层：总结次要对话
# 第三层：保留最近详细对话
pass

defselective_memory_loading(query, full_history):
"""根据查询选择加载相关历史"""
# 使用嵌入相似度筛选相关对话
pass

# 无论底层模型如何，接口一致
llm = ChatOpenAI()  # OpenAI
llm = ChatAnthropic()  # Anthropic
llm = ChatGoogleGenerativeAI()  # Google

# config.yaml
# model:
#   provider: "openai"
#   name: "gpt-4o"
#   temperature: 0.7

import yaml

with open("config.yaml", "r") as f:
    config = yaml.safe_load(f)

defcreate_model(config):
    provider = config["model"]["provider"]
if provider == "openai":
return ChatOpenAI(**config["model"])
elif provider == "anthropic":
return ChatAnthropic(**config["model"])
# ... 其他供应商

deffallback_invoke(primary_chain, fallback_chain, input_data):
"""主模型失败时自动切换备用模型"""
try:
return primary_chain.invoke(input_data)
except Exception:
        print("主模型失败，切换备用模型")
return fallback_chain.invoke(input_data)