网站开发进度源码什么是速成网站

网站开发进度源码,什么是速成网站,网站建设详细教程,网片的重量计算公式LangChain 极速入门与技术实战#xff1a;Python AI应用新范式 LangChain 技术架构深度解析 LangChain 的核心价值在于其模块化设计哲学和清晰的架构分层。理解其技术架构是构建复杂AI应用的基础。 1. 技术架构概览 ┌───────────────────────────…LangChain 极速入门与技术实战Python AI应用新范式LangChain 技术架构深度解析LangChain 的核心价值在于其模块化设计哲学和清晰的架构分层。理解其技术架构是构建复杂AI应用的基础。1. 技术架构概览┌─────────────────────────────────────────────────────┐ │ 应用层 (Application) │ │ - 智能助手 │ 文档问答 │ 数据分析 │ 代码生成 │ ├─────────────────────────────────────────────────────┤ │ 链式层 (Chains) │ │ - LLMChain │ SequentialChain │ RouterChain │ ├─────────────────────────────────────────────────────┤ │ 组件抽象层 (Abstractions) │ │ - 提示模板 │ 记忆系统 │ 检索器 │ 输出解析器 │ ├─────────────────────────────────────────────────────┤ │ 模型层 (Models) │ │ - LLMs │ ChatModels │ Embeddings │ ├─────────────────────────────────────────────────────┤ │ 集成与工具层 (Integrations) │ │ - 向量数据库 │ API工具 │ 搜索引擎 │ 文件系统 │ └─────────────────────────────────────────────────────┘2. 设计模式与模块化LangChain 应用通常遵循以下设计模式# 完整应用架构示例config.py 配置管理模块 - 集中管理所有配置项 importosfromtypingimportDict,Anyfromdotenvimportload_dotenv load_dotenv()classConfig:配置管理器def__init__(self):self.openai_api_keyos.getenv(OPENAI_API_KEY)self.model_nameos.getenv(MODEL_NAME,gpt-4)self.temperaturefloat(os.getenv(TEMPERATURE,0.7))self.vector_db_pathos.getenv(VECTOR_DB_PATH,./chroma_db)defget_llm_config(self)-Dict[str,Any]:获取LLM配置return{model:self.model_name,temperature:self.temperature,api_key:self.openai_api_key}# 单例配置实例configConfig()核心组件详解与实战代码1. 模型抽象层多模型统一接口# models_manager.py 多模型管理器 - 支持动态切换和回退机制 fromlangchain_openaiimportChatOpenAI,OpenAIEmbeddingsfromlangchain_anthropicimportChatAnthropicfromlangchain_google_genaiimportChatGoogleGenerativeAIfromtypingimportOptional,Union,Listimportlogging loggerlogging.getLogger(__name__)classModelManager:统一模型管理器def__init__(self,config:Config):self.configconfig self._models{}self._embeddingsNonedefget_chat_model(self,provider:stropenai,**kwargs)-Union[ChatOpenAI,ChatAnthropic,ChatGoogleGenerativeAI]:获取聊天模型实例# 合并配置model_kwargs{temperature:self.config.temperature,**kwargs}ifprovideropenai:ifnotself.config.openai_api_key:raiseValueError(OpenAI API key is required)modelChatOpenAI(api_keyself.config.openai_api_key,model_nameself.config.model_name,**model_kwargs)elifprovideranthropic:modelChatAnthropic(**model_kwargs)elifprovidergoogle:modelChatGoogleGenerativeAI(**model_kwargs)else:raiseValueError(fUnsupported provider:{provider})self._models[provider]modelreturnmodeldefget_embeddings(self)-OpenAIEmbeddings:获取嵌入模型ifself._embeddingsisNone:self._embeddingsOpenAIEmbeddings(api_keyself.config.openai_api_key)returnself._embeddingsdefget_available_models(self)-List[str]:获取可用模型列表returnlist(self._models.keys())# 使用示例if__name____main__:fromconfigimportconfig managerModelManager(config)# 获取OpenAI模型openai_llmmanager.get_chat_model(provideropenai,temperature0.5,max_tokens1000)# 测试模型responseopenai_llm.invoke(你好请介绍一下自己)print(f模型响应:{response.content})2. 高级提示工程与模板系统# prompt_system.py 智能提示系统 - 支持动态模板和上下文感知 fromlangchain.promptsimport(ChatPromptTemplate,SystemMessagePromptTemplate,HumanMessagePromptTemplate,MessagesPlaceholder)fromlangchain.schemaimportBaseMessage,HumanMessage,AIMessagefromtypingimportDict,List,AnyimportjsonclassPromptSystem:高级提示管理系统def__init__(self):self.templates{}self._register_default_templates()def_register_default_templates(self):注册默认模板# 1. 数据分析专家模板data_analyst_templateChatPromptTemplate.from_messages([SystemMessagePromptTemplate.from_template(你是一位资深数据分析专家拥有10年行业经验。 你的任务是根据用户提供的数据和分析需求给出专业、准确的分析结果。 分析原则 1. 数据驱动所有结论必须有数据支持 2. 业务导向分析要服务于业务目标 3. 可操作性建议必须具体可执行 当前数据上下文 {data_context} 请开始你的分析。),MessagesPlaceholder(variable_namechat_history),HumanMessagePromptTemplate.from_template({query})])self.templates[data_analyst]data_analyst_template# 2. 代码生成模板code_gen_templateChatPromptTemplate.from_messages([SystemMessagePromptTemplate.from_template(你是一位经验丰富的软件工程师擅长{language}编程。 代码生成规范 1. 遵循PEP8/{language}最佳实践 2. 包含必要的注释和文档字符串 3. 考虑错误处理和边缘情况 4. 代码要模块化、可复用 需求{requirements} 请生成高质量的代码。),HumanMessagePromptTemplate.from_template({task})])self.templates[code_generator]code_gen_template# 3. 文档总结模板summary_templateChatPromptTemplate.from_messages([SystemMessagePromptTemplate.from_template(你是专业的文档总结助手能够从复杂文档中提取关键信息。 总结要求 1. 提取核心观点3-5个 2. 保留关键数据和事实 3. 识别行动建议 4. 字数控制在{word_limit}字以内 文档类型{doc_type} 目标读者{audience} 开始总结。),HumanMessagePromptTemplate.from_template({content})])self.templates[summarizer]summary_templatedefcreate_dynamic_prompt(self,template_name:str,**kwargs)-ChatPromptTemplate:创建动态提示模板iftemplate_namenotinself.templates:raiseValueError(f模板不存在:{template_name})templateself.templates[template_name]returntemplate.partial(**kwargs)ifkwargselsetemplatedefformat_with_context(self,template_name:str,query:str,context:Dict[str,Any],chat_history:List[BaseMessage]None)-List[BaseMessage]:带上下文的提示格式化promptself.create_dynamic_prompt(template_name)format_kwargs{query:query,**context}ifchat_history:format_kwargs[chat_history]chat_historyreturnprompt.format_messages(**format_kwargs)# 使用示例if__name____main__:prompt_systemPromptSystem()# 使用数据分析模板data_context{data_source:销售数据库,time_range:2024年Q1,metrics:[销售额,利润率,客户增长率]}messagesprompt_system.format_with_context(template_namedata_analyst,query分析季度销售趋势并提出改进建议,context{data_context:json.dumps(data_context,ensure_asciiFalse)},chat_history[HumanMessage(content这是我们的销售数据),AIMessage(content我已了解数据概况)])print(生成的提示消息:)formsginmessages:print(f{msg.type}:{msg.content[:100]}...)3. 高级记忆系统实现# memory_system.py 智能记忆系统 - 支持短期/长期记忆和记忆检索 fromlangchain.memoryimport(ConversationBufferMemory,ConversationSummaryMemory,VectorStoreRetrieverMemory)fromlangchain.schemaimportBaseMemoryfromtypingimportDict,Any,List,OptionalfromdatetimeimportdatetimeimportpickleimporthashlibclassAdvancedMemorySystem:高级记忆管理系统def__init__(self,short_term_memory:BaseMemoryNone,long_term_memory:BaseMemoryNone,max_short_term_items:int20):# 短期记忆最近对话self.short_term_memoryshort_term_memoryorConversationBufferMemory(return_messagesTrue,memory_keychat_history,output_keyoutput)# 长期记忆总结性记忆self.long_term_memorylong_term_memoryorConversationSummaryMemory(llmChatOpenAI(temperature0),memory_keylong_term_history)self.max_short_term_itemsmax_short_term_items self.memory_filememory_store.pklself._load_memory()def_load_memory(self):从文件加载记忆try:withopen(self.memory_file,rb)asf:saved_memorypickle.load(f)self.short_term_memorysaved_memory.get(short_term,self.short_term_memory)self.long_term_memorysaved_memory.get(long_term,self.long_term_memory)exceptFileNotFoundError:print(未找到记忆存储文件创建新记忆系统)defsave_memory(self):保存记忆到文件memory_data{short_term:self.short_term_memory,long_term:self.long_term_memory,saved_at:datetime.now().isoformat()}withopen(self.memory_file,wb)asf:pickle.dump(memory_data,f)defadd_interaction(self,user_input:str,ai_response:str,metadata:Dict[str,Any]None):添加交互到记忆系统# 添加到短期记忆self.short_term_memory.save_context({input:user_input},{output:ai_response})# 定期总结到长期记忆ifself._should_summarize():self._update_long_term_memory()# 添加元数据ifmetadata:self._add_metadata(metadata)def_should_summarize(self)-bool:判断是否需要总结chat_historyself.short_term_memory.load_memory_variables({})messageschat_history.get(chat_history,[])returnlen(messages)self.max_short_term_itemsdef_update_long_term_memory(self):更新长期记忆chat_historyself.short_term_memory.load_memory_variables({})summary_textself._create_summary(chat_history)self.long_term_memory.save_context({input:历史对话总结},{output:summary_text})# 清空短期记忆self.short_term_memory.clear()def_create_summary(self,chat_history:Dict)-str:创建对话总结messageschat_history.get(chat_history,[])conversation_text\n.join([f{msg.type}:{msg.content}formsginmessages[-10:]# 最近10条])# 使用LLM生成总结summary_promptf请总结以下对话的核心内容和关键信息{conversation_text}总结要求 1. 提取主要讨论主题 2. 记录重要决定和结论 3. 识别待办事项 4. 字数不超过200字returnsummary_prompt# 实际应用中应调用LLM生成总结def_add_metadata(self,metadata:Dict[str,Any]):添加元数据metadata_filemetadata.jsontry:importjsonwithopen(metadata_file,r)asf:existing_metadatajson.load(f)exceptFileNotFoundError:existing_metadata[]existing_metadata.append({**metadata,timestamp:datetime.now().isoformat()})withopen(metadata_file,w)asf:json.dump(existing_metadata,f,indent2)defget_memory_context(self)-Dict[str,Any]:获取记忆上下文short_termself.short_term_memory.load_memory_variables({})long_termself.long_term_memory.load_memory_variables({})return{short_term_memory:short_term,long_term_memory:long_term,memory_summary:self.get_memory_summary()}defget_memory_summary(self)-str:获取记忆摘要short_termself.short_term_memory.load_memory_variables({})messagesshort_term.get(chat_history,[])returnf对话记忆{len(messages)}条消息defsearch_memory(self,query:str)-List[Dict[str,Any]]:在记忆中搜索相关内容all_memoryself.get_memory_context()results[]# 简单关键词搜索实际应用可用向量搜索forkey,valueinall_memory.items():ifisinstance(value,str)andquery.lower()invalue.lower():results.append({source:key,content:value[:200]...iflen(value)200elsevalue})returnresults# 使用示例if__name____main__:memory_systemAdvancedMemorySystem()# 模拟对话conversations[(你好我叫张三,你好张三很高兴认识你。),(我来自北京,北京是个美丽的城市),(我喜欢编程,编程是个很有用的技能),(我想学习Python,Python是很好的入门语言。)]foruser_input,ai_responseinconversations:memory_system.add_interaction(user_inputuser_input,ai_responseai_response,metadata{topic:自我介绍})# 获取记忆上下文contextmemory_system.get_memory_context()print(记忆上下文:,context[memory_summary])# 搜索记忆resultsmemory_system.search_memory(编程)print(搜索结果:,results)# 保存记忆memory_system.save_memory()4. 智能代理系统架构# agent_system.py 智能代理系统 - 支持工具调用和任务分解 fromlangchain.agentsimport(AgentExecutor,create_openai_functions_agent,create_react_agent)fromlangchain.toolsimportBaseTool,toolfromlangchain.promptsimportChatPromptTemplate,MessagesPlaceholderfromtypingimportList,Dict,Any,Optionalimportasynciofromconcurrent.futuresimportThreadPoolExecutorimportjsonclassToolRegistry:工具注册器def__init__(self):self.tools{}self._register_builtin_tools()def_register_builtin_tools(self):注册内置工具tooldefcalculator(expression:str)-str:计算数学表达式支持加减乘除和常用函数try:# 安全地评估表达式allowed_names{abs:abs,round:round,min:min,max:max,sum:sum,len:len}# 使用ast安全解析importast nodeast.parse(expression,modeeval)def_eval(node):ifisinstance(node,ast.Num):returnnode.nelifisinstance(node,ast.BinOp):left_eval(node.left)right_eval(node.right)ifisinstance(node.op,ast.Add):returnleftrightelifisinstance(node.op,ast.Sub):returnleft-rightelifisinstance(node.op,ast.Mult):returnleft*rightelifisinstance(node.op,ast.Div):returnleft/rightelifisinstance(node,ast.Name):ifnode.idinallowed_names:returnallowed_names[node.id]elifisinstance(node,ast.Call):func_eval(node.func)args[_eval(arg)forarginnode.args]returnfunc(*args)raiseValueError(f不支持的表达式:{node})result_eval(node.body)returnstr(result)exceptExceptionase:returnf计算错误:{str(e)}self.register_tool(calculator,calculator)tooldefweb_search(query:str)-str:搜索网络信息# 实际应用中集成搜索引擎APIreturnf网络搜索结果模拟:{query}self.register_tool(web_search,web_search)tooldefdata_analyzer(data_json:str,analysis_type:str)-str:分析数据try:datajson.loads(data_json)ifanalysis_typestatistics:stats{count:len(data),keys:list(data[0].keys())ifdataelse[]}returnjson.dumps(stats,ensure_asciiFalse)else:return分析类型暂不支持exceptExceptionase:returnf数据分析错误:{str(e)}self.register_tool(data_analyzer,data_analyzer)defregister_tool(self,name:str,tool_func:callable):注册新工具self.tools[name]tool_funcdefget_tool(self,name:str)-Optional[BaseTool]:获取工具returnself.tools.get(name)defget_all_tools(self)-List[BaseTool]:获取所有工具returnlist(self.tools.values())classIntelligentAgent:智能代理def__init__(self,llm,tool_registry:ToolRegistry,agent_type:strreact):self.llmllm self.tool_registrytool_registry self.agent_typeagent_type# 创建代理ifagent_typereact:self.agentself._create_react_agent()elifagent_typeopenai_functions:self.agentself._create_openai_functions_agent()else:raiseValueError(f不支持的代理类型:{agent_type})# 创建执行器self.executorAgentExecutor(agentself.agent,toolstool_registry.get_all_tools(),verboseTrue,handle_parsing_errorsTrue,max_iterations5)def_create_react_agent(self):创建ReAct代理promptChatPromptTemplate.from_messages([(system,你是一个智能助手可以调用工具解决问题。 请遵循以下步骤 1. 理解问题 2. 思考需要哪些工具 3. 调用工具 4. 分析结果 5. 给出最终答案 可用的工具 {tools} 开始处理问题。),MessagesPlaceholder(variable_namechat_history),(user,{input}),MessagesPlaceholder(variable_nameagent_scratchpad)])returncreate_react_agent(llmself.llm,toolsself.tool_registry.get_all_tools(),promptprompt)def_create_openai_functions_agent(self):创建OpenAI函数代理promptChatPromptTemplate.from_messages([(system,你是一个有帮助的助手),MessagesPlaceholder(variable_namechat_history),(user,{input}),MessagesPlaceholder(variable_nameagent_scratchpad)])returncreate_openai_functions_agent(llmself.llm,promptprompt,toolsself.tool_registry.get_all_tools())asyncdefainvoke(self,input_text:str,**kwargs)-Dict[str,Any]:异步调用代理try:resultawaitself.executor.ainvoke({input:input_text,chat_history:kwargs.get(chat_history,[])})return{success:True,output:result.get(output,),intermediate_steps:result.get(intermediate_steps,[]),execution_time:result.get(execution_time,0)}exceptExceptionase:return{success:False,error:str(e),output:}definvoke(self,input_text:str,**kwargs)-Dict[str,Any]:同步调用代理try:resultself.executor.invoke({input:input_text,chat_history:kwargs.get(chat_history,[])})return{success:True,output:result.get(output,),intermediate_steps:result.get(intermediate_steps,[])}exceptExceptionase:return{success:False,error:str(e),output:}defbatch_process(self,inputs:List[str])-List[Dict[str,Any]]:批量处理withThreadPoolExecutor(max_workers5)asexecutor:futures[executor.submit(self.invoke,input_text)forinput_textininputs]return[future.result()forfutureinfutures]# 使用示例if__name____main__:fromconfigimportconfigfrommodels_managerimportModelManager# 初始化model_managerModelManager(config)llmmodel_manager.get_chat_model(openai)# 创建工具注册器和代理tool_registryToolRegistry()# 注册自定义工具tooldefweather_checker(city:str)-str:查询城市天气# 模拟天气数据weather_data{北京:晴15-25°C,上海:多云18-28°C,广州:雨22-30°C}returnweather_data.get(city,城市天气信息暂不可用)tool_registry.register_tool(weather_checker,weather_checker)# 创建智能代理agentIntelligentAgent(llmllm,tool_registrytool_registry,agent_typereact)# 测试代理test_queries[计算 (15 23) * 2 的结果,查询北京的天气,分析这个数据 [{name: Alice, age: 25}, {name: Bob, age: 30}]]forqueryintest_queries:print(f\n查询:{query})resultagent.invoke(query)ifresult[success]:print(f结果:{result[output]})ifresult.get(intermediate_steps):print(f中间步骤:{len(result[intermediate_steps])}步)else:print(f错误:{result[error]})5. 企业级RAG系统实现# rag_system.py 企业级RAG系统 - 检索增强生成 fromtypingimportList,Dict,Any,Optionalfromlangchain.document_loadersimport(TextLoader,PDFLoader,WebBaseLoader,DirectoryLoader)fromlangchain.text_splitterimport(RecursiveCharacterTextSplitter,CharacterTextSplitter)fromlangchain.embeddingsimportOpenAIEmbeddingsfromlangchain.vectorstoresimport(Chroma,FAISS,Pinecone)fromlangchain.retrieversimport(MultiVectorRetriever,ContextualCompressionRetriever,EnsembleRetriever)fromlangchain.retrievers.document_compressorsimport(LLMChainExtractor,EmbeddingsFilter)fromlangchain.chainsimportRetrievalQAfromlangchain.schemaimportDocumentimportosimportpickleclassEnterpriseRAGSystem:企业级RAG系统def__init__(self,embedding_modelNone,llmNone,vector_store_type:strchroma,persist_directory:str./vector_store):self.embedding_modelembedding_modelorOpenAIEmbeddings()self.llmllm self.vector_store_typevector_store_type self.persist_directorypersist_directory self.vector_storeNoneself.retrieverNoneself.qa_chainNonedefload_documents(self,source_path:str,source_type:strdirectory,**kwargs)-List[Document]:加载文档loader_map{directory:DirectoryLoader,pdf:PDFLoader,text:TextLoader,web:WebBaseLoader}ifsource_typenotinloader_map:raiseValueError(f不支持的文档类型:{source_type})loader_classloader_map[source_type]ifsource_typedirectory:loaderloader_class(source_path,glob**/*.*,loader_kwargskwargs.get(loader_kwargs,{}))else:loaderloader_class(source_path,**kwargs)returnloader.load()defprocess_documents(self,documents:List[Document],chunk_size:int1000,chunk_overlap:int200)-List[Document]:处理文档# 文本分割text_splitterRecursiveCharacterTextSplitter(chunk_sizechunk_size,chunk_overlapchunk_overlap,length_functionlen,separators[\n\n,\n,。,,,,, ])chunkstext_splitter.split_documents(documents)# 添加元数据fori,chunkinenumerate(chunks):chunk.metadata.update({chunk_id:i,total_chunks:len(chunks),source_hash:hash(chunk.page_content[:100])})returnchunksdefcreate_vector_store(self,documents:List[Document]):创建向量存储ifself.vector_store_typechroma:self.vector_storeChroma.from_documents(documentsdocuments,embeddingself.embedding_model,persist_directoryself.persist_directory)self.vector_store.persist()elifself.vector_store_typefaiss:self.vector_storeFAISS.from_documents(documentsdocuments,embeddingself.embedding_model)# 保存FAISS索引self.vector_store.save_local(self.persist_directory)else:raiseValueError(f不支持的向量存储类型:{self.vector_store_type})print(f向量存储创建成功包含{len(documents)}个文档块)defcreate_advanced_retriever(self,search_type:strsimilarity,k:int4,score_threshold:float0.5):创建高级检索器ifnotself.vector_store:raiseValueError(请先创建向量存储)# 基础检索器base_retrieverself.vector_store.as_retriever(search_typesearch_type,search_kwargs{k:k*2,# 获取更多结果用于压缩score_threshold:score_threshold})# 上下文压缩检索器compressorLLMChainExtractor.from_llm(self.llm)compression_retrieverContextualCompressionRetriever(base_compressorcompressor,base_retrieverbase_retriever)# 嵌入过滤embeddings_filterEmbeddingsFilter(embeddingsself.embedding_model,similarity_thresholdscore_threshold)# 组合检索器self.retrievercompression_retrieverreturnself.retrieverdefcreate_qa_chain(self,chain_type:strstuff):创建问答链ifnotself.retriever:raiseValueError(请先创建检索器)self.qa_chainRetrievalQA.from_chain_type(llmself.llm,chain_typechain_type,retrieverself.retriever,return_source_documentsTrue,verboseTrue)returnself.qa_chaindefquery(self,question:str,**kwargs)-Dict[str,Any]:查询ifnotself.qa_chain:raiseValueError(请先创建问答链)try:resultself.qa_chain({query:question})return{success:True,answer:result[result],source_documents:[{content:doc.page_content[:200]...,metadata:doc.metadata}fordocinresult.get(source_documents,[])],source_count:len(result.get(source_documents,[]))}exceptExceptionase:return{success:False,error:str(e),answer:}defsave_system(self,filepath:strrag_system.pkl):保存系统状态state{vector_store_type:self.vector_store_type,persist_directory:self.persist_directory,config:{embedding_model:type(self.embedding_model).__name__,llm:type(self.llm).__name__ifself.llmelseNone}}withopen(filepath,wb)asf:pickle.dump(state,f)print(f系统状态已保存到{filepath})classmethoddefload_system(cls,filepath:strrag_system.pkl,embedding_modelNone,llmNone):加载系统withopen(filepath,rb)asf:statepickle.load(f)systemcls(embedding_modelembedding_model,llmllm,vector_store_typestate[vector_store_type],persist_directorystate[persist_directory])# 加载向量存储ifstate[vector_store_type]chroma:system.vector_storeChroma(persist_directorystate[persist_directory],embedding_functionembedding_model)system.retrieversystem.vector_store.as_retriever()elifstate[vector_store_type]faiss:system.vector_storeFAISS.load_local(state[persist_directory],embedding_model,allow_dangerous_deserializationTrue)system.retrieversystem.vector_store.as_retriever()# 重新创建QA链ifllm:system.create_qa_chain()returnsystem# 使用示例if__name____main__:fromconfigimportconfigfrommodels_managerimportModelManager# 初始化模型model_managerModelManager(config)llmmodel_manager.get_chat_model(openai,temperature0.1)embeddingsmodel_manager.get_embeddings()# 创建RAG系统rag_systemEnterpriseRAGSystem(embedding_modelembeddings,llmllm,vector_store_typechroma,persist_directory./data/vector_store)# 1. 加载文档print(加载文档...)documentsrag_system.load_documents(source_path./data/documents,source_typedirectory)print(f加载了{len(documents)}个文档)# 2. 处理文档print(处理文档...)processed_docsrag_system.process_documents(documents,chunk_size800,chunk_overlap100)print(f分割为{len(processed_docs)}个文档块)# 3. 创建向量存储print(创建向量存储...)rag_system.create_vector_store(processed_docs)# 4. 创建检索器print(创建检索器...)retrieverrag_system.create_advanced_retriever(search_typesimilarity_score_threshold,k3,score_threshold0.7)# 5. 创建QA链print(创建QA链...)qa_chainrag_system.create_qa_chain(chain_typemap_reduce)# 6. 测试查询test_questions[文档的主要内容是什么,有哪些关键的技术点,作者提出了什么建议]forquestionintest_questions:print(f\n问题:{question})resultrag_system.query(question)ifresult[success]:print(f回答:{result[answer][:200]}...)print(f参考文档:{result[source_count]}个)else:print(f错误:{result[error]})# 7. 保存系统rag_system.save_system(./data/rag_system_state.pkl)高级应用完整AI应用架构# ai_application.py 完整的AI应用架构示例 importasynciofromtypingimportDict,Any,List,Optionalfromdatetimeimportdatetimeimportloggingfromdataclassesimportdataclass,asdictimportjson# 配置日志logging.basicConfig(levellogging.INFO,format%(asctime)s - %(name)s - %(levelname)s - %(message)s)loggerlogging.getLogger(__name__)dataclassclassAIConfig:AI应用配置model_provider:stropenaimodel_name:strgpt-4temperature:float0.7max_tokens:int2000enable_memory:boolTrueenable_tools:boolTruecache_responses:boolTruedefto_dict(self)-Dict[str,Any]:returnasdict(self)classAIApplication:完整的AI应用def__init__(self,config:AIConfig):self.configconfig self.components{}self._initialize_components()def_initialize_components(self):初始化所有组件logger.info(初始化AI应用组件...)# 1. 初始化模型管理器frommodels_managerimportModelManagerfromconfigimportConfigasAppConfig app_configAppConfig()self.components[model_manager]ModelManager(app_config)# 2. 初始化LLMself.components[llm]self.components[model_manager].get_chat_model(providerself.config.model_provider,temperatureself.config.temperature,max_tokensself.config.max_tokens)# 3. 初始化记忆系统ifself.config.enable_memory:frommemory_systemimportAdvancedMemorySystem self.components[memory]AdvancedMemorySystem()# 4. 初始化代理系统ifself.config.enable_tools:fromagent_systemimportToolRegistry,IntelligentAgent tool_registryToolRegistry()self.components[agent]IntelligentAgent(llmself.components[llm],tool_registrytool_registry,agent_typereact)# 5. 初始化缓存ifself.config.cache_responses:self.components[cache]ResponseCache()logger.info(AI应用初始化完成)asyncdefprocess_request(self,user_input:str,user_id:Optional[str]None,context:Optional[Dict[str,Any]]None)-Dict[str,Any]:处理用户请求start_timedatetime.now()request_idfreq_{datetime.now().strftime(%Y%m%d_%H%M%S)}logger.info(f处理请求{request_id}:{user_input[:50]}...)try:# 检查缓存ifself.config.cache_responses:cached_responseself.components[cache].get(user_input)ifcached_response:logger.info(f请求{request_id}命中缓存)return{request_id:request_id,response:cached_response,from_cache:True,processing_time:0}# 获取记忆上下文memory_context{}ifself.config.enable_memoryanduser_id:memory_contextself.components[memory].get_memory_context()# 使用代理处理ifself.config.enable_tools:resultawaitself.components[agent].ainvoke(user_input,chat_historymemory_context.get(chat_history,[]))responseresult[output]tool_usedlen(result.get(intermediate_steps,[]))0else:# 直接使用LLMresponse_objawaitself.components[llm].ainvoke(user_input)responseresponse_obj.content tool_usedFalse# 更新记忆ifself.config.enable_memoryanduser_id:self.components[memory].add_interaction(user_inputuser_input,ai_responseresponse,metadata{user_id:user_id,request_id:request_id,used_tools:tool_used})# 保存到缓存ifself.config.cache_responses:self.components[cache].set(user_input,response)processing_time(datetime.now()-start_time).total_seconds()return{request_id:request_id,response:response,from_cache:False,tool_used:tool_used,processing_time:processing_time,timestamp:datetime.now().isoformat()}exceptExceptionase:logger.error(f处理请求{request_id}时出错:{str(e)})return{request_id:request_id,error:str(e),response:抱歉处理您的请求时出现了错误。,success:False,timestamp:datetime.now().isoformat()}defbatch_process(self,requests:List[Dict[str,Any]])-List[Dict[str,Any]]:批量处理请求asyncdefprocess_all():tasks[]forreqinrequests:taskself.process_request(user_inputreq.get(input,),user_idreq.get(user_id),contextreq.get(context,{}))tasks.append(task)returnawaitasyncio.gather(*tasks)returnasyncio.run(process_all())defget_system_status(self)-Dict[str,Any]:获取系统状态status{config:self.config.to_dict(),components:{name:type(comp).__name__forname,compinself.components.items()},memory_stats:{}}ifself.config.enable_memory:memoryself.components.get(memory)ifmemory:status[memory_stats]memory.get_memory_summary()returnstatusclassResponseCache:响应缓存def__init__(self,max_size:int1000,ttl:int3600):self.cache{}self.max_sizemax_size self.ttlttl# 生存时间秒self.access_times{}defget(self,key:str)-Optional[str]:获取缓存ifkeyinself.cache:# 检查是否过期last_accessself.access_times.get(key)iflast_accessand(datetime.now()-last_access).secondsself.ttl:self.delete(key)returnNoneself.access_times[key]datetime.now()returnself.cache[key]returnNonedefset(self,key:str,value:str):设置缓存iflen(self.cache)self.max_size:# 删除最久未使用的oldest_keymin(self.access_times,keyself.access_times.get)self.delete(oldest_key)self.cache[key]value self.access_times[key]datetime.now()defdelete(self,key:str):删除缓存ifkeyinself.cache:delself.cache[key]ifkeyinself.access_times:delself.access_times[key]defclear(self):清空缓存self.cache.clear()self.access_times.clear()defstats(self)-Dict[str,Any]:缓存统计return{size:len(self.cache),max_size:self.max_size,ttl:self.ttl}# 使用示例asyncdefmain():主函数示例# 创建配置configAIConfig(model_provideropenai,model_namegpt-4,temperature0.7,enable_memoryTrue,enable_toolsTrue,cache_responsesTrue)# 创建AI应用appAIApplication(config)# 获取系统状态statusapp.get_system_status()print(系统状态:)print(json.dumps(status,indent2,ensure_asciiFalse))# 处理单个请求print(\n处理单个请求...)resultawaitapp.process_request(user_input计算一下 (45 78) * 2 除以 3 的结果,user_iduser_123)print(f响应:{result.get(response,)})print(f是否使用工具:{result.get(tool_used,False)})print(f处理时间:{result.get(processing_time,0):.2f}秒)# 批量处理print(\n批量处理请求...)batch_requests[{input:你好介绍一下Python语言,user_id:user_123},{input:今天北京天气怎么样,user_id:user_456},{input:解释一下机器学习,user_id:user_789}]batch_resultsapp.batch_process(batch_requests)fori,resultinenumerate(batch_results):print(f\n请求{i1}:)print(f输入:{batch_requests[i][input]})print(f响应:{result.get(response,)[:100]}...)print(f是否缓存:{result.get(from_cache,False)})if__name____main__:asyncio.run(main())部署与监控# deployment.py 生产环境部署与监控 fromfastapiimportFastAPI,HTTPException,DependsfrompydanticimportBaseModel,FieldfromtypingimportList,Optionalimportuvicornfromdatetimeimportdatetimeimportloggingimportasynciofromcontextlibimportasynccontextmanagerimportprometheus_clientfromprometheus_fastapi_instrumentatorimportInstrumentator# 请求/响应模型classChatRequest(BaseModel):message:strField(...,min_length1,max_length1000)user_id:Optional[str]Nonesession_id:Optional[str]Nonecontext:Optional[dict]{}classChatResponse(BaseModel):message_id:strresponse:strprocessing_time:floatmodel_used:strtimestamp:str# 应用生命周期管理asynccontextmanagerasyncdeflifespan(app:FastAPI):应用生命周期管理# 启动时初始化print(启动AI应用...)# 初始化AI应用fromai_applicationimportAIApplication,AIConfig configAIConfig(model_provideropenai,model_namegpt-4,temperature0.7)app.state.ai_appAIApplication(config)app.state.request_counterprometheus_client.Counter(chat_requests_total,Total chat requests,[status])yield# 关闭时清理print(关闭AI应用...)# 清理资源# 创建FastAPI应用appFastAPI(titleAI Chat API,description基于LangChain的智能对话API,version1.0.0,lifespanlifespan)# 添加监控Instrumentator().instrument(app).expose(app)# 中间件app.middleware(http)asyncdefadd_process_time_header(request,call_next):start_timedatetime.now()responseawaitcall_next(request)processing_time(datetime.now()-start_time).total_seconds()response.headers[X-Processing-Time]str(processing_time)returnresponse# 路由app.post(/chat,response_modelChatResponse)asyncdefchat(request:ChatRequest):聊天接口try:# 记录请求app.state.request_counter.labels(statusreceived).inc()# 处理请求resultawaitapp.state.ai_app.process_request(user_inputrequest.message,user_idrequest.user_id,contextrequest.contextor{})ifresult.get(success,True):app.state.request_counter.labels(statussuccess).inc()returnChatResponse(message_idresult[request_id],responseresult[response],processing_timeresult.get(processing_time,0),model_usedapp.state.ai_app.config.model_name,timestampresult.get(timestamp,datetime.now().isoformat()))else:app.state.request_counter.labels(statuserror).inc()raiseHTTPException(status_code500,detailresult.get(error,处理失败))exceptExceptionase:app.state.request_counter.labels(statuserror).inc()raiseHTTPException(status_code500,detailstr(e))app.get(/health)asyncdefhealth_check():健康检查return{status:healthy,timestamp:datetime.now().isoformat(),service:AI Chat API}app.get(/metrics)asyncdefmetrics():Prometheus指标returnprometheus_client.generate_latest()app.get(/system/status)asyncdefsystem_status():系统状态ifhasattr(app.state,ai_app):returnapp.state.ai_app.get_system_status()return{status:initializing}# 批量处理app.post(/batch_chat)asyncdefbatch_chat(requests:List[ChatRequest]):批量聊天接口try:request_data[{input:req.message,user_id:req.user_id,context:req.contextor{}}forreqinrequests]resultsapp.state.ai_app.batch_process(request_data)return[{message_id:result.get(request_id,fbatch_{i}),response:result.get(response,),success:result.get(success,True)}fori,resultinenumerate(results)]exceptExceptionase:raiseHTTPException(status_code500,detailstr(e))if__name____main__:uvicorn.run(deployment:app,host0.0.0.0,port8000,reloadTrue,log_levelinfo)项目结构与最佳实践ai-application/ ├── README.md ├── requirements.txt ├── .env.example ├── config/ │ ├── __init__.py │ └── settings.py ├── src/ │ ├── core/ │ │ ├── __init__.py │ │ ├── models.py │ │ ├── prompts.py │ │ ├── memory.py │ │ └── agents.py │ ├── services/ │ │ ├── __init__.py │ │ ├── chat_service.py │ │ └── rag_service.py │ └── api/ │ ├── __init__.py │ ├── routes.py │ └── middleware.py ├── tests/ │ ├── __init__.py │ ├── test_models.py │ └── test_services.py ├── scripts/ │ ├── setup.py │ └── deploy.sh └── docker/ ├── Dockerfile └── docker-compose.yml最佳实践总结模块化设计每个组件职责单一便于测试和维护配置管理使用环境变量和配置文件避免硬编码错误处理完善的异常捕获和错误恢复机制性能监控集成监控和日志系统便于问题排查缓存策略合理使用缓存提高响应速度安全考虑输入验证、API密钥管理、沙箱执行版本控制API版本管理向后兼容总结LangChain 提供了一个强大而灵活的框架使得构建复杂的AI应用变得简单高效。通过本文提供的完整示例和最佳实践你可以快速搭建起符合企业级标准的AI应用系统。关键点包括架构清晰分层设计组件解耦代码完整提供可直接运行的完整示例生产就绪包含部署、监控、缓存等生产环境特性扩展性强易于添加新功能和集成新工具随着LangChain生态的不断发展开发者可以更专注于业务逻辑的实现而无需担心底层复杂性真正实现AI应用的快速开发和部署。