DCBrain API

DCBrain 类

DCBrain 是动态规划型大脑，结合 DSL 技术与有限状态机实现稳定的长程推理能力。

Bases: BaseBrain

动态规划型的大脑，非协程安全！

此封装会结合DSL技术与动态规划技术，来实现一个动态规划型的大脑，从而实现稳定的长程推理能力。

DCBrain会结合有限状态机相关能力实现对思维链的管理与调度。

目前DCBrain有效状态如下：

• BRAIN_STATES.init 主要负责配合Memory进行记忆召回，进而填充上下文
• BRAIN_STATES.planning 主要负责调用default_llm和planning_jinja2_template进行动态规划，生成一个Plan，用于解决问题
• BRAIN_STATES.running 调用TFChainInterpreter进行Plan执行，返回结果
• BRAIN_STATES.succeeded 根据running结果判断是否成功解决，如果成功解决将会向memory提交对应的msg信息
• BRAIN_STATES.failed 一般用于遇到已知异常的时候，判断尝试标记为失败
• BRAIN_STATES.aborted 一般用于遇到未知异常的时候，判断尝试标记为中止

注意：DCBrain的状态机是有限状态机，当前仅支持单一线程运行，不能并发运行多个任务，否则会导致状态混乱。虽然目前支持相关async方法，但也主要 是为了实现对主进程的协程支持，可以释放主线程的IO阻塞，并不意味着可以多线程并发运行。若需要多线程并发运行，请使用多个DCBrain实例。

default_llm cached property

default_llm: ChatLLM

DCBrain 的基础 LLM 实例（实例级缓存）。

首次访问时按 default_llm_uri 构造，后续访问复用同一对象—— Neural register/unregister 依赖稳定 id。

plan_llm cached property

plan_llm: ChatLLM

Plan LLM 实例（实例级缓存）。优先 default_plan_llm_uri，回退 default_llm_uri。

与 default_llm / validate_llm 是三个独立缓存实例，互不共享 （TFROB-222 mutation 隔离：每次进入 planning 状态前 clear_all_prompts 重配， 单线程 FSM 下安全）。

validate_llm cached property

validate_llm: ChatLLM

Validate LLM 实例（实例级缓存）。语义同 plan_llm。

tags_scope property

tags_scope: list[str]

获取当前Brain的tags_scope，会从所有子Chain中提取tags，合并后去重

Returns:

Type	Description
list[str]	list[str]: 当前Brain的tags_scope

model_post_init

model_post_init(__context: Any) -> None

模型初始化后的处理方法，用于在模型初始化完成后，进行一些额外的操作

Parameters:

Name	Type	Description	Default
__context	Any	传入的上下文信息	required

Source code in tfrobot/brain/dc_brain.py

def model_post_init(self, __context: Any) -> None:
    """
    模型初始化后的处理方法，用于在模型初始化完成后，进行一些额外的操作

    Args:
        __context (Any): 传入的上下文信息
    """
    self._machine = Machine(
        model=self,
        states=self._states,
        transitions=self._transitions,
        initial=self._states[0],
        auto_transitions=False,
        send_event=True,
    )
    self._async_machine = AsyncMachine(
        model=self,
        states=self._states,
        transitions=self._async_transitions,
        initial=self._states[0],
        auto_transitions=False,
        send_event=True,
    )

validate_token_usage

validate_token_usage(brain_result: BrainResult) -> None

判断BrainResult中的Token使用未超过当前Brain的限制

Parameters:

Name	Type	Description	Default
brain_result	BrainResult	BrainResult	required

Source code in tfrobot/brain/dc_brain.py

def validate_token_usage(self, brain_result: BrainResult) -> None:
    """
    判断BrainResult中的Token使用未超过当前Brain的限制

    Args:
        brain_result (BrainResult): BrainResult
    """
    usage_dict: Optional[dict[ModelName, TokenUsage]] = brain_result.usage
    # 求和各个Model的TotalTokens
    total_tokens = 0
    if usage_dict:
        for model_name, usage in usage_dict.items():
            total_tokens += usage.get("total_tokens", 0)
    if total_tokens > self.max_tokens:
        raise TFTokenLimitError(
            message=f"Brain planning over max tokens({self.max_tokens}), please check the logs for more details."
        )

construct_plan_chain

construct_plan_chain(
    current_suggestion: Optional[str] = None,
) -> Chain

使用当前的default_llm配合 PlanPrompt和BrainContext 动态构建一个可以用于规划Plan的Chain

需要注意构建plan_chain的时候没有动态注册neural。因为此时应该避免工具的注入导致其plan的上下文混乱，虽说neural不直接提供工具， 但是如果一个Chain注册到Neural，在LLM执行时，会在发现没有工具可用时动态召回工具。

Parameters:

Name	Type	Description	Default
current_suggestion	Optional[str]	上一轮 validate 的驳回建议，注入模板供 LLM 参考	None

Returns:

Name	Type	Description
Chain	Chain	返回的Chain

Source code in tfrobot/brain/dc_brain.py

def construct_plan_chain(self, current_suggestion: Optional[str] = None) -> Chain:
    """
    使用当前的default_llm配合 PlanPrompt和BrainContext 动态构建一个可以用于规划Plan的Chain

    需要注意构建plan_chain的时候没有动态注册neural。因为此时应该避免工具的注入导致其plan的上下文混乱，虽说neural不直接提供工具，
    但是如果一个Chain注册到Neural，在LLM执行时，会在发现没有工具可用时动态召回工具。

    Args:
        current_suggestion (Optional[str]): 上一轮 validate 的驳回建议，注入模板供 LLM 参考

    Returns:
        Chain: 返回的Chain
    """
    # 需要构建一个validate_result的函数，函数的入参与返回标准每个分支Chain.validate_result。标准是当前Chain输出时可以使用 ```tf_plan...```进行脚本合法的脚本提取
    import re

    from tfrobot.schema.brain.chain.chain_schema import ChainResult

    def validate_chain_res(chain_res: ChainResult) -> tuple[bool, Optional[BaseMessage]]:
        """
        验证Chain的结果是否合法

        Args:
            chain_res (ChainResult): Chain的结果

        Returns:
            tuple[bool, Optional[BaseMessage]]: 是否合法，合法则返回True和None，否则返回False和错误信息
        """
        if not chain_res.content:
            return False, TextMessage(
                content="生成的内容为空，请重新生成包含tf_plan脚本的文本内容或者直接以文本形式回复。",
                creator=INTERPRETER_USER,
            )

        # 使用正则表达式检查是否包含合法的tf_plan脚本
        pattern = r"```tf_plan\n(.+?)\n```"
        match = re.search(pattern, chain_res.content, re.DOTALL)

        if not match:
            # 将回复直接作为结果设置
            chain_res.origin = {"final_result": chain_res.content}
            return True, None
        else:
            # 检查脚本内容是否为空
            plan_content = match.group(1).strip()
            if not plan_content:
                return False, TextMessage(
                    content="tf_plan脚本内容为空，请提供有效的计划内容", creator=INTERPRETER_USER
                )

            try:
                parser = Lark(PYTHON_GRAMMAR, parser="lalr", start="file_input", postlex=PythonIndenter())
                if not plan_content.endswith("\n"):
                    # 目前Lark的Python.lark语法有个天然的要求，必须以 \n 结尾
                    plan_content += "\n"
                parser.parse(plan_content)
            except Exception as e:
                return False, TextMessage(
                    content=f"tf_plan脚本内容不合法，请提供有效的计划内容。错误信息: {e}", creator=INTERPRETER_USER
                )

            # 将ChainRes设置为当前的Plan
            chain_res.origin = {"tf_plan": plan_content}

            return True, None

    # plan 历史通过 intermediate_trace 自然积累（_batch_append_to_trace），
    # 使用 reformat_input_prompt 渲染，关闭 inject_trace 避免同一数据在 intermediate_msgs 通道重复注入。

    # 获取当前 Brain.run span_id，注入模板全局变量，用于过滤顶层 trace 记录
    from opentelemetry.trace import get_current_span

    span = get_current_span()
    span_ctx = span.get_span_context()
    brain_span_id = format(span_ctx.span_id, "016x") if span_ctx.is_valid else None

    plan_trace_template = Jinja2PromptTemplate(
        templates={
            Locale.ZH: (
                "{% if intermediate_trace and intermediate_trace.records %}"
                "以下是之前的执行记录：\n\n"
                "{% for record in intermediate_trace.records %}"
                "{% set meta = (record.additional_kwargs.__trace_meta | from_json) "
                "if record.additional_kwargs and record.additional_kwargs.__trace_meta "
                "else {} %}"
                "{% if meta.parent_span_id == brain_span_id %}"
                "{% if record.role == 'assistant' %}"
                "[Plan]:\n{{ record.content }}\n\n"
                "{% else %}"
                "[执行结果]:\n{{ record.content }}\n\n"
                "{% endif %}"
                "{% endif %}"
                "{% endfor %}"
                "---\n\n"
                "{% endif %}"
                "{% if current_suggestion %}"
                "上次验证反馈：{{ current_suggestion }}\n\n"
                "---\n\n"
                "{% endif %}"
                "以下是当前需要处理的用户输入：\n\n"
                "{{ input }}"
            ),
            Locale.EN: (
                "{% if intermediate_trace and intermediate_trace.records %}"
                "Here are the previous execution records:\n\n"
                "{% for record in intermediate_trace.records %}"
                "{% set meta = (record.additional_kwargs.__trace_meta | from_json) "
                "if record.additional_kwargs and record.additional_kwargs.__trace_meta "
                "else {} %}"
                "{% if meta.parent_span_id == brain_span_id %}"
                "{% if record.role == 'assistant' %}"
                "[Plan]:\n{{ record.content }}\n\n"
                "{% else %}"
                "[Execution Result]:\n{{ record.content }}\n\n"
                "{% endif %}"
                "{% endif %}"
                "{% endfor %}"
                "---\n\n"
                "{% endif %}"
                "{% if current_suggestion %}"
                "Last validation feedback: {{ current_suggestion }}\n\n"
                "---\n\n"
                "{% endif %}"
                "The following is the current user input to process:\n\n"
                "{{ input }}"
            ),
        }
    )
    plan_trace_template.add_global_var("brain_span_id", brain_span_id)
    plan_trace_template.add_global_var("current_suggestion", current_suggestion)

    return _construct_dynamic_chain(
        default_llm=self.plan_llm,
        sys_additional_prompt=[self.planning_prompt, MemoPrompt(), KnowledgePrompt()],
        # neural=self._neural,
        max_iterations=self.max_iterations,
        max_tokens=self.max_tokens,
        validate_result=validate_chain_res,
        inject_trace=False,
        reformat_input_prompt=[UserInputPrompt(template=plan_trace_template)],
    )

