langchain/
├── chains/                # 链组件，用于定义和管理处理流程
├── memory/                # 记忆组件，用于管理对话和交互的上下文
├── prompts/               # 提示模板组件，用于生成向LLM提交的提示
├── agents/                # 代理组件，用于实现智能决策和工具调用
├── tools/                 # 工具组件，封装各种外部功能
├── llms/                  # LLM接口层，支持多种LLM提供商
├── embeddings/            # 嵌入模型接口，用于文本向量化
├── vectorstores/          # 向量数据库接口，用于语义搜索
├── document_loaders/      # 文档加载器，用于处理各种格式的文档
├── text_splitter/         # 文本分割器，用于将长文本分割为小块
├── callbacks/             # 回调系统，用于监控和记录执行过程
└── utils/                 # 工具函数和辅助类

# langchain/llms/base.pyfrom abc import ABC, abstractmethod
from typing import Any, List, Mapping, Optional, Unionclass LLM(ABC):"""LLM抽象基类，定义了所有LLM实现必须遵循的接口"""@property@abstractmethoddef _llm_type(self) -> str:"""返回LLM类型的字符串标识"""pass@abstractmethoddef _call(self, prompt: str, stop: Optional[List[str]] = None) -> str:"""执行LLM调用的核心方法"""pass@propertydef _identifying_params(self) -> Mapping[str, Any]:"""返回标识LLM实例的参数"""return {}def __str__(self) -> str:"""返回LLM的字符串表示"""return f"{self._llm_type()}: {self._identifying_params}"

# langchain/llms/openai.pyimport openai
from typing import Any, Dict, List, Optional, Unionfrom langchain.llms.base import LLM
from langchain.utils import get_from_dict_or_envclass OpenAI(LLM):"""OpenAI LLM实现"""# 模型名称，默认为text-davinci-003model_name: str = "text-davinci-003"# 温度参数，控制输出的随机性temperature: float = 0.7# 最大生成token数max_tokens: int = 2049# 顶部概率采样参数top_p: float = 1# 频率惩罚参数frequency_penalty: float = 0# 存在惩罚参数presence_penalty: float = 0# API密钥openai_api_key: Optional[str] = None@propertydef _llm_type(self) -> str:"""返回LLM类型标识"""return "openai"def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str:"""执行OpenAI API调用"""# 获取API密钥，优先从实例属性获取，否则从环境变量获取openai_api_key = get_from_dict_or_env(self.__dict__, "openai_api_key", "OPENAI_API_KEY")# 构建API调用参数params = {"model": self.model_name,"prompt": prompt,"temperature": self.temperature,"max_tokens": self.max_tokens,"top_p": self.top_p,"frequency_penalty": self.frequency_penalty,"presence_penalty": self.presence_penalty,}# 如果提供了停止词，添加到参数中if stop is not None:params["stop"] = stop# 执行API调用response = openai.Completion.create(api_key=openai_api_key,**params)# 提取并返回生成的文本return response.choices[0].text.strip()@propertydef _identifying_params(self) -> Dict[str, Any]:"""返回标识LLM实例的参数"""return {"model_name": self.model_name,"temperature": self.temperature,"max_tokens": self.max_tokens,"top_p": self.top_p,"frequency_penalty": self.frequency_penalty,"presence_penalty": self.presence_penalty,}

# langchain/llms/base.pyimport asyncio
from typing import AsyncGeneratorclass LLM(ABC):# ... 其他方法 ...async def agenerate(self, prompts: List[str], stop: Optional[List[str]] = None) -> LLMResult:"""异步生成多个提示的响应"""tasks = [self._agenerate_one(prompt, stop=stop) for prompt in prompts]results = await asyncio.gather(*tasks)return LLMResult(generations=[[res] for res in results])async def _agenerate_one(self, prompt: str, stop: Optional[List[str]] = None) -> Generation:"""异步生成单个提示的响应"""text = await self._acall(prompt, stop=stop)return Generation(text=text)@abstractmethodasync def _acall(self, prompt: str, stop: Optional[List[str]] = None) -> str:"""异步调用LLM的核心方法，由子类实现"""pass

# langchain/llms/openai.pyclass OpenAI(LLM):# ... 其他方法 ...async def _acall(self, prompt: str, stop: Optional[List[str]] = None) -> str:"""异步执行OpenAI API调用"""openai_api_key = get_from_dict_or_env(self.__dict__, "openai_api_key", "OPENAI_API_KEY")params = {"model": self.model_name,"prompt": prompt,"temperature": self.temperature,"max_tokens": self.max_tokens,"top_p": self.top_p,"frequency_penalty": self.frequency_penalty,"presence_penalty": self.presence_penalty,}if stop is not None:params["stop"] = stop# 异步执行API调用response = await openai.Completion.acreate(api_key=openai_api_key,**params)return response.choices[0].text.strip()

