LangChain主库深度解析：Chain、Agent与Memory的实现机制

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

# langchain/__init__.py
"""包的主要入口点。

Main entrypoint into package."""

import warnings
from importlib import metadata
from typing import Any, Optional

from langchain_core._api.deprecation import surface_langchain_deprecation_warnings

try:
    __version__ = metadata.version(__package__)
except metadata.PackageNotFoundError:
    # Case where package metadata is not available.
    __version__ = ""
del metadata  # optional, avoids polluting the results of dir(__package__)


def _warn_on_import(name: str, replacement: Optional[str] = None) -> None:
    """导入已弃用模块时发出警告。

    Warn on import of deprecated module."""
    from langchain._api.interactive_env import is_interactive_env

    if is_interactive_env():
        # No warnings for interactive environments.
        # This is done to avoid polluting the output of interactive environments
        # where users rely on auto-complete and may trigger this warning
        # even if they are not using any deprecated modules
        return

    if replacement:
        warnings.warn(
            f"Importing {name} from langchain root module is no longer supported. "
            f"Please use {replacement} instead.",
            stacklevel=3,
        )
    else:
        warnings.warn(
            f"Importing {name} from langchain root module is no longer supported.",
            stacklevel=3,
        )


# Surfaces Deprecation and Pending Deprecation warnings from langchain.
surface_langchain_deprecation_warnings()

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425

from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Union
from langchain_core.runnables import RunnableSerializable
from langchain_core.memory import BaseMemory
from langchain_core.callbacks import Callbacks