prepare_init

prepare_init(event_data: TFEventData) -> None

准备init状态的处理。主要是设置当前的PlanChain

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def prepare_init(self, event_data: TFEventData) -> None:
    """
    准备init状态的处理。主要是设置当前的PlanChain

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_prepare_init async

async_prepare_init(event_data: TFEventData) -> None

准备init状态的处理。主要是设置当前的PlanChain

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_prepare_init(self, event_data: TFEventData) -> None:
    """
    准备init状态的处理。主要是设置当前的PlanChain

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

condition_init

condition_init(event_data: TFEventData) -> bool

判断是否满足init状态的条件

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Returns:

Name	Type	Description
bool	bool	是否满足init状态的条件

Source code in tfrobot/brain/dc_brain.py

def condition_init(self, event_data: TFEventData) -> bool:  # noqa
    """
    判断是否满足init状态的条件

    Args:
        event_data (TFEventData): 事件数据

    Returns:
        bool: 是否满足init状态的条件
    """
    return True

async_condition_init async

async_condition_init(event_data: TFEventData) -> bool

判断是否满足init状态的条件

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Returns:

Name	Type	Description
bool	bool	是否满足init状态的条件

Source code in tfrobot/brain/dc_brain.py

async def async_condition_init(self, event_data: TFEventData) -> bool:  # noqa
    """
    判断是否满足init状态的条件

    Args:
        event_data (TFEventData): 事件数据

    Returns:
        bool: 是否满足init状态的条件
    """
    return True

before_init

before_init(event_data: TFEventData) -> None

init状态之前的处理。主要

1. 调用Memory进行内容召回并且丰富上下文

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def before_init(self, event_data: TFEventData) -> None:
    """
    init状态之前的处理。主要

    1. 调用Memory进行内容召回并且丰富上下文

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_before_init async

async_before_init(event_data: TFEventData) -> None

init状态之前的处理。主要

1. 调用Memory进行内容召回并且丰富上下文

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_before_init(self, event_data: TFEventData) -> None:
    """
    init状态之前的处理。主要

    1. 调用Memory进行内容召回并且丰富上下文

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

on_enter_init

on_enter_init(event_data: TFEventData) -> None

1. 将当前Brain中的Chains相关设置添加到current_input.additional_info中去
2. 将当前Brian中可用Tags注入到current_input.additional_info中去

Parameters:

Name	Type	Description	Default
event_data	TFEventData	状态机过程数据	required

Source code in tfrobot/brain/dc_brain.py

def on_enter_init(self, event_data: TFEventData) -> None:
    """
    1. 将当前Brain中的Chains相关设置添加到current_input.additional_info中去
    2. 将当前Brian中可用Tags注入到current_input.additional_info中去

    Args:
        event_data (TFEventData): 状态机过程数据
    """
    brain_ctx = event_data.brain_intermediate.context
    input_key_sch = {
        "type": "object",
        "properties": {
            INPUT_KEY: {
                "type": "string",
                "description": "请使用自然语言描述欲使用此工具达到的目标。您可以结合上下文获取到的内容进行动态组装。",
                "title": "目标描述",
            }
        },
        "required": [INPUT_KEY],
    }
    chains_info = {
        c_name: {
            "abilities_purpose": c.instance_description,
            "input_schema": merge_schemas([input_key_sch, c.additional_kwargs_schema])
            if c.additional_kwargs_schema and c.additional_kwargs_schema["type"] == "object"
            else input_key_sch,
            "output_schema": c.output_json_schema,
        }
        for c_name, c in self.spec_chains.items()
    }

    if not brain_ctx.current_input.additional_kwargs:
        brain_ctx.current_input.additional_kwargs = {}
    # 设置思维链信息上下文
    cast(dict, brain_ctx.current_input.additional_kwargs)[AVAILABLE_CHAINS_KEY] = json.dumps(
        chains_info, indent=2, ensure_ascii=False
    )
    # 设置可用Tags上下文
    available_tags = self.tags_scope
    cast(dict, brain_ctx.current_input.additional_kwargs)[BRAIN_TAGS_SCOPE] = json.dumps(
        available_tags, indent=2, ensure_ascii=False
    )

on_aenter_init async

on_aenter_init(event_data: TFEventData) -> None

进入init状态时的操作（异步版本）

Parameters:

Name	Type	Description	Default
event_data	TFEventData	状态机过程数据	required

Source code in tfrobot/brain/dc_brain.py

async def on_aenter_init(self, event_data: TFEventData) -> None:
    """
    进入init状态时的操作（异步版本）

    Args:
        event_data (TFEventData): 状态机过程数据
    """
    self.on_enter_init(event_data)

after_init

after_init(event_data: TFEventData) -> None

init状态之后的处理。主要是设置当前的PlanChain

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def after_init(self, event_data: TFEventData) -> None:
    """
    init状态之后的处理。主要是设置当前的PlanChain

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_after_init async

async_after_init(event_data: TFEventData) -> None

init状态之后的处理。主要是设置当前的PlanChain

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_after_init(self, event_data: TFEventData) -> None:
    """
    init状态之后的处理。主要是设置当前的PlanChain

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

prepare_plan

prepare_plan(event_data: TFEventData) -> None

准备plan状态的处理。

在准备过程中动态判断是否有：event_data.brain_intermediate.context.current_suggestion 如果有完善建议，则需要重新将current_input与current_suggestion连接

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def prepare_plan(self, event_data: TFEventData) -> None:
    """
    准备plan状态的处理。

    在准备过程中动态判断是否有：event_data.brain_intermediate.context.current_suggestion
    如果有完善建议，则需要重新将current_input与current_suggestion连接

    Args:
        event_data (TFEventData): 事件数据
    """
    brain_ctx = event_data.brain_intermediate.context
    if current_suggestion := event_data.brain_intermediate.context.current_suggestion:
        if brain_ctx.current_input.additional_kwargs is None:
            brain_ctx.current_input.additional_kwargs = {}
        cast(dict, brain_ctx.current_input.additional_kwargs)[MSG_ADDITIONAL_REPR] = current_suggestion
    chunk_size = (
        self.memory_chunk_size if not isinstance(self.memory_chunk_size, tuple) else self.memory_chunk_size[0]
    )
    len_func = self.memory_chunk_size[1] if isinstance(self.memory_chunk_size, tuple) else cl100k_base_length
    conversation, elements, knowledge = self.memory.recall(
        current_input=brain_ctx.current_input, chunk_size=chunk_size, length_function=len_func
    )
    brain_ctx.conversation = cast(list[BaseMessage], conversation)
    brain_ctx.elements = elements
    brain_ctx.knowledge = knowledge