# langchain/chains/base.pyfrom abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Tuple, Unionfrom langchain.callbacks.base import BaseCallbackManager
from langchain.callbacks.manager import (AsyncCallbackManagerForChainRun,CallbackManagerForChainRun,Callbacks,
)class Chain(ABC):"""Chain抽象基类，定义了所有Chain实现必须遵循的接口"""# 回调管理器，用于记录和监控Chain的执行过程callback_manager: Optional[BaseCallbackManager] = None@property@abstractmethoddef input_keys(self) -> List[str]:"""返回Chain期望的输入键列表"""pass@property@abstractmethoddef output_keys(self) -> List[str]:"""返回Chain产生的输出键列表"""pass@abstractmethoddef _call(self, inputs: Dict[str, Any],run_manager: Optional[CallbackManagerForChainRun] = None) -> Dict[str, Any]:"""执行Chain的核心方法，由子类实现"""passasync def _acall(self, inputs: Dict[str, Any],run_manager: Optional[AsyncCallbackManagerForChainRun] = None) -> Dict[str, Any]:"""异步执行Chain的方法，默认同步实现，可被子类重写"""return self._call(inputs, run_manager=run_manager)def run(self, callbacks: Callbacks = None,**kwargs: Any) -> Union[str, Dict[str, Any]]:"""便捷方法，用于直接运行Chain并获取结果"""# 检查输入键是否匹配if set(kwargs.keys()) != set(self.input_keys):raise ValueError(f"输入键不匹配。期望: {self.input_keys}, 实际: {list(kwargs.keys())}")# 执行Chainoutput = self(kwargs, callbacks=callbacks)# 如果只有一个输出键，直接返回该值if len(self.output_keys) == 1:return output[self.output_keys[0]]else:return outputasync def arun(self, callbacks: Callbacks = None,**kwargs: Any) -> Union[str, Dict[str, Any]]:"""异步便捷方法，用于直接运行Chain并获取结果"""if set(kwargs.keys()) != set(self.input_keys):raise ValueError(f"输入键不匹配。期望: {self.input_keys}, 实际: {list(kwargs.keys())}")output = await self.acall(kwargs, callbacks=callbacks)if len(self.output_keys) == 1:return output[self.output_keys[0]]else:return output# 其他方法...

# langchain/chains/sequential.pyfrom typing import Any, Dict, List, Optionalfrom langchain.chains.base import Chainclass SimpleSequentialChain(Chain):"""简单顺序链，按顺序执行多个Chain，前一个Chain的输出作为后一个Chain的输入"""# 子Chain列表chains: List[Chain]# 是否只返回最后一个Chain的输出return_only_final_output: bool = False@propertydef input_keys(self) -> List[str]:"""返回第一个Chain的输入键"""return self.chains[0].input_keys@propertydef output_keys(self) -> List[str]:"""返回最后一个Chain的输出键"""return self.chains[-1].output_keysdef _call(self, inputs: Dict[str, Any],run_manager: Optional[CallbackManagerForChainRun] = None) -> Dict[str, Any]:"""执行顺序链"""current_inputs = inputsintermediate_outputs = {}# 依次执行每个子Chainfor i, chain in enumerate(self.chains):# 为当前Chain创建回调管理器if run_manager:new_run_manager = run_manager.get_child()else:new_run_manager = None# 执行当前Chainoutputs = chain(current_inputs, callbacks=new_run_manager)# 记录中间输出if not self.return_only_final_output:for k, v in outputs.items():intermediate_outputs[f"{chain.__class__.__name__}_{k}"] = v# 更新当前输入为当前Chain的输出current_inputs = outputs# 返回最终输出或所有中间输出if self.return_only_final_output:return current_inputselse:return intermediate_outputsasync def _acall(self, inputs: Dict[str, Any],run_manager: Optional[AsyncCallbackManagerForChainRun] = None) -> Dict[str, Any]:"""异步执行顺序链"""current_inputs = inputsintermediate_outputs = {}for i, chain in enumerate(self.chains):if run_manager:new_run_manager = run_manager.get_child()else:new_run_manager = Noneoutputs = await chain.acall(current_inputs, callbacks=new_run_manager)if not self.return_only_final_output:for k, v in outputs.items():intermediate_outputs[f"{chain.__class__.__name__}_{k}"] = vcurrent_inputs = outputsif self.return_only_final_output:return current_inputselse:return intermediate_outputs