class Chain(RunnableSerializable[Dict[str, Any], Dict[str, Any]], ABC):
    """创建结构化组件调用序列的抽象基类。

    Chain应该用于编码对模型、文档检索器、其他Chain等组件的调用序列，
    并为此序列提供简单的接口。

    Chain接口使创建以下应用变得容易：
    - 有状态的：向任何Chain添加Memory以赋予其状态
    - 可观察的：向Chain传递Callbacks以在主要组件调用序列之外执行额外功能
    - 可组合的：Chain API足够灵活，可以轻松地将Chain与其他组件（包括其他Chain）结合

    主要暴露的方法：
    - __call__：Chain是可调用的。__call__方法是执行Chain的主要方式
    - run：一个便利方法，接受args/kwargs作为输入并返回字符串或对象输出
    """

    # === 核心属性 ===

    memory: Optional[BaseMemory] = None
    """可选的内存对象。默认为None。
    Memory是一个在每个链的开始和结束时被调用的类。
    在开始时，memory加载变量并在链中传递它们。
    在结束时，它保存任何返回的变量。
    有许多不同类型的memory - 请参阅memory文档以获取完整目录。"""

    callbacks: Callbacks = Field(default=None, exclude=True)
    """可选的回调处理器列表（或回调管理器）。默认为None。
    回调处理器在链调用的整个生命周期中被调用，
    从on_chain_start开始，以on_chain_end或on_chain_error结束。
    每个自定义链可以选择调用额外的回调方法，详见Callback文档。"""

    verbose: bool = Field(default_factory=_get_verbosity)
    """是否以详细模式运行。在详细模式下，一些中间日志将被打印到控制台。
    默认为`langchain.verbose`值。"""

    tags: Optional[List[str]] = Field(default=None, exclude=True)
    """与链关联的可选标签列表。默认为None。
    这些标签将与对此链的每次调用相关联，
    并作为参数传递给在`callbacks`中定义的处理器。
    您可以使用这些来例如识别链的特定实例与其用例。"""

    metadata: Optional[Dict[str, Any]] = Field(default=None, exclude=True)
    """与链关联的可选元数据。默认为None。
    此元数据将与对此链的每次调用相关联，
    并作为参数传递给在`callbacks`中定义的处理器。
    您可以使用这些来例如识别链的特定实例与其用例。"""

    # === 抽象属性 ===

    @property
    @abstractmethod
    def input_keys(self) -> List[str]:
        """输入键。

        :meta private:
        """

    @property
    @abstractmethod
    def output_keys(self) -> List[str]:
        """输出键。

        :meta private:
        """

    # === 抽象方法 ===

    @abstractmethod
    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """执行链。

        这是一个不面向用户的私有方法。它只在Chain.__call__内部调用，
        Chain.__call__是处理回调配置和一些输入/输出处理的面向用户的包装方法。

        Args:
            inputs: 链的命名输入字典。假定包含Chain.input_keys中指定的所有输入，
                   包括memory添加的任何输入。
            run_manager: 包含此链运行的回调处理器的回调管理器。

        Returns:
            命名输出字典。应包含Chain.output_keys中指定的所有输出。
        """

    async def _acall(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """异步执行链。

        这是一个不面向用户的私有方法。它只在Chain.acall内部调用，
        Chain.acall是处理回调配置和一些输入/输出处理的面向用户的包装方法。

        Args:
            inputs: 链的命名输入字典。假定包含Chain.input_keys中指定的所有输入，
                   包括memory添加的任何输入。
            run_manager: 包含此链运行的回调处理器的回调管理器。

        Returns:
            命名输出字典。应包含Chain.output_keys中指定的所有输出。
        """
        return await run_in_executor(
            None,
            self._call,
            inputs,
            run_manager.get_sync() if run_manager else None,
        )

    # === 输入输出处理 ===

    def prep_inputs(self, inputs: Union[Dict[str, Any], Any]) -> Dict[str, Any]:
        """验证并准备链输入，包括添加来自memory的输入。

        Args:
            inputs: 字典输入或单个输入（如果链只期望一个输入）。

        Returns:
            包含所有输入的字典，包括来自memory的输入。
        """
        if not isinstance(inputs, dict):
            _input_keys = set(self.input_keys)
            if self.memory is not None:
                # 如果有memory，我们需要检查它添加了哪些键
                _input_keys = _input_keys.difference(self.memory.memory_variables)
            if len(_input_keys) != 1:
                raise ValueError(
                    f"单个输入传递给期望多个输入的链: {_input_keys}。"
                    f"请传递一个字典，例如 {{'foo': 'bar'}}"
                )
            inputs = {list(_input_keys)[0]: inputs}

        if self.memory is not None:
            external_context = self.memory.load_memory_variables(inputs)
            inputs = dict(inputs, **external_context)

        self._validate_inputs(inputs)
        return inputs

    def prep_outputs(
        self,
        inputs: Dict[str, Any],
        outputs: Dict[str, Any],
        return_only_outputs: bool = False,
    ) -> Dict[str, Any]:
        """验证并准备链输出，并保存到memory。

        Args:
            inputs: 链输入字典。
            outputs: 链输出字典。
            return_only_outputs: 是否只返回输出。
                如果为False，输入也会包含在最终输出中。

        Returns:
            包含链输出的字典，如果return_only_outputs为False，还包含输入。
        """
        self._validate_outputs(outputs)
        if self.memory is not None:
            self.memory.save_context(inputs, outputs)
        if return_only_outputs:
            return outputs
        else:
            return {**inputs, **outputs}

    def _validate_inputs(self, inputs: Dict[str, Any]) -> None:
        """检查链输入是否有效。

        Args:
            inputs: 链输入字典。

        Raises:
            ValueError: 如果输入无效。
        """
        missing_keys = set(self.input_keys).difference(inputs)
        if missing_keys:
            raise ValueError(f"缺少一些输入键: {missing_keys}")

    def _validate_outputs(self, outputs: Dict[str, Any]) -> None:
        """检查链输出是否有效。

        Args:
            outputs: 链输出字典。

        Raises:
            ValueError: 如果输出无效。
        """
        missing_keys = set(self.output_keys).difference(outputs)
        if missing_keys:
            raise ValueError(f"缺少一些输出键: {missing_keys}")

    # === Runnable接口实现 ===

    def invoke(
        self,
        input: Dict[str, Any],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        """Runnable接口的invoke实现。"""
        config = ensure_config(config)

        return self(
            input,
            callbacks=config.get("callbacks"),
            tags=config.get("tags"),
            metadata=config.get("metadata"),
            run_name=config.get("run_name"),
            **kwargs,
        )

    async def ainvoke(
        self,
        input: Dict[str, Any],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        """异步invoke实现。"""
        config = ensure_config(config)

        return await self.acall(
            input,
            callbacks=config.get("callbacks"),
            tags=config.get("tags"),
            metadata=config.get("metadata"),
            run_name=config.get("run_name"),
            **kwargs,
        )

    # === 主要执行方法 ===

    @deprecated("0.1.0", alternative="invoke", removal="1.0")
    def __call__(
        self,
        inputs: Union[Dict[str, Any], Any],
        return_only_outputs: bool = False,
        callbacks: Callbacks = None,
        *,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        run_name: Optional[str] = None,
        include_run_info: bool = False,
    ) -> Dict[str, Any]:
        """执行链。

        Args:
            inputs: 链输入字典。
            return_only_outputs: 是否只返回输出。
            callbacks: 要使用的回调。
            tags: 要附加到运行的标签。
            metadata: 要附加到运行的元数据。
            run_name: 运行的名称。
            include_run_info: 是否在输出中包含运行信息。

        Returns:
            链输出字典。
        """
        config = {
            "callbacks": callbacks,
            "tags": tags,
            "metadata": metadata,
            "run_name": run_name,
        }

        inputs = self.prep_inputs(inputs)
        callback_manager = get_callback_manager_for_config(config)
        new_arg_supported = inspect.signature(self._call).parameters.get("run_manager")

        with callback_manager.on_chain_start(
            dumpd(self),
            inputs,
            name=run_name,
        ) as run_manager:
            try:
                if new_arg_supported:
                    outputs = self._call(inputs, run_manager=run_manager)
                else:
                    outputs = self._call(inputs)
            except Exception as e:
                run_manager.on_chain_error(e)
                raise
            else:
                run_manager.on_chain_end(outputs)
                final_outputs: Dict[str, Any] = self.prep_outputs(
                    inputs, outputs, return_only_outputs
                )
                if include_run_info:
                    final_outputs[RUN_KEY] = RunInfo(run_id=run_manager.run_id)
                return final_outputs

    async def acall(
        self,
        inputs: Union[Dict[str, Any], Any],
        return_only_outputs: bool = False,
        callbacks: Callbacks = None,
        *,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        run_name: Optional[str] = None,
        include_run_info: bool = False,
    ) -> Dict[str, Any]:
        """异步执行链。"""
        config = {
            "callbacks": callbacks,
            "tags": tags,
            "metadata": metadata,
            "run_name": run_name,
        }

        inputs = self.prep_inputs(inputs)
        callback_manager = get_async_callback_manager_for_config(config)
        new_arg_supported = inspect.signature(self._acall).parameters.get("run_manager")

        async with callback_manager.on_chain_start(
            dumpd(self),
            inputs,
            name=run_name,
        ) as run_manager:
            try:
                if new_arg_supported:
                    outputs = await self._acall(inputs, run_manager=run_manager)
                else:
                    outputs = await self._acall(inputs)
            except Exception as e:
                await run_manager.on_chain_error(e)
                raise
            else:
                await run_manager.on_chain_end(outputs)
                final_outputs: Dict[str, Any] = self.prep_outputs(
                    inputs, outputs, return_only_outputs
                )
                if include_run_info:
                    final_outputs[RUN_KEY] = RunInfo(run_id=run_manager.run_id)
                return final_outputs

    # === 便利方法 ===

    def run(
        self,
        *args: Any,
        callbacks: Callbacks = None,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> Any:
        """便利方法，用于执行链并返回单个字符串输出。

        Args:
            *args: 如果链只期望一个输入，则可以作为位置参数传递。
            callbacks: 要使用的回调。
            tags: 要附加到运行的标签。
            metadata: 要附加到运行的元数据。
            **kwargs: 如果链期望多个输入，则可以作为关键字参数传递。

        Returns:
            链输出。如果链有单个输出键，则返回该键的值。
            否则，返回包含所有输出的字典。
        """
        if args and not kwargs:
            if len(args) != 1:
                raise ValueError("`run` 支持单个位置参数或关键字参数，不支持两者。")
            if self.input_keys and len(self.input_keys) == 1:
                inputs = {self.input_keys[0]: args[0]}
            else:
                inputs = args[0]
        elif kwargs and not args:
            inputs = kwargs
        else:
            raise ValueError("`run` 支持单个位置参数或关键字参数，不支持两者。")

        outputs = self(
            inputs,
            callbacks=callbacks,
            tags=tags,
            metadata=metadata,
            return_only_outputs=True,
        )

        # 如果只有一个输出键，返回该值
        if len(self.output_keys) == 1:
            return outputs[self.output_keys[0]]
        else:
            return outputs

    async def arun(
        self,
        *args: Any,
        callbacks: Callbacks = None,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> Any:
        """异步便利方法。"""
        if args and not kwargs:
            if len(args) != 1:
                raise ValueError("`arun` 支持单个位置参数或关键字参数，不支持两者。")
            if self.input_keys and len(self.input_keys) == 1:
                inputs = {self.input_keys[0]: args[0]}
            else:
                inputs = args[0]
        elif kwargs and not args:
            inputs = kwargs
        else:
            raise ValueError("`arun` 支持单个位置参数或关键字参数，不支持两者。")

        outputs = await self.acall(
            inputs,
            callbacks=callbacks,
            tags=tags,
            metadata=metadata,
            return_only_outputs=True,
        )

        if len(self.output_keys) == 1:
            return outputs[self.output_keys[0]]
        else:
            return outputs

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276

from typing import Any, Dict, List, Optional
from langchain_core.language_models import BaseLanguageModel
from langchain_core.prompts import BasePromptTemplate
from langchain_core.output_parsers import BaseOutputParser

class LLMChain(Chain):
    """使用语言模型的链。

    这是最基础的链实现，将提示模板、语言模型和输出解析器组合在一起。

    示例:
        .. code-block:: python

            from langchain import LLMChain, OpenAI, PromptTemplate
            prompt_template = "Tell me a {adjective} joke"
            prompt = PromptTemplate(
                input_variables=["adjective"], template=prompt_template
            )
            llm = LLMChain(llm=OpenAI(), prompt=prompt)
    """

    prompt: BasePromptTemplate
    """要使用的提示模板。"""

    llm: BaseLanguageModel
    """要使用的语言模型。"""

    output_parser: BaseOutputParser = Field(default_factory=NoOpOutputParser)
    """要使用的输出解析器。默认为简单的字符串解析器。"""

    output_key: str = "text"  #: :meta private:
    """输出键。"""

    return_final_only: bool = True
    """是否只返回最终解析的结果。如果为False，将返回原始文本和解析结果。"""

    llm_kwargs: dict = Field(default_factory=dict)
    """传递给LLM的额外关键字参数。"""

    @property
    def input_keys(self) -> List[str]:
        """输入键。

        :meta private:
        """
        return self.prompt.input_variables

    @property
    def output_keys(self) -> List[str]:
        """输出键。

        :meta private:
        """
        if self.return_final_only:
            return [self.output_key]
        else:
            return [self.output_key, f"{self.output_key}_raw"]

    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, str]:
        """执行LLM链。

        Args:
            inputs: 输入字典，包含提示模板所需的所有变量。
            run_manager: 回调管理器。

        Returns:
            包含LLM输出的字典。
        """
        # 1. 格式化提示
        prompt_value = self.prompt.format_prompt(**inputs)

        # 2. 调用LLM
        response = self.llm.generate_prompt(
            [prompt_value],
            callbacks=run_manager.get_child() if run_manager else None,
            **self.llm_kwargs,
        )

        # 3. 提取文本
        if self.llm.llm_type == "openai-chat":
            # 聊天模型返回消息
            text_output = response.generations[0][0].message.content
        else:
            # 文本模型返回文本
            text_output = response.generations[0][0].text

        # 4. 解析输出
        if self.output_parser:
            parsed_output = self.output_parser.parse(text_output)
        else:
            parsed_output = text_output

        # 5. 构建返回结果
        if self.return_final_only:
            return {self.output_key: parsed_output}
        else:
            return {
                self.output_key: parsed_output,
                f"{self.output_key}_raw": text_output,
            }

    async def _acall(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
    ) -> Dict[str, str]:
        """异步执行LLM链。"""
        # 1. 格式化提示
        prompt_value = self.prompt.format_prompt(**inputs)

        # 2. 异步调用LLM
        response = await self.llm.agenerate_prompt(
            [prompt_value],
            callbacks=run_manager.get_child() if run_manager else None,
            **self.llm_kwargs,
        )

        # 3. 提取文本
        if self.llm.llm_type == "openai-chat":
            text_output = response.generations[0][0].message.content
        else:
            text_output = response.generations[0][0].text

        # 4. 解析输出
        if self.output_parser:
            parsed_output = self.output_parser.parse(text_output)
        else:
            parsed_output = text_output

        # 5. 构建返回结果
        if self.return_final_only:
            return {self.output_key: parsed_output}
        else:
            return {
                self.output_key: parsed_output,
                f"{self.output_key}_raw": text_output,
            }

    def predict(self, **kwargs: Any) -> str:
        """便利方法，用于预测单个输出。

        Args:
            **kwargs: 提示模板的输入变量。

        Returns:
            LLM的预测结果。
        """
        return self.run(**kwargs)

    async def apredict(self, **kwargs: Any) -> str:
        """异步预测方法。"""
        return await self.arun(**kwargs)

    def predict_and_parse(self, **kwargs: Any) -> Union[str, List[str], Dict]:
        """预测并解析输出。

        Args:
            **kwargs: 提示模板的输入变量。

        Returns:
            解析后的输出。
        """
        result = self.predict(**kwargs)
        if self.output_parser:
            return self.output_parser.parse(result)
        else:
            return result

    async def apredict_and_parse(self, **kwargs: Any) -> Union[str, List[str], Dict]:
        """异步预测并解析输出。"""
        result = await self.apredict(**kwargs)
        if self.output_parser:
            return self.output_parser.parse(result)
        else:
            return result

    def apply(self, input_list: List[Dict[str, Any]]) -> List[Dict[str, str]]:
        """对输入列表应用链。

        Args:
            input_list: 输入字典列表。

        Returns:
            输出字典列表。
        """
        response = self.generate(input_list)
        return self.create_outputs(response)

    async def aapply(self, input_list: List[Dict[str, Any]]) -> List[Dict[str, str]]:
        """异步对输入列表应用链。"""
        response = await self.agenerate(input_list)
        return self.create_outputs(response)

    def generate(
        self,
        input_list: List[Dict[str, Any]],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> LLMResult:
        """为输入列表生成LLM结果。

        Args:
            input_list: 输入字典列表。
            run_manager: 回调管理器。

        Returns:
            LLM结果。
        """
        prompts = []
        for inputs in input_list:
            prompt_value = self.prompt.format_prompt(**inputs)
            prompts.append(prompt_value)

        return self.llm.generate_prompt(
            prompts,
            callbacks=run_manager.get_child() if run_manager else None,
            **self.llm_kwargs,
        )

    async def agenerate(
        self,
        input_list: List[Dict[str, Any]],
        run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
    ) -> LLMResult:
        """异步为输入列表生成LLM结果。"""
        prompts = []
        for inputs in input_list:
            prompt_value = self.prompt.format_prompt(**inputs)
            prompts.append(prompt_value)

        return await self.llm.agenerate_prompt(
            prompts,
            callbacks=run_manager.get_child() if run_manager else None,
            **self.llm_kwargs,
        )

    def create_outputs(self, llm_result: LLMResult) -> List[Dict[str, str]]:
        """从LLM结果创建输出。

        Args:
            llm_result: LLM结果。

        Returns:
            输出字典列表。
        """
        outputs = []
        for generation in llm_result.generations:
            # 提取文本
            if self.llm.llm_type == "openai-chat":
                text_output = generation[0].message.content
            else:
                text_output = generation[0].text

            # 解析输出
            if self.output_parser:
                parsed_output = self.output_parser.parse(text_output)
            else:
                parsed_output = text_output

            # 构建输出字典
            if self.return_final_only:
                outputs.append({self.output_key: parsed_output})
            else:
                outputs.append({
                    self.output_key: parsed_output,
                    f"{self.output_key}_raw": text_output,
                })

        return outputs

    @property
    def _chain_type(self) -> str:
        return "llm"

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237

class SequentialChain(Chain):
    """其中一个链的输出直接馈送到下一个链的链。

    这个链允许你将多个链串联起来，其中每个链的输出
    成为下一个链的输入。

    示例:
        .. code-block:: python

            from langchain import SequentialChain, LLMChain, OpenAI, PromptTemplate

            # 第一个链：生成故事大纲
            outline_template = "Write an outline for a story about {topic}"
            outline_prompt = PromptTemplate(
                input_variables=["topic"], template=outline_template
            )
            outline_chain = LLMChain(
                llm=OpenAI(), prompt=outline_prompt, output_key="outline"
            )

            # 第二个链：基于大纲写故事
            story_template = "Write a story based on this outline: {outline}"
            story_prompt = PromptTemplate(
                input_variables=["outline"], template=story_template
            )
            story_chain = LLMChain(
                llm=OpenAI(), prompt=story_prompt, output_key="story"
            )

            # 组合成序列链
            sequential_chain = SequentialChain(
                chains=[outline_chain, story_chain],
                input_variables=["topic"],
                output_variables=["story"]
            )
    """

    chains: List[Chain]
    """要按顺序运行的链列表。"""

    input_variables: List[str]
    """序列链期望的输入变量。"""

    output_variables: List[str]  #: :meta private:
    """序列链将输出的变量。"""

    return_all: bool = False
    """是否返回所有中间输出，还是只返回最终输出。"""

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        extra="forbid",
    )

    @property
    def input_keys(self) -> List[str]:
        """返回链的期望输入键。

        :meta private:
        """
        return self.input_variables

    @property
    def output_keys(self) -> List[str]:
        """返回输出键。

        :meta private:
        """
        return self.output_variables

    @model_validator(mode="before")
    @classmethod
    def validate_chains(cls, values: Dict[str, Any]) -> Dict[str, Any]:
        """验证链配置。

        Args:
            values: 配置值字典。

        Returns:
            验证后的配置值。

        Raises:
            ValueError: 如果链配置无效。
        """
        chains = values.get("chains", [])
        input_variables = values.get("input_variables", [])
        output_variables = values.get("output_variables", [])

        if not chains:
            raise ValueError("必须提供至少一个链")

        # 验证输入变量
        if not input_variables:
            raise ValueError("必须指定input_variables")

        # 验证输出变量
        if not output_variables:
            raise ValueError("必须指定output_variables")

        # 检查链的连接性
        cls._validate_chain_connectivity(chains, input_variables, output_variables)

        return values

    @classmethod
    def _validate_chain_connectivity(
        cls,
        chains: List[Chain],
        input_variables: List[str],
        output_variables: List[str],
    ) -> None:
        """验证链之间的连接性。

        Args:
            chains: 链列表。
            input_variables: 输入变量。
            output_variables: 输出变量。

        Raises:
            ValueError: 如果链连接无效。
        """
        # 跟踪可用变量
        available_vars = set(input_variables)

        for i, chain in enumerate(chains):
            # 检查当前链的输入是否都可用
            missing_inputs = set(chain.input_keys) - available_vars
            if missing_inputs:
                raise ValueError(
                    f"链 {i} 缺少输入变量: {missing_inputs}。"
                    f"可用变量: {available_vars}"
                )

            # 添加当前链的输出到可用变量
            available_vars.update(chain.output_keys)

        # 检查最终输出变量是否都可用
        missing_outputs = set(output_variables) - available_vars
        if missing_outputs:
            raise ValueError(
                f"输出变量 {missing_outputs} 在任何链中都不可用。"
                f"可用变量: {available_vars}"
            )

    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """执行序列链。

        Args:
            inputs: 输入字典。
            run_manager: 回调管理器。

        Returns:
            输出字典。
        """
        # 初始化已知变量
        known_values = inputs.copy()

        # 按顺序执行每个链
        for i, chain in enumerate(self.chains):
            # 准备当前链的输入
            chain_inputs = {
                key: known_values[key]
                for key in chain.input_keys
                if key in known_values
            }

            # 执行链
            if run_manager:
                child_manager = run_manager.get_child(f"chain_{i}")
                chain_outputs = chain(chain_inputs, callbacks=[child_manager])
            else:
                chain_outputs = chain(chain_inputs)

            # 更新已知变量
            known_values.update(chain_outputs)

        # 准备最终输出
        if self.return_all:
            # 返回所有变量（除了原始输入）
            outputs = {
                key: known_values[key]
                for key in known_values
                if key not in inputs
            }
        else:
            # 只返回指定的输出变量
            outputs = {
                key: known_values[key]
                for key in self.output_variables
            }

        return outputs

    async def _acall(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """异步执行序列链。"""
        known_values = inputs.copy()

        for i, chain in enumerate(self.chains):
            chain_inputs = {
                key: known_values[key]
                for key in chain.input_keys
                if key in known_values
            }

            if run_manager:
                child_manager = await run_manager.get_child(f"chain_{i}")
                chain_outputs = await chain.acall(chain_inputs, callbacks=[child_manager])
            else:
                chain_outputs = await chain.acall(chain_inputs)

            known_values.update(chain_outputs)

        if self.return_all:
            outputs = {
                key: known_values[key]
                for key in known_values
                if key not in inputs
            }
        else:
            outputs = {
                key: known_values[key]
                for key in self.output_variables
            }

        return outputs

    @property
    def _chain_type(self) -> str:
        return "sequential"

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529

from typing import Any, Dict, List, Optional, Sequence, Union
from langchain_core.agents import AgentAction, AgentFinish, AgentStep
from langchain_core.tools import BaseTool
from langchain.agents.agent import BaseSingleActionAgent, BaseMultiActionAgent

class AgentExecutor(Chain):
    """使用工具的Agent。

    AgentExecutor负责运行Agent并管理其与工具的交互。
    它实现了Agent的推理-行动循环，包括：
    1. 调用Agent进行推理
    2. 执行Agent选择的工具
    3. 将工具结果反馈给Agent
    4. 重复直到Agent决定停止

    设计特点：
    1. 灵活的Agent接口：支持单动作和多动作Agent
    2. 工具管理：统一的工具调用和错误处理
    3. 执行控制：迭代限制、时间限制、早停策略
    4. 可观测性：完整的回调和追踪支持
    """

    agent: Union[BaseSingleActionAgent, BaseMultiActionAgent, Runnable]
    """要运行的Agent，用于在执行循环的每个步骤创建计划并确定要采取的行动。"""

    tools: Sequence[BaseTool]
    """Agent可以调用的有效工具。"""

    return_intermediate_steps: bool = False
    """是否在最终输出之外还返回Agent的中间步骤轨迹。"""

    max_iterations: Optional[int] = 15
    """结束执行循环之前要采取的最大步骤数。

    设置为'None'可能导致无限循环。"""

    max_execution_time: Optional[float] = None
    """在执行循环中花费的最大墙钟时间量。"""

    early_stopping_method: str = "force"
    """如果Agent从未返回`AgentFinish`，用于早停的方法。
    可以是'force'或'generate'。

    `"force"`返回一个字符串，说明它因为遇到时间或迭代限制而停止。

    `"generate"`最后一次调用Agent的LLM Chain，
    基于之前的步骤生成最终答案。"""

    handle_parsing_errors: Union[bool, str, Callable[[OutputParserException], str]] = (
        False
    )
    """如何处理Agent输出解析器引发的错误。
    默认为`False`，会抛出错误。
    如果为`true`，错误将作为观察发送回LLM。
    如果为字符串，字符串本身将作为观察发送给LLM。
    如果为可调用函数，函数将以异常作为参数调用，
    函数的结果将作为观察传递给Agent。"""

    trim_intermediate_steps: Union[
        int,
        Callable[[List[Tuple[AgentAction, str]]], List[Tuple[AgentAction, str]]],
    ] = -1
    """如何修剪传递给Agent的中间步骤。
    如果为int，将保留最后N个步骤。
    如果为callable，将调用该函数来修剪步骤。"""

    def __init__(
        self,
        agent: Union[BaseSingleActionAgent, BaseMultiActionAgent, Runnable],
        tools: Sequence[BaseTool],
        callback_manager: Optional[BaseCallbackManager] = None,
        **kwargs: Any,
    ):
        """初始化AgentExecutor。

        Args:
            agent: 要使用的Agent。
            tools: 工具列表。
            callback_manager: 回调管理器。
            **kwargs: 额外参数。
        """
        super().__init__(**kwargs)
        self.agent = agent
        self.tools = tools

        # 验证工具名称唯一性
        tool_names = [tool.name for tool in tools]
        if len(tool_names) != len(set(tool_names)):
            raise ValueError("工具名称必须唯一")

    @property
    def input_keys(self) -> List[str]:
        """返回期望的输入键。"""
        return self.agent.input_keys

    @property
    def output_keys(self) -> List[str]:
        """返回输出键。"""
        if self.return_intermediate_steps:
            return self.agent.return_values + ["intermediate_steps"]
        else:
            return self.agent.return_values

    def _should_continue(self, iterations: int, time_elapsed: float) -> bool:
        """检查是否应该继续执行循环。

        Args:
            iterations: 当前迭代次数。
            time_elapsed: 已经过的时间。

        Returns:
            是否应该继续。
        """
        if self.max_iterations is not None and iterations >= self.max_iterations:
            return False
        if (
            self.max_execution_time is not None
            and time_elapsed >= self.max_execution_time
        ):
            return False
        return True

    def _call(
        self,
        inputs: Dict[str, str],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """运行文本并获取Agent响应。

        Args:
            inputs: 输入字典。
            run_manager: 回调管理器。

        Returns:
            包含Agent输出的字典。
        """
        # 构建工具名称到工具的映射，便于查找
        name_to_tool_map = {tool.name: tool for tool in self.tools}

        # 为日志记录构建从每个工具到颜色的映射
        color_mapping = get_color_mapping(
            [tool.name for tool in self.tools],
            excluded_colors=["green", "red"],
        )

        intermediate_steps: List[Tuple[AgentAction, str]] = []
        # 开始跟踪迭代次数和经过的时间
        iterations = 0
        time_elapsed = 0.0
        start_time = time.time()

        # 现在进入Agent循环（直到它返回某些东西）
        while self._should_continue(iterations, time_elapsed):
            next_step_output = self._take_next_step(
                name_to_tool_map,
                color_mapping,
                inputs,
                intermediate_steps,
                run_manager=run_manager,
            )

            if isinstance(next_step_output, AgentFinish):
                return self._return(
                    next_step_output,
                    intermediate_steps,
                    run_manager=run_manager,
                )

            intermediate_steps.extend(next_step_output)

            if len(next_step_output) == 1:
                next_step_action = next_step_output[0]
                # 查看工具是否应该直接返回
                tool_return = self._get_tool_return(next_step_action)
                if tool_return is not None:
                    return self._return(
                        tool_return,
                        intermediate_steps,
                        run_manager=run_manager,
                    )

            iterations += 1
            time_elapsed = time.time() - start_time

        output = self._return(
            self._get_action_agent().return_stopped_response(
                self.early_stopping_method, intermediate_steps
            ),
            intermediate_steps,
            run_manager=run_manager,
        )
        return output

    def _take_next_step(
        self,
        name_to_tool_map: Dict[str, BaseTool],
        color_mapping: Dict[str, str],
        inputs: Dict[str, str],
        intermediate_steps: List[Tuple[AgentAction, str]],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Union[AgentFinish, List[Tuple[AgentAction, str]]]:
        """采取单个步骤在Agent循环中。

        这包括：
        1. 调用Agent规划下一个动作
        2. 执行该动作
        3. 返回结果

        Args:
            name_to_tool_map: 工具名称到工具的映射。
            color_mapping: 工具到颜色的映射。
            inputs: 用户输入。
            intermediate_steps: 到目前为止的中间步骤。
            run_manager: 回调管理器。

        Returns:
            AgentFinish或步骤列表。
        """
        try:
            # 修剪中间步骤
            intermediate_steps = self._prepare_intermediate_steps(intermediate_steps)

            # 调用Agent规划
            output = self.agent.plan(
                intermediate_steps,
                callbacks=run_manager.get_child() if run_manager else None,
                **inputs,
            )
        except OutputParserException as e:
            if isinstance(self.handle_parsing_errors, bool):
                raise_error = not self.handle_parsing_errors
            else:
                raise_error = False
            if raise_error:
                raise ValueError(
                    "Agent输出解析出错。"
                    "请传递`handle_parsing_errors=True`以自动处理这些错误。"
                ) from e
            text = str(e)
            if isinstance(self.handle_parsing_errors, bool):
                if e.send_to_llm:
                    observation = str(e.observation)
                    text = str(e.llm_output)
                else:
                    observation = "解析LLM输出时出现无效格式"
            elif isinstance(self.handle_parsing_errors, str):
                observation = self.handle_parsing_errors
            elif callable(self.handle_parsing_errors):
                observation = self.handle_parsing_errors(e)
            else:
                raise ValueError("Got unexpected type of `handle_parsing_errors`")
            output = AgentAction("_Exception", observation, text)

        # 如果Agent返回AgentFinish，则完成
        if isinstance(output, AgentFinish):
            return output

        actions: List[AgentAction]
        if isinstance(output, AgentAction):
            actions = [output]
        else:
            actions = output

        result = []
        for agent_action in actions:
            if run_manager:
                run_manager.on_agent_action(agent_action, color="green")

            # 否则我们查找工具并运行它
            if agent_action.tool in name_to_tool_map:
                tool = name_to_tool_map[agent_action.tool]
                return_direct = tool.return_direct
                color = color_mapping[agent_action.tool]
                tool_run_kwargs = self.agent.tool_run_logging_kwargs()
                if return_direct:
                    tool_run_kwargs["llm_prefix"] = ""

                # 执行工具
                observation = tool.run(
                    agent_action.tool_input,
                    verbose=self.verbose,
                    color=color,
                    callbacks=run_manager.get_child() if run_manager else None,
                    **tool_run_kwargs,
                )
            else:
                tool_run_kwargs = self.agent.tool_run_logging_kwargs()
                observation = InvalidTool().run(
                    {
                        "requested_tool_name": agent_action.tool,
                        "available_tool_names": list(name_to_tool_map.keys()),
                    },
                    verbose=self.verbose,
                    color=None,
                    callbacks=run_manager.get_child() if run_manager else None,
                    **tool_run_kwargs,
                )

            result.append((agent_action, observation))

        return result

    def _prepare_intermediate_steps(
        self, steps: List[Tuple[AgentAction, str]]
    ) -> List[Tuple[AgentAction, str]]:
        """准备中间步骤以传递给Agent。

        这可能涉及修剪步骤以保持在上下文长度限制内。

        Args:
            steps: 原始中间步骤。

        Returns:
            处理后的中间步骤。
        """
        if isinstance(self.trim_intermediate_steps, int):
            if self.trim_intermediate_steps == -1:
                # 不修剪
                return steps
            else:
                # 保留最后N个步骤
                return steps[-self.trim_intermediate_steps :]
        elif callable(self.trim_intermediate_steps):
            # 使用自定义函数修剪
            return self.trim_intermediate_steps(steps)
        else:
            raise ValueError(
                f"Got unexpected type of `trim_intermediate_steps`: "
                f"{type(self.trim_intermediate_steps)}"
            )

    def _return(
        self,
        output: AgentFinish,
        intermediate_steps: List[Tuple[AgentAction, str]],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """返回Agent的输出。

        Args:
            output: Agent的最终输出。
            intermediate_steps: 中间步骤列表。
            run_manager: 回调管理器。

        Returns:
            格式化的输出字典。
        """
        if run_manager:
            run_manager.on_agent_finish(output, color="green", verbose=self.verbose)

        final_output = output.return_values
        if self.return_intermediate_steps:
            final_output["intermediate_steps"] = intermediate_steps

        return final_output

    def _get_tool_return(
        self, next_step_output: Tuple[AgentAction, str]
    ) -> Optional[AgentFinish]:
        """检查工具是否应该直接返回。

        Args:
            next_step_output: 下一步输出。

        Returns:
            如果工具应该直接返回，则返回AgentFinish，否则返回None。
        """
        agent_action, observation = next_step_output
        name_to_tool_map = {tool.name: tool for tool in self.tools}

        if agent_action.tool in name_to_tool_map:
            if name_to_tool_map[agent_action.tool].return_direct:
                return AgentFinish(
                    {self.agent.return_values[0]: observation},
                    "",
                )
        return None

    async def _acall(
        self,
        inputs: Dict[str, str],
        run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """异步运行Agent。"""
        name_to_tool_map = {tool.name: tool for tool in self.tools}
        color_mapping = get_color_mapping(
            [tool.name for tool in self.tools],
            excluded_colors=["green", "red"],
        )

        intermediate_steps: List[Tuple[AgentAction, str]] = []
        iterations = 0
        time_elapsed = 0.0
        start_time = time.time()

        while self._should_continue(iterations, time_elapsed):
            next_step_output = await self._atake_next_step(
                name_to_tool_map,
                color_mapping,
                inputs,
                intermediate_steps,
                run_manager=run_manager,
            )

            if isinstance(next_step_output, AgentFinish):
                return self._return(
                    next_step_output,
                    intermediate_steps,
                    run_manager=run_manager,
                )

            intermediate_steps.extend(next_step_output)

            if len(next_step_output) == 1:
                next_step_action = next_step_output[0]
                tool_return = self._get_tool_return(next_step_action)
                if tool_return is not None:
                    return self._return(
                        tool_return,
                        intermediate_steps,
                        run_manager=run_manager,
                    )

            iterations += 1
            time_elapsed = time.time() - start_time

        output = self._return(
            self._get_action_agent().return_stopped_response(
                self.early_stopping_method, intermediate_steps
            ),
            intermediate_steps,
            run_manager=run_manager,
        )
        return output

    async def _atake_next_step(
        self,
        name_to_tool_map: Dict[str, BaseTool],
        color_mapping: Dict[str, str],
        inputs: Dict[str, str],
        intermediate_steps: List[Tuple[AgentAction, str]],
        run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
    ) -> Union[AgentFinish, List[Tuple[AgentAction, str]]]:
        """异步采取下一步。"""
        try:
            intermediate_steps = self._prepare_intermediate_steps(intermediate_steps)

            # 调用Agent规划
            output = await self.agent.aplan(
                intermediate_steps,
                callbacks=run_manager.get_child() if run_manager else None,
                **inputs,
            )
        except OutputParserException as e:
            # 错误处理逻辑与同步版本相同
            if isinstance(self.handle_parsing_errors, bool):
                raise_error = not self.handle_parsing_errors
            else:
                raise_error = False
            if raise_error:
                raise ValueError(
                    "Agent输出解析出错。"
                    "请传递`handle_parsing_errors=True`以自动处理这些错误。"
                ) from e
            text = str(e)
            if isinstance(self.handle_parsing_errors, bool):
                if e.send_to_llm:
                    observation = str(e.observation)
                    text = str(e.llm_output)
                else:
                    observation = "解析LLM输出时出现无效格式"
            elif isinstance(self.handle_parsing_errors, str):
                observation = self.handle_parsing_errors
            elif callable(self.handle_parsing_errors):
                observation = self.handle_parsing_errors(e)
            else:
                raise ValueError("Got unexpected type of `handle_parsing_errors`")
            output = AgentAction("_Exception", observation, text)

        if isinstance(output, AgentFinish):
            return output

        actions: List[AgentAction]
        if isinstance(output, AgentAction):
            actions = [output]
        else:
            actions = output

        result = []
        for agent_action in actions:
            if run_manager:
                await run_manager.on_agent_action(agent_action, color="green")

            if agent_action.tool in name_to_tool_map:
                tool = name_to_tool_map[agent_action.tool]
                return_direct = tool.return_direct
                color = color_mapping[agent_action.tool]
                tool_run_kwargs = self.agent.tool_run_logging_kwargs()
                if return_direct:
                    tool_run_kwargs["llm_prefix"] = ""

                # 异步执行工具
                observation = await tool.arun(
                    agent_action.tool_input,
                    verbose=self.verbose,
                    color=color,
                    callbacks=run_manager.get_child() if run_manager else None,
                    **tool_run_kwargs,
                )
            else:
                tool_run_kwargs = self.agent.tool_run_logging_kwargs()
                observation = await InvalidTool().arun(
                    {
                        "requested_tool_name": agent_action.tool,
                        "available_tool_names": list(name_to_tool_map.keys()),
                    },
                    verbose=self.verbose,
                    color=None,
                    callbacks=run_manager.get_child() if run_manager else None,
                    **tool_run_kwargs,
                )

            result.append((agent_action, observation))

        return result

    @property
    def _chain_type(self) -> str:
        return "agent_executor"