async_prepare_plan async

async_prepare_plan(event_data: TFEventData) -> None

准备plan状态的处理。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_prepare_plan(self, event_data: TFEventData) -> None:
    """
    准备plan状态的处理。

    Args:
        event_data (TFEventData): 事件数据
    """
    brain_ctx = event_data.brain_intermediate.context
    if current_suggestion := event_data.brain_intermediate.context.current_suggestion:
        if brain_ctx.current_input.additional_kwargs is None:
            brain_ctx.current_input.additional_kwargs = {}
        cast(dict, brain_ctx.current_input.additional_kwargs)[MSG_ADDITIONAL_REPR] = current_suggestion
    chunk_size = (
        self.memory_chunk_size if not isinstance(self.memory_chunk_size, tuple) else self.memory_chunk_size[0]
    )
    len_func = self.memory_chunk_size[1] if isinstance(self.memory_chunk_size, tuple) else cl100k_base_length
    conversation, elements, knowledge = await self.memory.async_recall(
        current_input=brain_ctx.current_input, chunk_size=chunk_size, length_function=len_func
    )
    brain_ctx.conversation = cast(list[BaseMessage], conversation)
    brain_ctx.elements = elements
    brain_ctx.knowledge = knowledge

condition_plan

condition_plan(event_data: TFEventData) -> bool

判断是否满足plan状态的条件

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Returns:

Name	Type	Description
bool	bool	是否满足plan状态的条件

Source code in tfrobot/brain/dc_brain.py

def condition_plan(self, event_data: TFEventData) -> bool:  # noqa
    """
    判断是否满足plan状态的条件

    Args:
        event_data (TFEventData): 事件数据

    Returns:
        bool: 是否满足plan状态的条件
    """
    if event_data.brain_intermediate.plan_times >= self.max_iterations:
        raise TFBrainError(
            f"Brain planning over max iterations({self.max_iterations}), please check the logs for more details.",
            category="over_max_iterations",
        )
    return True

async_condition_plan async

async_condition_plan(event_data: TFEventData) -> bool

判断是否满足plan状态的条件

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Returns:

Name	Type	Description
bool	bool	是否满足plan状态的条件

Source code in tfrobot/brain/dc_brain.py

async def async_condition_plan(self, event_data: TFEventData) -> bool:
    """
    判断是否满足plan状态的条件

    Args:
        event_data (TFEventData): 事件数据

    Returns:
        bool: 是否满足plan状态的条件
    """
    if event_data.brain_intermediate.plan_times >= self.max_iterations:
        raise TFBrainError(
            f"Brain planning over max iterations({self.max_iterations}), please check the logs for more details.",
            category="over_max_iterations",
        )
    return True

before_plan

before_plan(event_data: TFEventData) -> None

如果当前有Plan与Exec的历史，将其追加到intermediate_plans中

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def before_plan(self, event_data: TFEventData) -> None:
    """
    如果当前有Plan与Exec的历史，将其追加到intermediate_plans中

    Args:
        event_data (TFEventData): 事件数据
    """
    brain_ctx = event_data.brain_intermediate.context
    if bool(brain_ctx.current_plan) != bool(brain_ctx.current_result):
        raise TFBrainError(
            message=f"当前的计划与计划执行结果需要同时存在或同时不存在，"
            f"当前计划{bool(brain_ctx.current_plan)}，当前结果{bool(brain_ctx.current_result)}",
            category="brain_run_err",
        )
    if bool(brain_ctx.current_plan):
        brain_ctx.intermediate_plans_and_results.append(
            (
                f"```tf_plan\n{brain_ctx.current_plan}```",
                cast(ConsoleResult, brain_ctx.current_result).result,
                brain_ctx.current_suggestion,
            )
        )
        # 追加记录后，清理残留plan。因为进入plan阶段，要么生成新Plan，要么直接生成答案
        brain_ctx.current_plan = None

async_before_plan async

async_before_plan(event_data: TFEventData) -> None

如果当前有Plan与Exec的历史，将其追加到intermediate_plans中

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_before_plan(self, event_data: TFEventData) -> None:
    """
    如果当前有Plan与Exec的历史，将其追加到intermediate_plans中

    Args:
        event_data (TFEventData): 事件数据
    """
    brain_ctx = event_data.brain_intermediate.context
    if bool(brain_ctx.current_plan) != bool(brain_ctx.current_result):
        raise TFBrainError(
            message=f"当前的计划与计划执行结果需要同时存在或同时不存在，"
            f"当前计划{bool(brain_ctx.current_plan)}，当前结果{bool(brain_ctx.current_result)}",
            category="brain_run_err",
        )
    if bool(brain_ctx.current_plan):
        brain_ctx.intermediate_plans_and_results.append(
            (
                f"```tf_plan\n{brain_ctx.current_plan}```",
                cast(ConsoleResult, brain_ctx.current_result).result,
                brain_ctx.current_suggestion,
            )
        )
        # 追加记录后，清理残留plan。因为进入plan阶段，要么生成新Plan，要么直接生成答案
        brain_ctx.current_plan = None

on_enter_planning

on_enter_planning(event_data: TFEventData) -> None

进入planning状态时的处理：验证用量 + 构建上下文 + 运行 LLM 生成计划

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def on_enter_planning(self, event_data: TFEventData) -> None:
    """
    进入planning状态时的处理：验证用量 + 构建上下文 + 运行 LLM 生成计划

    Args:
        event_data (TFEventData): 事件数据
    """
    # 判断当前的用量
    brain_res = event_data.brain_intermediate.to_brain_result()
    self.validate_token_usage(brain_res)
    # plan 历史通过各链的 _batch_append_to_trace 自然积累在 current_input.intermediate_trace 中，
    # 无需手动注入。
    brain_ctx = event_data.brain_intermediate.context
    # 读取上一轮 validate 的驳回建议（清理前），注入 plan_chain 模板
    suggestion = brain_ctx.current_suggestion
    # 旧的 plan/result/suggestion 已通过 trace 保留为上下文，
    # 在此清理，避免残留数据导致 condition_exec_plan 拒绝执行新 plan，
    # 以及 condition_succeed 使用新 plan + 旧 result 的错配数据进行验证。
    brain_ctx.current_result = None
    brain_ctx.current_suggestion = None
    plan_chain = self.construct_plan_chain(current_suggestion=suggestion)
    try:
        plan_chain_result = plan_chain.run(
            current_input=brain_ctx.current_input,
            conversation=brain_ctx.conversation,
            elements=brain_ctx.elements,
            knowledge=brain_ctx.knowledge,
            # tools=brain_ctx.tools,  # 注释掉工具是避免工具的注入导致其plan的上下文混乱
        )
        # 设置当前上下文中的执行计划，或处理直接回答
        if isinstance(plan_chain_result.origin, dict):
            if plan_chain_result.origin.get("tf_plan"):
                brain_ctx.current_plan = plan_chain_result.origin["tf_plan"]
            elif plan_chain_result.origin.get("final_result"):
                # LLM 直接回答，不需要执行 Plan，直接设置最终结果
                event_data.brain_intermediate.set_final_result(plan_chain_result.origin["final_result"])
        event_data.brain_intermediate.append_res(plan_chain_result)
    except Exception as e:
        raise TFBrainError(
            message=f"Brain planning failed, please check the logs for more details. Error: {e}",
            category="brain_run_err",
        ) from e
    finally:
        event_data.brain_intermediate.plan_times += 1

on_aenter_planning async

on_aenter_planning(event_data: TFEventData) -> None