# langchain/chains/llm.pyfrom typing import Any, Dict, List, Optional, Unionfrom langchain.base_language import BaseLanguageModel
from langchain.callbacks.manager import (AsyncCallbackManagerForChainRun,CallbackManagerForChainRun,
)
from langchain.chains.base import Chain
from langchain.prompts.base import BasePromptTemplateclass LLMChain(Chain):"""LLMChain将提示模板和LLM结合，用于生成文本"""# 提示模板prompt: BasePromptTemplate# 语言模型llm: BaseLanguageModel# 输出键名称output_key: str = "text"@propertydef input_keys(self) -> List[str]:"""返回提示模板所需的输入键"""return self.prompt.input_variables@propertydef output_keys(self) -> List[str]:"""返回输出键"""return [self.output_key]def _call(self, inputs: Dict[str, Any],run_manager: Optional[CallbackManagerForChainRun] = None) -> Dict[str, str]:"""执行LLMChain，生成文本"""# 格式化提示prompt_value = self.prompt.format_prompt(**inputs)# 记录提示信息if run_manager:run_manager.on_text("Prompt after formatting:\n", end="\n", verbose=self.verbose)run_manager.on_text(prompt_value.to_string(), color="green", end="\n", verbose=self.verbose)# 调用LLM生成文本response = self.llm.generate_prompt([prompt_value],callbacks=run_manager.get_child() if run_manager else None)# 提取生成的文本text = response.generations[0][0].text# 记录生成的文本if run_manager:run_manager.on_text("Generated response:", end="\n", verbose=self.verbose)run_manager.on_text(text, color="yellow", end="\n", verbose=self.verbose)return {self.output_key: text}async def _acall(self, inputs: Dict[str, Any],run_manager: Optional[AsyncCallbackManagerForChainRun] = None) -> Dict[str, str]:"""异步执行LLMChain，生成文本"""prompt_value = self.prompt.format_prompt(**inputs)if run_manager:await run_manager.on_text("Prompt after formatting:\n", end="\n", verbose=self.verbose)await run_manager.on_text(prompt_value.to_string(), color="green", end="\n", verbose=self.verbose)response = await self.llm.agenerate_prompt([prompt_value],callbacks=run_manager.get_child() if run_manager else None)text = response.generations[0][0].textif run_manager:await run_manager.on_text("Generated response:", end="\n", verbose=self.verbose)await run_manager.on_text(text, color="yellow", end="\n", verbose=self.verbose)return {self.output_key: text}

# langchain/memory/base.pyfrom abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Tuplefrom langchain.schema import BaseMessageclass BaseMemory(ABC):"""Memory抽象基类，定义了所有Memory实现必须遵循的接口"""@property@abstractmethoddef memory_variables(self) -> List[str]:"""返回Memory中存储的变量名称列表"""pass@abstractmethoddef load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:"""根据输入加载Memory中的变量"""pass@abstractmethoddef save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> None:"""保存上下文信息到Memory中"""passdef clear(self) -> None:"""清除Memory中的所有信息，默认实现为空"""pass

# langchain/memory/buffer.pyfrom typing import Any, Dict, List, Optional, Tuplefrom langchain.memory.base import BaseMemory
from langchain.schema import BaseMessage, HumanMessage, AIMessageclass ConversationBufferMemory(BaseMemory):"""对话缓冲区Memory，存储整个对话历史"""# 对话历史chat_memory: BaseChatMemory# 是否返回聊天历史作为字符串return_messages: bool = False# 输入键名称input_key: Optional[str] = None# 输出键名称output_key: Optional[str] = None@propertydef memory_variables(self) -> List[str]:"""返回Memory中存储的变量名称"""return ["history"]def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:"""加载对话历史"""messages = self.chat_memory.messagesif self.return_messages:# 直接返回消息列表history = messageselse:# 将消息转换为字符串history = get_buffer_string(messages)return {"history": history}def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> None:"""保存上下文信息到对话历史"""# 获取用户输入input_str = inputs[self.input_key] if self.input_key else inputs[list(inputs.keys())[0]]# 添加用户消息到对话历史self.chat_memory.add_user_message(input_str)# 获取模型输出output_str = outputs[self.output_key] if self.output_key else outputs[list(outputs.keys())[0]]# 添加模型回复到对话历史self.chat_memory.add_ai_message(output_str)def clear(self) -> None:"""清除对话历史"""self.chat_memory.clear()

# langchain/memory/summary.pyfrom typing import Any, Dict, List, Optional, Tuplefrom langchain.base_language import BaseLanguageModel
from langchain.chains import LLMChain
from langchain.memory.base import BaseMemory
from langchain.prompts import PromptTemplate
from langchain.schema import BaseMessage, HumanMessage, AIMessageSUMMARY_PROMPT = """总结以下对话:{history}总结:"""class ConversationSummaryMemory(BaseMemory):"""对话摘要Memory，维护对话的摘要"""# 语言模型，用于生成摘要llm: BaseLanguageModel# 对话历史chat_memory: BaseChatMemory# 摘要summary: str = ""# 摘要提示模板prompt: PromptTemplate = PromptTemplate(input_variables=["history"], template=SUMMARY_PROMPT)# 输入键名称input_key: Optional[str] = None# 输出键名称output_key: Optional[str] = None@propertydef memory_variables(self) -> List[str]:"""返回Memory中存储的变量名称"""return ["history"]def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:"""加载对话摘要"""return {"history": self.summary}def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> None:"""保存上下文信息并更新摘要"""# 保存用户输入和模型输出到对话历史input_str = inputs[self.input_key] if self.input_key else inputs[list(inputs.keys())[0]]self.chat_memory.add_user_message(input_str)output_str = outputs[self.output_key] if self.output_key else outputs[list(outputs.keys())[0]]self.chat_memory.add_ai_message(output_str)# 获取最新的对话历史messages = self.chat_memory.messagesmessage_text = get_buffer_string(messages)# 创建LLMChain用于生成摘要chain = LLMChain(llm=self.llm, prompt=self.prompt)# 生成新的摘要self.summary = chain.run(history=message_text)def clear(self) -> None:"""清除对话历史和摘要"""self.chat_memory.clear()self.summary = ""

from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory# 创建提示模板
prompt = PromptTemplate(input_variables=["input", "history"],template="""你是一个智能助手，帮助用户解决问题。以下是之前的对话历史:{history}用户提问: {input}助手回答:"""
)# 创建Memory
memory = ConversationBufferMemory(input_key="input", output_key="text")# 创建LLMChain并集成Memory
chain = LLMChain(llm=OpenAI(),prompt=prompt,memory=memory,verbose=True
)# 第一次调用
result = chain.run("你好，我是小明。")
print(result)# 第二次调用，Memory会自动包含第一次的对话历史
result = chain.run("我今天感觉很累，有什么建议吗？")
print(result)

# langchain/prompts/base.pyfrom abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Unionfrom langchain.schema import BaseMessageclass BasePromptTemplate(ABC):"""提示模板抽象基类，定义了所有提示模板必须遵循的接口"""@property@abstractmethoddef input_variables(self) -> List[str]:"""返回提示模板期望的输入变量列表"""pass@abstractmethoddef format_prompt(self, **kwargs: Any) -> PromptValue:"""格式化提示模板，返回PromptValue对象"""pass@abstractmethoddef format(self, **kwargs: Any) -> str:"""格式化提示模板，返回字符串"""passdef _get_prompt_dict(self) -> Dict[str, Any]:"""返回提示模板的字典表示"""return {"input_variables": self.input_variables,"template": self.format(),}

# langchain/prompts/prompt.pyfrom typing import Any, Dict, List, Optional, Unionfrom langchain.prompts.base import BasePromptTemplate, PromptValue
from langchain.schema import BaseMessage, HumanMessageclass PromptTemplate(BasePromptTemplate):"""基于字符串格式化的提示模板"""# 模板字符串template: str# 输入变量列表input_variables: List[str]# 模板格式，默认为f-stringtemplate_format: str = "f-string"# 是否验证模板validate_template: bool = Truedef __init__(self,input_variables: List[str],template: str,template_format: str = "f-string",validate_template: bool = True,**kwargs: Any,):"""初始化PromptTemplate"""super().__init__(**kwargs)self.input_variables = input_variablesself.template = templateself.template_format = template_format# 验证模板if validate_template:self._validate_template()def _validate_template(self) -> None:"""验证模板格式是否正确"""if self.template_format == "f-string":# 简单验证f-string格式try:# 使用空字典进行格式化，检查是否有未定义的变量self.format(**{v: "" for v in self.input_variables})except Exception as e:raise ValueError(f"模板验证失败: {e}")# 其他格式的验证...def format_prompt(self, **kwargs: Any) -> PromptValue:"""格式化提示模板"""# 检查所有必需的输入变量都已提供missing_vars = set(self.input_variables) - set(kwargs.keys())if missing_vars:raise ValueError(f"缺少必需的输入变量: {missing_vars}")# 格式化模板formatted = self.format(**kwargs)# 返回PromptValue对象return StringPromptValue(text=formatted)def format(self, **kwargs: Any) -> str:"""格式化模板字符串"""if self.template_format == "f-string":# 使用f-string格式化return self.template.format(**kwargs)elif self.template_format == "jinja2":# 使用Jinja2格式化from jinja2 import Templatereturn Template(self.template).render(**kwargs)else:raise ValueError(f"不支持的模板格式: {self.template_format}")def _get_prompt_dict(self) -> Dict[str, Any]:"""返回提示模板的字典表示"""return {"input_variables": self.input_variables,"template": self.template,"template_format": self.template_format,}