进入planning状态时的处理（异步版本）：验证用量 + 构建上下文 + 运行 LLM 生成计划

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def on_aenter_planning(self, event_data: TFEventData) -> None:
    """
    进入planning状态时的处理（异步版本）：验证用量 + 构建上下文 + 运行 LLM 生成计划

    Args:
        event_data (TFEventData): 事件数据
    """
    # 判断当前的用量
    brain_res = event_data.brain_intermediate.to_brain_result()
    self.validate_token_usage(brain_res)
    # plan 历史通过各链的 _batch_append_to_trace 自然积累在 current_input.intermediate_trace 中，
    # 无需手动注入。
    brain_ctx = event_data.brain_intermediate.context
    # 读取上一轮 validate 的驳回建议（清理前），注入 plan_chain 模板
    suggestion = brain_ctx.current_suggestion
    # 旧的 plan/result/suggestion 已通过 trace 保留为上下文，
    # 在此清理，避免残留数据导致 condition_exec_plan 拒绝执行新 plan，
    # 以及 condition_succeed 使用新 plan + 旧 result 的错配数据进行验证。
    brain_ctx.current_result = None
    brain_ctx.current_suggestion = None
    plan_chain = self.construct_plan_chain(current_suggestion=suggestion)
    try:
        plan_chain_result = await plan_chain.async_run(
            current_input=brain_ctx.current_input,
            conversation=brain_ctx.conversation,
            elements=brain_ctx.elements,
            knowledge=brain_ctx.knowledge,
            # tools=brain_ctx.tools,  # 注释掉工具是避免工具的注入导致其plan的上下文混乱
        )
        # 设置当前上下文中的执行计划，或处理直接回答
        if isinstance(plan_chain_result.origin, dict):
            if plan_chain_result.origin.get("tf_plan"):
                brain_ctx.current_plan = plan_chain_result.origin["tf_plan"]
            elif plan_chain_result.origin.get("final_result"):
                # LLM 直接回答，不需要执行 Plan，直接设置最终结果
                event_data.brain_intermediate.set_final_result(plan_chain_result.origin["final_result"])
        event_data.brain_intermediate.append_res(plan_chain_result)
    except Exception as e:
        raise TFBrainError(
            message=f"Brain planning failed, please check the logs for more details. Error: {e}",
            category="brain_run_err",
        ) from e
    finally:
        event_data.brain_intermediate.plan_times += 1

after_plan

after_plan(event_data: TFEventData) -> None

plan 转换后处理（no-op，核心逻辑已迁移至 on_enter_planning）

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def after_plan(self, event_data: TFEventData) -> None:
    """
    plan 转换后处理（no-op，核心逻辑已迁移至 on_enter_planning）

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_after_plan async

async_after_plan(event_data: TFEventData) -> None

plan 转换后处理（no-op，核心逻辑已迁移至 on_aenter_planning）

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_after_plan(self, event_data: TFEventData) -> None:
    """
    plan 转换后处理（no-op，核心逻辑已迁移至 on_aenter_planning）

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

prepare_exec_plan

prepare_exec_plan(event_data: TFEventData) -> None

准备exec_plan状态的处理。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def prepare_exec_plan(self, event_data: TFEventData) -> None:
    """
    准备exec_plan状态的处理。

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_prepare_exec_plan async

async_prepare_exec_plan(event_data: TFEventData) -> None

准备exec_plan状态的处理。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_prepare_exec_plan(self, event_data: TFEventData) -> None:
    """
    准备exec_plan状态的处理。

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

condition_exec_plan

condition_exec_plan(event_data: TFEventData) -> bool

判断是否满足exec_plan状态的条件

当前有未执行的计划

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Returns:

Name	Type	Description
bool	bool	是否满足exec_plan状态的条件

Source code in tfrobot/brain/dc_brain.py

def condition_exec_plan(self, event_data: TFEventData) -> bool:
    """
    判断是否满足exec_plan状态的条件

    当前有未执行的计划

    Args:
        event_data (TFEventData): 事件数据

    Returns:
        bool: 是否满足exec_plan状态的条件
    """
    self.validate_token_usage(event_data.brain_intermediate.to_brain_result())
    return bool(event_data.brain_intermediate.context.current_plan) and not bool(
        event_data.brain_intermediate.context.current_result
    )

async_condition_exec_plan async

async_condition_exec_plan(event_data: TFEventData) -> bool

判断是否满足exec_plan状态的条件

当前有未执行的计划

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Returns:

Name	Type	Description
bool	bool	是否满足exec_plan状态的条件

Source code in tfrobot/brain/dc_brain.py

async def async_condition_exec_plan(self, event_data: TFEventData) -> bool:
    """
    判断是否满足exec_plan状态的条件

    当前有未执行的计划

    Args:
        event_data (TFEventData): 事件数据

    Returns:
        bool: 是否满足exec_plan状态的条件
    """
    self.validate_token_usage(event_data.brain_intermediate.to_brain_result())
    return bool(event_data.brain_intermediate.context.current_plan) and not bool(
        event_data.brain_intermediate.context.current_result
    )

before_exec_plan

before_exec_plan(event_data: TFEventData) -> None

执行计划之前的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def before_exec_plan(self, event_data: TFEventData) -> None:
    """
    执行计划之前的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_before_exec_plan async

async_before_exec_plan(event_data: TFEventData) -> None

执行计划之前的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_before_exec_plan(self, event_data: TFEventData) -> None:
    """
    执行计划之前的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

on_enter_running

on_enter_running(event_data: TFEventData) -> None

进入running状态时的操作：解析并执行 Plan

使用 interpreter.evaluate() 获取 ConsoleResult，其中： - TFLLMInterrupt / TFUserInterruptError / TFUserNewInputError 会穿透 raise（不可恢复中断） - TFInterpreterError / TFChainError 等执行异常由解释器内部消化，通过 meta.success=False 标记

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def on_enter_running(self, event_data: TFEventData) -> None:
    """
    进入running状态时的操作：解析并执行 Plan

    使用 interpreter.evaluate() 获取 ConsoleResult，其中：
    - TFLLMInterrupt / TFUserInterruptError / TFUserNewInputError 会穿透 raise（不可恢复中断）
    - TFInterpreterError / TFChainError 等执行异常由解释器内部消化，通过 meta.success=False 标记

    Args:
        event_data (TFEventData): 事件数据
    """
    # 拿到当前需要执行的Plan
    current_plan = cast(str, event_data.brain_intermediate.context.current_plan)
    # 构建TFChainInterpreter与Transformer
    brain_ctx = event_data.brain_intermediate.context
    # isinstance 守卫：只有 UserMessage 才有 intermediate_trace
    trace = brain_ctx.current_input.intermediate_trace if isinstance(brain_ctx.current_input, UserMessage) else None
    interpreter = TFChainInterpreter(
        chains=self.spec_chains,
        additional_info=brain_ctx.current_input.additional_kwargs,
        input_key=INPUT_KEY,
        user=INTERPRETER_USER,
        tools=brain_ctx.tools,
        conversation=brain_ctx.conversation,
        elements=brain_ctx.elements,
        knowledge=brain_ctx.knowledge,
        strict=False,
        intermediate_trace=trace,
        msg_id=brain_ctx.current_input.msg_id,
        conversation_id=brain_ctx.current_input.conversation_id,
    )
    parser = Lark(PYTHON_GRAMMAR, parser="lalr", start="file_input", postlex=PythonIndenter())
    transformer = TFASTTransformer()
    tree = parser.parse(current_plan)
    ast = transformer.transform(tree)
    console_res: ConsoleResult | None = None
    try:
        console_res = interpreter.evaluate(ast)
        brain_ctx.current_result = console_res
    except (TFLLMInterrupt, TFUserInterruptError, TFUserNewInputError) as e:
        # 不可恢复中断，尽可能保留已有结果后 re-raise
        brain_ctx.current_result = console_res
        raise e
    finally:
        for cr in interpreter.chain_results:
            event_data.brain_intermediate.append_res(cr)
        # TFROB-240: ConsoleResult → intermediate_trace
        if console_res is not None and isinstance(brain_ctx.current_input, UserMessage):
            self._append_console_result_to_trace(brain_ctx.current_input, console_res)

on_aenter_running async

on_aenter_running(event_data: TFEventData) -> None

进入running状态时的操作（异步版本）：解析并执行 Plan