# langchain/prompts/few_shot.pyfrom typing import Any, Dict, List, Optional, Unionfrom langchain.prompts.base import BasePromptTemplate, PromptValue
from langchain.prompts.example_selector.base import BaseExampleSelector
from langchain.prompts.prompt import PromptTemplate
from langchain.schema import BaseMessage, HumanMessageclass FewShotPromptTemplate(BasePromptTemplate):"""少样本提示模板，用于生成包含示例的提示"""# 示例选择器或示例列表examples: Union[List[Dict[str, Any]], BaseExampleSelector]# 示例格式提示example_prompt: PromptTemplate# 连接示例的字符串example_separator: str = "\n\n"# 前缀文本prefix: str = ""# 后缀文本suffix: str = ""# 输入变量input_variables: List[str]# 模板格式template_format: str = "f-string"def format_prompt(self, **kwargs: Any) -> PromptValue:"""格式化提示模板"""# 获取示例if isinstance(self.examples, BaseExampleSelector):examples = self.examples.select_examples(kwargs)else:examples = self.examples# 格式化每个示例example_strings = [self.example_prompt.format(**example) for example in examples]# 连接示例example_string = self.example_separator.join(example_strings)# 构建完整的提示pieces = [self.prefix, example_string, self.suffix]full_prompt = "\n\n".join([p for p in pieces if p])# 格式化最终提示formatted_prompt = full_prompt.format(**kwargs)return StringPromptValue(text=formatted_prompt)def format(self, **kwargs: Any) -> str:"""格式化提示模板，返回字符串"""return self.format_prompt(**kwargs).to_string()def _get_prompt_dict(self) -> Dict[str, Any]:"""返回提示模板的字典表示"""return {"examples": self.examples,"example_prompt": self.example_prompt,"example_separator": self.example_separator,"prefix": self.prefix,"suffix": self.suffix,"input_variables": self.input_variables,"template_format": self.template_format,}

# langchain/agents/base.pyfrom abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Tuple, Unionfrom langchain.base_language import BaseLanguageModel
from langchain.callbacks.base import BaseCallbackManager
from langchain.chains.base import Chain
from langchain.prompts.base import BasePromptTemplate
from langchain.schema import AgentAction, AgentFinish, OutputParserExceptionclass Agent(ABC):"""Agent抽象基类，定义了所有Agent必须遵循的接口"""# 语言模型llm: BaseLanguageModel# 工具列表tools: List[BaseTool]# 提示模板prompt: BasePromptTemplate# 回调管理器callback_manager: Optional[BaseCallbackManager] = None@classmethod@abstractmethoddef create_prompt(cls, tools: List[BaseTool]) -> BasePromptTemplate:"""创建Agent使用的提示模板"""pass@classmethod@abstractmethoddef _get_default_output_parser(cls, **kwargs: Any) -> AgentOutputParser:"""获取默认的输出解析器"""pass@property@abstractmethoddef observation_prefix(self) -> str:"""观察结果的前缀"""pass@property@abstractmethoddef llm_prefix(self) -> str:"""LLM思考的前缀"""pass@abstractmethoddef _construct_scratchpad(self, intermediate_steps: List[Tuple[AgentAction, str]]) -> str:"""构建思考过程的草稿本"""pass@classmethod@abstractmethoddef from_llm_and_tools(cls,llm: BaseLanguageModel,tools: List[BaseTool],callback_manager: Optional[BaseCallbackManager] = None,**kwargs: Any,) -> "Agent":"""从LLM和工具列表创建Agent"""pass@abstractmethoddef plan(self,intermediate_steps: List[Tuple[AgentAction, str]],callbacks: Optional[BaseCallbackManager] = None,**kwargs: Any,) -> Union[AgentAction, AgentFinish]:"""根据中间步骤和输入计划下一步行动"""passasync def aplan(self,intermediate_steps: List[Tuple[AgentAction, str]],callbacks: Optional[BaseCallbackManager] = None,**kwargs: Any,) -> Union[AgentAction, AgentFinish]:"""异步根据中间步骤和输入计划下一步行动"""return self.plan(intermediate_steps, callbacks=callbacks, **kwargs)

# langchain/agents/agent.pyfrom abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Tuple, Unionfrom langchain.schema import AgentAction, AgentFinishclass AgentOutputParser(ABC):"""Agent输出解析器抽象基类"""@abstractmethoddef parse(self, text: str) -> Union[AgentAction, AgentFinish]:"""解析LLM输出"""passdef get_format_instructions(self) -> str:"""获取格式说明"""return ""