使用 interpreter.aevaluate() 获取 ConsoleResult，其中： - TFLLMInterrupt / TFUserInterruptError / TFUserNewInputError 会穿透 raise（不可恢复中断） - TFInterpreterError / TFChainError 等执行异常由解释器内部消化，通过 meta.success=False 标记

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def on_aenter_running(self, event_data: TFEventData) -> None:
    """
    进入running状态时的操作（异步版本）：解析并执行 Plan

    使用 interpreter.aevaluate() 获取 ConsoleResult，其中：
    - TFLLMInterrupt / TFUserInterruptError / TFUserNewInputError 会穿透 raise（不可恢复中断）
    - TFInterpreterError / TFChainError 等执行异常由解释器内部消化，通过 meta.success=False 标记

    Args:
        event_data (TFEventData): 事件数据
    """
    # 拿到当前需要执行的Plan
    current_plan = cast(str, event_data.brain_intermediate.context.current_plan)
    # 构建TFChainInterpreter与Transformer
    brain_ctx = event_data.brain_intermediate.context
    # isinstance 守卫：只有 UserMessage 才有 intermediate_trace
    trace = brain_ctx.current_input.intermediate_trace if isinstance(brain_ctx.current_input, UserMessage) else None
    interpreter = TFChainInterpreter(
        chains=self.spec_chains,
        additional_info=brain_ctx.current_input.additional_kwargs,
        input_key=INPUT_KEY,
        user=INTERPRETER_USER,
        tools=brain_ctx.tools,
        conversation=brain_ctx.conversation,
        elements=brain_ctx.elements,
        knowledge=brain_ctx.knowledge,
        strict=False,
        intermediate_trace=trace,
        msg_id=brain_ctx.current_input.msg_id,
        conversation_id=brain_ctx.current_input.conversation_id,
    )
    parser = Lark(PYTHON_GRAMMAR, parser="lalr", start="file_input", postlex=PythonIndenter())
    transformer = TFASTTransformer()
    tree = parser.parse(current_plan)
    ast = transformer.transform(tree)
    console_res: ConsoleResult | None = None
    try:
        console_res = await interpreter.aevaluate(ast)
        brain_ctx.current_result = console_res
    except (TFLLMInterrupt, TFUserInterruptError, TFUserNewInputError) as e:
        # 不可恢复中断，尽可能保留已有结果后 re-raise
        brain_ctx.current_result = console_res
        raise e
    finally:
        for cr in interpreter.chain_results:
            event_data.brain_intermediate.append_res(cr)
        # TFROB-240: ConsoleResult → intermediate_trace
        if console_res is not None and isinstance(brain_ctx.current_input, UserMessage):
            self._append_console_result_to_trace(brain_ctx.current_input, console_res)

after_exec_plan

after_exec_plan(event_data: TFEventData) -> None

exec_plan 转换后处理（no-op，核心逻辑已迁移至 on_enter_running）

Parameters:

Name	Type	Description	Default
event_data	TFEventData		required

Source code in tfrobot/brain/dc_brain.py

def after_exec_plan(self, event_data: TFEventData) -> None:
    """
    exec_plan 转换后处理（no-op，核心逻辑已迁移至 on_enter_running）

    Args:
        event_data (TFEventData):
    """
    ...

async_after_exec_plan async

async_after_exec_plan(event_data: TFEventData) -> None

exec_plan 转换后处理（no-op，核心逻辑已迁移至 on_aenter_running）

Parameters:

Name	Type	Description	Default
event_data	TFEventData		required

Source code in tfrobot/brain/dc_brain.py

async def async_after_exec_plan(self, event_data: TFEventData) -> None:
    """
    exec_plan 转换后处理（no-op，核心逻辑已迁移至 on_aenter_running）

    Args:
        event_data (TFEventData):
    """
    ...

construct_validate_chain

construct_validate_chain(
    plan: str,
    result: str,
    rejection_history: Optional[
        Sequence[tuple[str, str, Optional[str]]]
    ] = None,
) -> Chain

构建用户校验当前任务是否完成的思维链

Parameters:

Name	Type	Description	Default
plan	str	当前的计划	required
result	str	当前的执行结果	required
rejection_history	Optional[Sequence[tuple[str, str, Optional[str]]]]	历史驳回记录列表， 每项为 (plan, result, suggestion)，用于让 validate LLM 了解之前的尝试与驳回原因	None

Returns:

Name	Type	Description
Chain	Chain	用户校验当前任务是否完成的思维链

Source code in tfrobot/brain/dc_brain.py

def construct_validate_chain(
    self, plan: str, result: str, rejection_history: Optional[Sequence[tuple[str, str, Optional[str]]]] = None
) -> Chain:
    """
    构建用户校验当前任务是否完成的思维链

    Args:
        plan (str): 当前的计划
        result (str): 当前的执行结果
        rejection_history (Optional[Sequence[tuple[str, str, Optional[str]]]]): 历史驳回记录列表，
            每项为 (plan, result, suggestion)，用于让 validate LLM 了解之前的尝试与驳回原因

    Returns:
        Chain: 用户校验当前任务是否完成的思维链
    """
    # 转义 plan 和 result 中的大括号。因为FStr模板需要对大括号进行逃逸处理。
    plan_escaped = plan.replace("{", "{{").replace("}", "}}")
    result_escaped = result.replace("{", "{{").replace("}", "}}")
    # 构建历史驳回记录文本
    history_section_zh = ""
    history_section_en = ""
    if rejection_history:
        zh_parts = []
        en_parts = []
        for idx, (h_plan, h_result, h_suggestion) in enumerate(rejection_history, 1):
            h_plan_esc = h_plan.replace("{", "{{").replace("}", "}}")
            h_result_esc = h_result.replace("{", "{{").replace("}", "}}")
            h_suggestion_esc = (h_suggestion or "无").replace("{", "{{").replace("}", "}}")
            zh_parts.append(
                f"## 第{idx}次尝试\nPlan：\n{h_plan_esc}\n\n执行结果：{h_result_esc}\n\n驳回建议：{h_suggestion_esc}"
            )
            en_parts.append(
                f"## Attempt {idx}\nPlan:\n{h_plan_esc}\n\nExecution result: {h_result_esc}\n\n"
                f"Rejection suggestion: {h_suggestion_esc}"
            )
        history_section_zh = (
            "\n\n# 历史尝试记录\n以下是之前的尝试与驳回记录，请结合这些历史判断当前结果是否已是最佳产出：\n\n"
            + "\n\n".join(zh_parts)
            + "\n\n---\n\n"
        )
        history_section_en = (
            "\n\n# Previous Attempts\nBelow are prior attempts and rejection records. Consider this history when "
            "evaluating whether the current result is the best achievable:\n\n"
            + "\n\n".join(en_parts)
            + "\n\n---\n\n"
        )
    res_format = {
        "properties": {
            "is_solved": {"title": "Is Validated", "type": "boolean"},
            "suggestion": {"anyOf": [{"type": "string"}, {"type": "null"}], "title": "Suggestion"},
            "final_result": {
                "anyOf": [{"type": "string"}, {"type": "null"}],
                "title": "Final Result",
                "description": "A user-friendly summary derived ONLY from the execution results. "
                "Must not contain any information not present in the execution results.",
            },
        },
        "required": ["is_solved"],
        "title": "ResFormat",
        "type": "object",
    }
    res_example = [
        {
            "is_solved": False,
            "suggestion": "执行结果中缺少API调用示例。建议在Plan中增加一个步骤："
            "使用browser_reader访问文档的「快速开始」页面，提取代码示例后补充到总结中。",
        },
        {
            "is_solved": True,
            "final_result": "根据查询结果，贵公司2024年第三季度的营收为1.2亿元，同比增长15%。其中线上渠道贡献了8000万元，线下渠道贡献了4000万元。",
        },
    ]
    # 构建模板（plan 使用 tf_plan 代码块包裹，result 已是 ConsoleResult.result 格式自带 console 包裹）
    zh_template = (
        f"{history_section_zh}"
        "当前的用户输入为：\n\n---\n\n{input}\n\n---\n\n"
        f"当前的Plan为：\n```tf_plan\n{plan_escaped}\n```\n\n"
        f"当前的执行结果为：\n{result_escaped}\n\n"
        "请按要求开始分析"
    )
    en_template = (
        f"{history_section_en}"
        "The current user input is: \n\n---\n\n{input}\n\n---\n\n"
        f"The current Plan is:\n```tf_plan\n{plan_escaped}\n```\n\n"
        f"The current execution result is:\n{result_escaped}\n\n"
        "Please start analyzing as required"
    )
    reformat_prompt = UserInputPrompt(
        template=FStrPromptTemplate(templates={Locale.ZH: zh_template, Locale.EN: en_template})
    )
    # validate chain 仅做结果评估，不需要注入历史执行记录（inject_trace），
    # 否则 current_input 携带的 trace 会被展开为多余的 intermediate_msgs，污染 validate 上下文。
    # 同时关闭 enable_trace_intermediate，避免 JSON 评判结果写入 trace（suggestion 通过模板 global var 注入 plan_chain）。
    return _construct_dynamic_chain(
        default_llm=self.validate_llm,
        sys_additional_prompt=[MemoPrompt(), KnowledgePrompt(), self.validate_prompt],
        # neural=self._neural,  # 当前的运行仅作为评估，不引入Neural进行处杂逻辑处理
        max_iterations=self.max_iterations,
        max_tokens=self.max_tokens,
        response_format=res_format,
        response_example=res_example,
        inject_trace=False,
        reformat_input_prompt=[reformat_prompt],
        enable_trace_intermediate=False,
    )

prepare_succeed

prepare_succeed(event_data: TFEventData) -> None