# langchain/agents/zero_shot.pyfrom typing import Any, Dict, List, Optional, Tuple, Unionfrom langchain.agents.agent import Agent, AgentOutputParser
from langchain.agents.agent_types import AgentType
from langchain.base_language import BaseLanguageModel
from langchain.callbacks.base import BaseCallbackManager
from langchain.chains import LLMChain
from langchain.prompts import BasePromptTemplate, PromptTemplate
from langchain.schema import AgentAction, AgentFinish, OutputParserExceptionclass ZeroShotAgent(Agent):"""零样本Agent，能够在没有预定义示例的情况下使用工具"""@classmethoddef create_prompt(cls,tools: List[BaseTool],prefix: str = DEFAULT_PREFIX,suffix: str = DEFAULT_SUFFIX,format_instructions: str = FORMAT_INSTRUCTIONS,input_variables: Optional[List[str]] = None,) -> BasePromptTemplate:"""创建提示模板"""tool_strings = "\n".join([f"{tool.name}: {tool.description}" for tool in tools])tool_names = ", ".join([tool.name for tool in tools])format_instructions = format_instructions.format(tool_names=tool_names)template = "\n\n".join([prefix, tool_strings, format_instructions, suffix])if input_variables is None:input_variables = ["input", "agent_scratchpad"]return PromptTemplate(template=template, input_variables=input_variables)@classmethoddef _get_default_output_parser(cls, **kwargs: Any) -> AgentOutputParser:"""获取默认的输出解析器"""return ZeroShotAgentOutputParser()@propertydef observation_prefix(self) -> str:"""观察结果的前缀"""return "Observation: "@propertydef llm_prefix(self) -> str:"""LLM思考的前缀"""return "Thought: "def _construct_scratchpad(self, intermediate_steps: List[Tuple[AgentAction, str]]) -> str:"""构建思考过程的草稿本"""thoughts = ""for action, observation in intermediate_steps:thoughts += f"{action.log}\n{self.observation_prefix}{observation}\n{self.llm_prefix}"return thoughts@classmethoddef from_llm_and_tools(cls,llm: BaseLanguageModel,tools: List[BaseTool],callback_manager: Optional[BaseCallbackManager] = None,**kwargs: Any,) -> "ZeroShotAgent":"""从LLM和工具列表创建Agent"""cls._validate_tools(tools)prompt = cls.create_prompt(tools, **kwargs)output_parser = kwargs.pop("output_parser", cls._get_default_output_parser())return cls(llm_chain=LLMChain(llm=llm,prompt=prompt,callback_manager=callback_manager,),tools=tools,output_parser=output_parser,**kwargs,)def plan(self,intermediate_steps: List[Tuple[AgentAction, str]],callbacks: Optional[BaseCallbackManager] = None,**kwargs: Any,) -> Union[AgentAction, AgentFinish]:"""根据中间步骤和输入计划下一步行动"""# 构建思考草稿本thoughts = self._construct_scratchpad(intermediate_steps)# 构建输入inputs = {"input": kwargs["input"],"agent_scratchpad": thoughts,**{k: v for k, v in kwargs.items() if k != "input"},}# 调用LLM生成下一步行动output = self.llm_chain.predict(callbacks=callbacks, **inputs)# 解析LLM输出try:return self.output_parser.parse(output)except OutputParserException as e:return AgentFinish(return_values={"output": str(e)},log=output,)