准备succeed状态的处理。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def prepare_succeed(self, event_data: TFEventData) -> None:
    """
    准备succeed状态的处理。

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_prepare_succeed async

async_prepare_succeed(event_data: TFEventData) -> None

准备succeed状态的处理。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_prepare_succeed(self, event_data: TFEventData) -> None:
    """
    准备succeed状态的处理。

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

condition_succeed

condition_succeed(event_data: TFEventData) -> bool

判断是否可以进入成功状态

通过 current_result.meta.success 判断解释器执行是否成功， 成功后再调用 validate_chain 验证结果是否充分回应了用户问题。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	状态机过程数据	required

Returns:

Name	Type	Description
bool	bool	是否可以进入成功状态

Source code in tfrobot/brain/dc_brain.py

def condition_succeed(self, event_data: TFEventData) -> bool:
    """
    判断是否可以进入成功状态

    通过 current_result.meta.success 判断解释器执行是否成功，
    成功后再调用 validate_chain 验证结果是否充分回应了用户问题。

    Args:
        event_data (TFEventData): 状态机过程数据

    Returns:
        bool: 是否可以进入成功状态
    """
    # 直接回答场景：planning 阶段已设置 final_result，无需 validate
    if event_data.brain_intermediate.to_brain_result().final_result:
        return True
    console_result = event_data.brain_intermediate.context.current_result
    if not console_result or not console_result.meta.success:
        return False
    current_plan = cast(str, event_data.brain_intermediate.context.current_plan)
    # LLM 视图：带 ```console``` fence 和 DONE/SUCCESS 元数据，供 validate chain 判断执行成败
    validate_input_text = console_result.result
    # 用户视图：纯 markdown 文本，供 set_final_result 兜底作为最终 assistant 回复（TFROB-217）。
    # 极端情况下脚本无 print 无 return，user_content 为 None，此时兜底到 .result（LLM 视图），
    # 让用户至少看到 fenced 的 DONE/SUCCESS 元数据，而不是空白。
    user_facing_text = console_result.user_content or console_result.result or ""
    # 从当前 run 的 intermediate_plans_and_results 获取历史 plan-result-suggestion，注入 validate chain
    rejection_history = list(event_data.brain_intermediate.context.intermediate_plans_and_results)
    validate_chain = self.construct_validate_chain(current_plan, validate_input_text, rejection_history)
    validate_res = validate_chain.run(
        current_input=event_data.brain_intermediate.context.current_input,
        conversation=event_data.brain_intermediate.context.conversation,
        elements=event_data.brain_intermediate.context.elements,
        knowledge=event_data.brain_intermediate.context.knowledge,
    )
    event_data.brain_intermediate.append_res(validate_res)
    if isinstance(validate_res.origin, dict):
        if validate_res.origin.get("is_solved"):
            final_result = validate_res.origin.get("final_result")
            event_data.brain_intermediate.set_final_result(final_result if final_result else user_facing_text)
            return True
        else:
            event_data.brain_intermediate.context.current_suggestion = validate_res.origin.get("suggestion")
            # 兜底：连续驳回达到上限时自动放行，避免因客观限制导致的无限循环
            # rejection_history 只含前 N-1 次，当前是第 N 次驳回，故 +1
            if 0 < self.max_validate_rejections <= len(rejection_history) + 1:
                event_data.brain_intermediate.set_final_result(user_facing_text)
                return True
            return False
    else:
        raise TFBrainError(message="验证结果不合法，请检查日志", category="brain_run_err")

async_condition_succeed async

async_condition_succeed(event_data: TFEventData) -> bool

判断是否可以进入成功状态

通过 current_result.meta.success 判断解释器执行是否成功， 成功后再调用 validate_chain 验证结果是否充分回应了用户问题。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	状态机过程数据	required

Returns:

Name	Type	Description
bool	bool	是否可以进入成功状态

Source code in tfrobot/brain/dc_brain.py

async def async_condition_succeed(self, event_data: TFEventData) -> bool:
    """
    判断是否可以进入成功状态

    通过 current_result.meta.success 判断解释器执行是否成功，
    成功后再调用 validate_chain 验证结果是否充分回应了用户问题。

    Args:
        event_data (TFEventData): 状态机过程数据

    Returns:
        bool: 是否可以进入成功状态
    """
    # 直接回答场景：planning 阶段已设置 final_result，无需 validate
    if event_data.brain_intermediate.to_brain_result().final_result:
        return True
    console_result = event_data.brain_intermediate.context.current_result
    if not console_result or not console_result.meta.success:
        return False
    current_plan = cast(str, event_data.brain_intermediate.context.current_plan)
    # LLM 视图：带 ```console``` fence 和 DONE/SUCCESS 元数据，供 validate chain 判断执行成败
    validate_input_text = console_result.result
    # 用户视图：纯 markdown 文本，供 set_final_result 兜底作为最终 assistant 回复（TFROB-217）。
    # 极端情况下脚本无 print 无 return，user_content 为 None，此时兜底到 .result（LLM 视图），
    # 让用户至少看到 fenced 的 DONE/SUCCESS 元数据，而不是空白。
    user_facing_text = console_result.user_content or console_result.result or ""
    # 从当前 run 的 intermediate_plans_and_results 获取历史 plan-result-suggestion，注入 validate chain
    rejection_history = list(event_data.brain_intermediate.context.intermediate_plans_and_results)
    validate_chain = self.construct_validate_chain(current_plan, validate_input_text, rejection_history)
    validate_res = await validate_chain.async_run(
        current_input=event_data.brain_intermediate.context.current_input,
        conversation=event_data.brain_intermediate.context.conversation,
        elements=event_data.brain_intermediate.context.elements,
        knowledge=event_data.brain_intermediate.context.knowledge,
    )
    event_data.brain_intermediate.append_res(validate_res)
    if isinstance(validate_res.origin, dict):
        if validate_res.origin.get("is_solved"):
            final_result = validate_res.origin.get("final_result")
            event_data.brain_intermediate.set_final_result(final_result if final_result else user_facing_text)
            return True
        else:
            event_data.brain_intermediate.context.current_suggestion = validate_res.origin.get("suggestion")
            # 兜底：连续驳回达到上限时自动放行，避免因客观限制导致的无限循环
            # rejection_history 只含前 N-1 次，当前是第 N 次驳回，故 +1
            if 0 < self.max_validate_rejections <= len(rejection_history) + 1:
                event_data.brain_intermediate.set_final_result(user_facing_text)
                return True
            return False
    else:
        raise TFBrainError(message="验证结果不合法，请检查日志", category="brain_run_err")

before_succeed

before_succeed(event_data: TFEventData) -> None

成功状态之前的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def before_succeed(self, event_data: TFEventData) -> None:
    """
    成功状态之前的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_before_succeed async

async_before_succeed(event_data: TFEventData) -> None

成功状态之前的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_before_succeed(self, event_data: TFEventData) -> None:
    """
    成功状态之前的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

on_enter_succeeded

on_enter_succeeded(event_data: TFEventData) -> None

进入succeeded状态时的操作

1. 如果当前会话有缓存未完成的Plan与Result，进行清理

注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	状态机过程数据	required

Source code in tfrobot/brain/dc_brain.py

def on_enter_succeeded(self, event_data: TFEventData) -> None:
    """
    进入succeeded状态时的操作

    1. 如果当前会话有缓存未完成的Plan与Result，进行清理

    注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

    Args:
        event_data (TFEventData): 状态机过程数据
    """
    self._prepare_succeeded_messages(event_data)

on_aenter_succeeded async

on_aenter_succeeded(event_data: TFEventData) -> None

进入succeeded状态时的操作（异步版本）

注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	状态机过程数据	required

Source code in tfrobot/brain/dc_brain.py

async def on_aenter_succeeded(self, event_data: TFEventData) -> None:
    """
    进入succeeded状态时的操作（异步版本）

    注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

    Args:
        event_data (TFEventData): 状态机过程数据
    """
    self._prepare_succeeded_messages(event_data)

after_succeed

after_succeed(event_data: TFEventData) -> None

成功状态之后的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def after_succeed(self, event_data: TFEventData) -> None:
    """
    成功状态之后的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_after_succeed async

async_after_succeed(event_data: TFEventData) -> None

成功状态之后的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_after_succeed(self, event_data: TFEventData) -> None:
    """
    成功状态之后的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

prepare_fail

prepare_fail(event_data: TFEventData) -> None

准备failed状态的处理。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def prepare_fail(self, event_data: TFEventData) -> None:
    """
    准备failed状态的处理。

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_prepare_fail async

async_prepare_fail(event_data: TFEventData) -> None

准备failed状态的处理。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_prepare_fail(self, event_data: TFEventData) -> None:
    """
    准备failed状态的处理。

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

condition_fail

condition_fail(event_data: TFEventData) -> bool

判断是否满足失败状态的条件。Fail状态会生成一条消息记录到当前的conversation中。而Abort不会。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Returns:

Name	Type	Description
bool	bool	是否满足失败状态的条件

Source code in tfrobot/brain/dc_brain.py

def condition_fail(self, event_data: TFEventData) -> bool:
    """
    判断是否满足失败状态的条件。Fail状态会生成一条消息记录到当前的conversation中。而Abort不会。

    Args:
        event_data (TFEventData): 事件数据

    Returns:
        bool: 是否满足失败状态的条件
    """
    err = cast(Optional[Exception], event_data.kwargs.get("error"))
    return bool(err)

async_condition_fail async

async_condition_fail(event_data: TFEventData) -> bool

判断是否满足失败状态的条件。Fail状态会生成一条消息记录到当前的conversation中。而Abort不会。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Returns:

Name	Type	Description
bool	bool	是否满足失败状态的条件

Source code in tfrobot/brain/dc_brain.py

async def async_condition_fail(self, event_data: TFEventData) -> bool:
    """
    判断是否满足失败状态的条件。Fail状态会生成一条消息记录到当前的conversation中。而Abort不会。

    Args:
        event_data (TFEventData): 事件数据

    Returns:
        bool: 是否满足失败状态的条件
    """
    err = cast(Optional[Exception], event_data.kwargs.get("error"))
    return bool(err)

before_fail

before_fail(event_data: TFEventData) -> None

失败状态之前的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def before_fail(self, event_data: TFEventData) -> None:
    """
    失败状态之前的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_before_fail async

async_before_fail(event_data: TFEventData) -> None

失败状态之前的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_before_fail(self, event_data: TFEventData) -> None:
    """
    失败状态之前的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

on_enter_failed

on_enter_failed(event_data: TFEventData) -> None

进入失败状态时的操作

1. 生成一条错误消息用于提示用户

注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	状态机过程数据	required

Source code in tfrobot/brain/dc_brain.py

def on_enter_failed(self, event_data: TFEventData) -> None:
    """
    进入失败状态时的操作

    1. 生成一条错误消息用于提示用户

    注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

    Args:
        event_data (TFEventData): 状态机过程数据
    """
    self._prepare_failed_messages(event_data, is_async=False)

on_aenter_failed async

on_aenter_failed(event_data: TFEventData) -> None

进入失败状态时的操作（异步版本）

1. 生成一条错误消息用于提示用户

注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	状态机过程数据	required

Source code in tfrobot/brain/dc_brain.py

async def on_aenter_failed(self, event_data: TFEventData) -> None:
    """
    进入失败状态时的操作（异步版本）

    1. 生成一条错误消息用于提示用户

    注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

    Args:
        event_data (TFEventData): 状态机过程数据
    """
    self._prepare_failed_messages(event_data, is_async=True)

after_fail

after_fail(event_data: TFEventData) -> None

失败状态之后的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def after_fail(self, event_data: TFEventData) -> None:
    """
    失败状态之后的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_after_fail async

async_after_fail(event_data: TFEventData) -> None

失败状态之后的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_after_fail(self, event_data: TFEventData) -> None:
    """
    失败状态之后的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

prepare_abort

prepare_abort(event_data: TFEventData) -> None

准备abort状态的处理。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def prepare_abort(self, event_data: TFEventData) -> None:
    """
    准备abort状态的处理。

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_prepare_abort async

async_prepare_abort(event_data: TFEventData) -> None

准备abort状态的处理。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_prepare_abort(self, event_data: TFEventData) -> None:
    """
    准备abort状态的处理。

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

condition_abort

condition_abort(event_data: TFEventData) -> bool

判断是否满足abort状态的条件

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Returns:

Name	Type	Description
bool	bool	是否满足abort状态的条件

Source code in tfrobot/brain/dc_brain.py

def condition_abort(self, event_data: TFEventData) -> bool:
    """
    判断是否满足abort状态的条件

    Args:
        event_data (TFEventData): 事件数据

    Returns:
        bool: 是否满足abort状态的条件
    """
    err = cast(Optional[Exception], event_data.kwargs.get("error"))
    return bool(err)

async_condition_abort async

async_condition_abort(event_data: TFEventData) -> bool

判断是否满足abort状态的条件

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Returns:

Name	Type	Description
bool	bool	是否满足abort状态的条件

Source code in tfrobot/brain/dc_brain.py

async def async_condition_abort(self, event_data: TFEventData) -> bool:
    """
    判断是否满足abort状态的条件

    Args:
        event_data (TFEventData): 事件数据

    Returns:
        bool: 是否满足abort状态的条件
    """
    err = cast(Optional[Exception], event_data.kwargs.get("error"))
    return bool(err)

before_abort

before_abort(event_data: TFEventData) -> None

abort状态之前的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def before_abort(self, event_data: TFEventData) -> None:
    """
    abort状态之前的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_before_abort async

async_before_abort(event_data: TFEventData) -> None

abort状态之前的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_before_abort(self, event_data: TFEventData) -> None:
    """
    abort状态之前的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

on_enter_aborted

on_enter_aborted(event_data: TFEventData) -> None

进入abort状态时的操作

注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	状态机过程数据	required

Source code in tfrobot/brain/dc_brain.py

def on_enter_aborted(self, event_data: TFEventData) -> None:
    """
    进入abort状态时的操作

    注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

    Args:
        event_data (TFEventData): 状态机过程数据
    """
    self._prepare_aborted_data(event_data)

on_aenter_aborted async

on_aenter_aborted(event_data: TFEventData) -> None

进入abort状态时的操作（异步版本）

注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

Parameters:

Name	Type	Description	Default
event_data	TFEventData	状态机过程数据	required

Source code in tfrobot/brain/dc_brain.py

async def on_aenter_aborted(self, event_data: TFEventData) -> None:
    """
    进入abort状态时的操作（异步版本）

    注意：消息持久化（commit）由 run() / async_run() 统一负责，此处不重复提交。

    Args:
        event_data (TFEventData): 状态机过程数据
    """
    self._prepare_aborted_data(event_data)

after_abort

after_abort(event_data: TFEventData) -> None

abort状态之后的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