# langchain/agents/agent_executor.pyfrom typing import Any, Dict, List, Optional, Tuple, Unionfrom langchain.agents.agent import Agent, AgentOutputParser
from langchain.agents.tools import BaseTool
from langchain.base_language import BaseLanguageModel
from langchain.callbacks.base import BaseCallbackManager
from langchain.chains.base import Chain
from langchain.schema import AgentAction, AgentFinish, OutputParserExceptionclass AgentExecutor(Chain):"""Agent执行器，负责执行Agent并管理工具调用"""# Agentagent: Agent# 工具列表tools: List[BaseTool]# 最大迭代次数max_iterations: Optional[int] = 15# 是否返回中间步骤return_intermediate_steps: bool = False# 工具名称到工具的映射tool_names: List[str]# 工具名称到工具的映射name_to_tool_map: Dict[str, BaseTool]@propertydef input_keys(self) -> List[str]:"""返回期望的输入键"""return self.agent.input_keys@propertydef output_keys(self) -> List[str]:"""返回产生的输出键"""return self.agent.output_keys@classmethoddef from_agent_and_tools(cls,agent: Agent,tools: List[BaseTool],callback_manager: Optional[BaseCallbackManager] = None,**kwargs: Any,) -> "AgentExecutor":"""从Agent和工具列表创建执行器"""name_to_tool_map = {tool.name: tool for tool in tools}tool_names = list(name_to_tool_map.keys())return cls(agent=agent,tools=tools,name_to_tool_map=name_to_tool_map,tool_names=tool_names,callback_manager=callback_manager,**kwargs,)def _call(self,inputs: Dict[str, Any],run_manager: Optional[CallbackManagerForChainRun] = None,) -> Dict[str, Any]:"""执行Agent"""# 准备输入intermediate_steps: List[Tuple[AgentAction, str]] = []agent_inputs = inputs# 获取终止消息stop = agent_inputs.get("stop", None)# 主循环iterations = 0while self._should_continue(iterations, intermediate_steps):# 调用Agent规划下一步行动next_step_output = self._take_next_step(agent_inputs, intermediate_steps, run_manager=run_manager)# 如果Agent返回完成，直接返回结果if isinstance(next_step_output, AgentFinish):if self.return_intermediate_steps:return {**next_step_output.return_values,"intermediate_steps": intermediate_steps,}else:return next_step_output.return_values# 否则，执行工具并记录结果action = next_step_outputif run_manager:run_manager.on_agent_action(action)# 执行工具observation = self._call_tool(action, run_manager=run_manager)# 记录中间步骤intermediate_steps.append((action, observation))# 更新迭代次数iterations += 1# 如果达到最大迭代次数，返回错误output = self.agent.return_stopped_response(self.agent.llm_chain.llm, intermediate_steps, False)if self.return_intermediate_steps:return {**output.return_values, "intermediate_steps": intermediate_steps}else:return output.return_valuesdef _take_next_step(self,inputs: Dict[str, Any],intermediate_steps: List[Tuple[AgentAction, str]],run_manager: Optional[CallbackManagerForChainRun] = None,) -> Union[AgentAction, AgentFinish]:"""获取下一步行动"""# 调用Agent规划下一步return self.agent.plan(intermediate_steps,callbacks=run_manager.get_child() if run_manager else None,**inputs,)def _call_tool(self,action: AgentAction,run_manager: Optional[CallbackManagerForChainRun] = None,) -> str:"""调用工具并返回观察结果"""if action.tool not in self.name_to_tool_map:return (f"错误: 工具 '{action.tool}' 不存在。可用工具: {', '.join(self.tool_names)}")# 获取工具tool = self.name_to_tool_map[action.tool]# 执行工具observation = tool.run(action.tool_input,verbose=self.verbose,callbacks=run_manager.get_child() if run_manager else None,)return observationdef _should_continue(self, iterations: int, intermediate_steps: List[Tuple[AgentAction, str]]) -> bool:"""判断是否应该继续执行"""if self.max_iterations is not None and iterations >= self.max_iterations:return Falsereturn True

# langchain/tools/base.pyfrom abc import ABC, abstractmethod
from typing import Any, Callable, Dict, List, Optional, Sequence, Type, Unionfrom langchain.callbacks.base import BaseCallbackManager
from langchain.schema import BaseToolclass Tool(BaseTool, ABC):"""Tool抽象基类，定义了所有工具必须遵循的接口"""# 工具名称name: str# 工具描述description: str# 是否异步支持is_async: bool = Falsedef __init__(self,name: str,description: str,func: Optional[Callable[..., str]] = None,coroutine: Optional[Callable[..., str]] = None,**kwargs: Any,):"""初始化工具"""super().__init__(**kwargs)self.name = nameself.description = descriptionself.func = funcself.coroutine = coroutine@abstractmethoddef _run(self, *args: Any, **kwargs: Any) -> str:"""同步运行工具的核心方法"""passasync def _arun(self, *args: Any, **kwargs: Any) -> str:"""异步运行工具的核心方法，默认调用同步方法"""return self._run(*args, **kwargs)def run(self,tool_input: Union[str, Dict],verbose: bool = False,callbacks: Optional[BaseCallbackManager] = None,) -> str:"""运行工具"""# 处理输入if isinstance(tool_input, str):parsed_input = (tool_input,)kwargs = {}else:parsed_input = ()kwargs = tool_input# 执行工具if self.is_async:# 如果是异步工具，使用asyncio运行import asyncioreturn asyncio.run(self._arun(*parsed_input, **kwargs))else:return self._run(*parsed_input, **kwargs)async def arun(self,tool_input: Union[str, Dict],verbose: bool = False,callbacks: Optional[BaseCallbackManager] = None,) -> str:"""异步运行工具"""if isinstance(tool_input, str):parsed_input = (tool_input,)kwargs = {}else:parsed_input = ()kwargs = tool_inputreturn await self._arun(*parsed_input, **kwargs)

# langchain/tools/python/tool.pyclass PythonREPLTool(BaseTool):"""Python REPL工具，允许执行Python代码"""name = "python_repl"description = ("一个Python shell。适合执行Python代码。""输入应该是有效的Python代码。""如果您期望输出，应该打印它。")def __init__(self):super().__init__()self.repl = PythonREPL()def _run(self, code: str) -> str:"""执行Python代码并返回结果"""return self.repl.run(code)async def _arun(self, code: str) -> str:"""异步执行Python代码并返回结果"""return await asyncio.to_thread(self._run, code)