def after_abort(self, event_data: TFEventData) -> None:
    """
    abort状态之后的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

async_after_abort async

async_after_abort(event_data: TFEventData) -> None

abort状态之后的处理

Parameters:

Name	Type	Description	Default
event_data	TFEventData	事件数据	required

Source code in tfrobot/brain/dc_brain.py

async def async_after_abort(self, event_data: TFEventData) -> None:
    """
    abort状态之后的处理

    Args:
        event_data (TFEventData): 事件数据
    """
    ...

commit_message

commit_message(msg: UserAndAssMsg) -> None

提交消息

Parameters:

Name	Type	Description	Default
msg	UserAndAssMsg	消息	required

Source code in tfrobot/brain/dc_brain.py

def commit_message(self, msg: UserAndAssMsg) -> None:
    """
    提交消息

    Args:
        msg (UserAndAssMsg): 消息
    """
    self.memory.commit(msg)

async_commit_message async

async_commit_message(msg: UserAndAssMsg) -> None

提交消息

Parameters:

Name	Type	Description	Default
msg	UserAndAssMsg	消息	required

Source code in tfrobot/brain/dc_brain.py

async def async_commit_message(self, msg: UserAndAssMsg) -> None:
    """
    提交消息

    Args:
        msg (UserAndAssMsg): 消息
    """
    await self.memory.acommit(msg)

connect_to_neural

connect_to_neural(neural: Neural) -> None

实现NeuralProtocol协议，向Neural注册自己

Parameters:

Name	Type	Description	Default
neural	Neural	Neural实例	required

Source code in tfrobot/brain/dc_brain.py

def connect_to_neural(self, neural: Neural) -> None:
    """
    实现NeuralProtocol协议，向Neural注册自己

    Args:
        neural(Neural): Neural实例
    """
    if self._neural and self._neural is not neural:
        raise ValueError("The neural is already registered by another neural.")
    self._neural = neural
    for chain in self.spec_chains.values():
        self._neural.register(chain)
    self._neural.register(self.default_llm)
    self._neural.register(self.memory)

disconnect_from_neural

disconnect_from_neural(neural: Neural) -> None

实现NeuralProtocol协议，从Neural注销自己

Parameters:

Name	Type	Description	Default
neural	Neural	Neural实例	required

Source code in tfrobot/brain/dc_brain.py

def disconnect_from_neural(self, neural: Neural) -> None:
    """
    实现NeuralProtocol协议，从Neural注销自己

    Args:
        neural(Neural): Neural实例
    """
    if self._neural is not neural:  # pragma: no cover
        raise ValueError("The neural is not the same as the registered neural.")  # pragma: no cover
    for chain in self.spec_chains.values():
        self._neural.unregister(chain)
    self._neural.unregister(self.default_llm)
    self._neural.unregister(self.memory)
    self._neural = None  # pragma: no cover

辅助函数

构造新的Chain的全局私有函数。

此函数会清空default_llm的所有prompt并重新设置，因此在default_llm中配置prompt是没有意义的。

注意：Chain仅支持纯文本或字典类型的JsonSchema返回。如需返回list/int/float等类型， 请使用字典包装，如：{"result": [1, 2, 3]} 而不是直接 [1, 2, 3]。

如果提供了response_example，本函数在构建Chain的时候会自动向SystemPrompt中添加JSON示例相关Prompt。因此调用方准备Prompt可以不用刻意准备JSON生成引导了。

Parameters:

Name	Type	Description	Default
default_llm	ChatLLM	Chain使用的基础LLM	required
sys_additional_prompt	AdditionalInfoPrompt | list[AdditionalInfoPrompt] | None	动态构建的SysPrompt，用于动态构建Chain， 强调当前Chain的目标与上下文环境	None
before_input_prompt	AdditionalInfoPrompt | list[AdditionalInfoPrompt] | None	动态构建的SysPrompt，用于动态构建Chain， 强调当前Chain的目标与上下文环境	None
neural	Optional[Neural]	Neural实例，如果构建处有身处于Neural环境中，尽量传递此参数	None
max_iterations	int	Chain的最大迭代次数。默认为10	10
max_tokens	int	Chain的最大token数。默认为128,000	128000
response_format	Optional[Union[dict, LLMResponseFormat]]	Chain的响应格式。一个符合JsonSchema的字典	None
response_example	Optional[Union[dict, LLMResponseFormat]]	Chain的响应示例。一个符合JsonSchema的字典	None
inject_trace	bool | None	设置LLM的inject_trace。None表示不修改（保持LLM默认值）	None
reformat_input_prompt	list[BasePrompt] | None	设置LLM的reformat_input_prompt	None
enable_trace_intermediate	bool | None	设置Chain的enable_trace_intermediate。None表示不修改（保持默认值）	None
**kwargs	Any	传递给Chain构造函数的额外参数	{}

Returns:

Name	Type	Description
Chain	Chain	构造的Chain实例。Chain中的llm仅有可能包括system_msg_prompt，其它的prompt均已清空，可以自行继续设定。

Source code in tfrobot/brain/dc_brain.py

def _construct_dynamic_chain(
    default_llm: ChatLLM,
    sys_additional_prompt: BasePrompt | list[BasePrompt] | None = None,
    before_input_prompt: BasePrompt | list[BasePrompt] | None = None,
    neural: Optional[Neural] = None,
    max_iterations: int = 10,
    max_tokens: int = 128_000,
    response_format: dict | None = None,
    response_example: dict | list[dict] | None = None,
    inject_trace: bool | None = None,
    reformat_input_prompt: list[BasePrompt] | None = None,
    enable_trace_intermediate: bool | None = None,
    **kwargs: Any,
) -> Chain:
    """
    构造新的Chain的全局私有函数。

    此函数会清空default_llm的所有prompt并重新设置，因此在default_llm中配置prompt是没有意义的。

    注意：Chain仅支持纯文本或字典类型的JsonSchema返回。如需返回list/int/float等类型，
    请使用字典包装，如：{"result": [1, 2, 3]} 而不是直接 [1, 2, 3]。

    如果提供了response_example，本函数在构建Chain的时候会自动向SystemPrompt中添加JSON示例相关Prompt。因此调用方准备Prompt可以不用刻意准备JSON生成引导了。

    Args:
        default_llm (ChatLLM): Chain使用的基础LLM
        sys_additional_prompt (AdditionalInfoPrompt | list[AdditionalInfoPrompt] | None): 动态构建的SysPrompt，用于动态构建Chain，
            强调当前Chain的目标与上下文环境
        before_input_prompt (AdditionalInfoPrompt | list[AdditionalInfoPrompt] | None): 动态构建的SysPrompt，用于动态构建Chain，
            强调当前Chain的目标与上下文环境
        neural (Optional[Neural]): Neural实例，如果构建处有身处于Neural环境中，尽量传递此参数
        max_iterations (int): Chain的最大迭代次数。默认为10
        max_tokens (int): Chain的最大token数。默认为128,000
        response_format (Optional[Union[dict, LLMResponseFormat]]): Chain的响应格式。一个符合JsonSchema的字典
        response_example (Optional[Union[dict, LLMResponseFormat]]): Chain的响应示例。一个符合JsonSchema的字典
        inject_trace (bool | None): 设置LLM的inject_trace。None表示不修改（保持LLM默认值）
        reformat_input_prompt (list[BasePrompt] | None): 设置LLM的reformat_input_prompt
        enable_trace_intermediate (bool | None): 设置Chain的enable_trace_intermediate。None表示不修改（保持默认值）
        **kwargs: 传递给Chain构造函数的额外参数

    Returns:
        Chain: 构造的Chain实例。Chain中的llm仅有可能包括system_msg_prompt，其它的prompt均已清空，可以自行继续设定。
    """
    # 清空所有现有的prompts
    default_llm.clear_all_prompts()

    # 设置planning prompt作为system message
    if sys_additional_prompt:
        if isinstance(sys_additional_prompt, BasePrompt):
            default_llm.system_msg_prompt.append(sys_additional_prompt)
        elif isinstance(sys_additional_prompt, list):
            default_llm.system_msg_prompt.extend(sys_additional_prompt)
        else:
            raise TypeError("sys_additional_prompt must be BasePrompt or list[BasePrompt] or None")
    if before_input_prompt:
        if isinstance(before_input_prompt, BasePrompt):
            default_llm.before_input_msg_prompt.append(before_input_prompt)
        elif isinstance(before_input_prompt, list):
            default_llm.before_input_msg_prompt.extend(before_input_prompt)
        else:
            raise TypeError("before_input_prompt must be BasePrompt or list[BasePrompt] or None")

    # 创建chain配置
    chain_config: dict = {"llm": default_llm, "max_iterations": max_iterations, "max_tokens": max_tokens}

    # 添加response format如果提供了
    if response_format is not None and response_format.get("type") == "object":
        # 如果是一个要求返回string的response format
        chain_config["response_format"] = {
            "type": "json_schema",
            "json_schema": {"name": "DynamicChainResultSchema", "schema": response_format, "strict": False},
            "examples": response_example,
        }
        example_prompt = AdditionalInfoPrompt(
            template=Jinja2PromptTemplate(
                templates={
                    Locale.ZH: "请以JSON结构返回结果，示例如下:\n{{additional_info.__tf_llm_json_result_examples__}}",
                    Locale.EN: "Please return the result in JSON structure, as shown below:\n"
                    "{{additional_info.__tf_llm_json_result_examples__}}",
                },
                params_schema={
                    "properties": {
                        "additional_info": {
                            "title": "Additional Info",
                            "type": "object",
                            "properties": {
                                "__tf_llm_json_result_examples__": {
                                    "type": "string",
                                    "description": "示例返回结果",
                                    "title": "Example Result",
                                }
                            },
                            "required": ["__tf_llm_json_result_examples__"],
                        }
                    },
                    "required": ["additional_info"],
                    "title": "A",
                    "type": "object",
                },
            )
        )
        # 保证可以在System中引导输出JSON
        default_llm.system_msg_prompt.append(example_prompt)

    # 设置 LLM 的 inject_trace 和 reformat_input_prompt（在构造 Chain 之前完成，避免调用方后置访问 chain.llm）
    if inject_trace is not None:
        default_llm.inject_trace = inject_trace
    if reformat_input_prompt is not None:
        default_llm.reformat_input_prompt = reformat_input_prompt

    # 设置 Chain 的 enable_trace_intermediate
    if enable_trace_intermediate is not None:
        chain_config["enable_trace_intermediate"] = enable_trace_intermediate

    # 添加任何额外的kwargs
    chain_config.update(kwargs)

    # 构造chain
    chain = Chain.model_validate(chain_config)

    # 如果提供了neural，确保chain连接到它
    if neural:
        neural.register(chain)

    return (
        chain  # 这里需要注意，利益于Neural是使用WeakSet管理的组件，因此不需要显式调用unregister，也可以完成内存释放。
    )