# langchain/tools/serpapi.pyclass SerpAPIWrapper(BaseTool):"""SerpAPI工具，用于搜索引擎查询"""name = "serpapi"description = ("一个搜索引擎。适合回答关于当前事件的问题。""输入应该是一个搜索查询。")def __init__(self, serpapi_api_key: Optional[str] = None):super().__init__()self.serpapi_api_key = serpapi_api_key or os.environ.get("SERPAPI_API_KEY")if not self.serpapi_api_key:raise ValueError("需要设置SERPAPI_API_KEY环境变量或提供serpapi_api_key参数")def _run(self, query: str) -> str:"""执行搜索查询并返回结果"""params = {"q": query,"api_key": self.serpapi_api_key,"engine": "google","gl": "us","hl": "en",}response = requests.get("https://serpapi.com/search", params=params)response.raise_for_status()return response.json()async def _arun(self, query: str) -> str:"""异步执行搜索查询并返回结果"""return await asyncio.to_thread(self._run, query)

# langchain/tools/calculator.pyclass CalculatorTool(BaseTool):"""计算器工具，用于执行数学计算"""name = "calculator"description = ("一个计算器。适合执行复杂的数学计算。""输入应该是一个数学表达式。")def _run(self, query: str) -> str:"""执行数学计算并返回结果"""try:# 使用ast来安全地评估数学表达式return str(eval（query, {"__builtins__": None}, {}))except Exception as e:return f"错误: {str(e)}"async def _arun(self, query: str) -> str:"""异步执行数学计算并返回结果"""return await asyncio.to_thread(self._run, query)

from langchain.tools import BaseToolclass GetWeatherTool(BaseTool):"""获取天气信息的工具"""name = "get_weather"description = ("获取指定城市的当前天气信息。""输入应该是城市名称。")def _run(self, city: str) -> str:"""获取并返回城市的天气信息"""# 实际实现中，这里会调用天气API# 为简化示例，我们返回模拟数据weather_data = {"北京": "晴朗，25°C","上海": "多云，28°C","广州": "小雨，27°C","深圳": "多云，29°C",}if city in weather_data:return f"{city}的当前天气是{weather_data[city]}"else:return f"抱歉，没有找到{city}的天气信息"async def _arun(self, city: str) -> str:"""异步获取并返回城市的天气信息"""return await asyncio.to_thread(self._run, city)

from langchain.tools import BaseTool
from pydantic import BaseModel, Fieldclass GetWeatherInput(BaseModel):"""获取天气工具的输入参数"""city: str = Field(..., description="城市名称")class GetWeatherTool(BaseTool):"""获取天气信息的工具"""name = "get_weather"description = ("获取指定城市的当前天气信息。""输入应该是城市名称。")args_schema: Type[BaseModel] = GetWeatherInputdef _run(self, city: str) -> str:"""获取并返回城市的天气信息"""# 实际实现中，这里会调用天气APIweather_data = {"北京": "晴朗，25°C","上海": "多云，28°C","广州": "小雨，27°C","深圳": "多云，29°C",}if city in weather_data:return f"{city}的当前天气是{weather_data[city]}"else:return f"抱歉，没有找到{city}的天气信息"

from langchain.tools import BaseToolclass AsyncHTTPRequestTool(BaseTool):"""异步HTTP请求工具"""name = "http_request"description = ("发送HTTP请求并获取响应。""输入应该是一个包含URL和可选参数的字典。")is_async: bool = Trueasync def _arun(self, params: dict) -> str:"""异步发送HTTP请求并返回响应"""url = params.get("url")method = params.get("method", "GET")headers = params.get("headers", {})data = params.get("data", {})if not url:return "错误: 缺少URL参数"async with aiohttp.ClientSession() as session:async with session.request(method, url, headers=headers, json=data) as response:return await response.text()

# langchain/llms/base.pyclass LLM(ABC):# ... 其他方法 ...async def agenerate(self, prompts: List[str], stop: Optional[List[str]] = None) -> LLMResult:"""异步生成多个提示的响应"""passasync def _acall(self, prompt: str, stop: Optional[List[str]] = None) -> str:"""异步调用LLM的核心方法"""pass

# langchain/llms/openai.pyclass OpenAI(LLM):# ... 其他方法 ...async def _acall(self, prompt: str, stop: Optional[List[str]] = None) -> str:"""异步执行OpenAI API调用"""params = {"model": self.model_name,"prompt": prompt,"temperature": self.temperature,"max_tokens": self.max_tokens,"top_p": self.top_p,