evofabric.core.typing

evofabric.core.typing#

General#

MISSING = object(): 空参数描述符。

Messages#

class evofabric.core.typing.ChatUsage(BaseModel)[源代码]#

定义 LLM 聊天客户端的使用信息。

参数:

completion_tokens (Optional[int]) -- 生成补全内容所使用的 token 数。
prompt_tokens (Optional[int]) -- 提示词中使用的 token 数。
total_tokens (Optional[int]) -- 请求中使用的总 token 数（提示词 + 补全内容）。
generation_time (Optional[float]) -- 生成所耗时间（秒）。

class evofabric.core.typing.EmbedUsage(BaseModel)[源代码]#

定义嵌入（embedding）的使用信息。

参数:: generation_time (int) -- 生成所耗时间（秒）。

class evofabric.core.typing.Reranking(BaseModel)#

定义重排序（reranking）的使用信息。

参数:: generation_time (int) -- 生成所耗时间（秒）。

class evofabric.core.typing.Function(BaseModel)[源代码]#

定义函数调用的信息。

参数:

arguments (str) -- JSON 格式的函数参数。
name (str) -- 函数名称。

class evofabric.core.typing.ToolCall(BaseModel)[源代码]#

定义工具调用信息。

参数:

id (str) -- 函数调用的唯一标识符。
function (Function) -- 模型调用的函数对象。
type (Literal['function']) -- 调用类型。

class evofabric.core.typing.ChatStreamChunk(BaseModel)[源代码]#

定义 LLM 聊天客户端的流式分块内容。

参数:

reasoning_content (Optional[str]) -- 推理内容。
content (Optional[str]) -- 输出内容。

class evofabric.core.typing.LLMChatResponse(BaseModel)[源代码]#

定义 LLM 客户端的响应格式。

参数:

content (str) -- 响应内容。
tool_calls (Optional[List[ToolCall]]) -- 工具调用列表。
reasoning_content (Optional[str]) -- 响应的推理内容。
usage (Optional[ChatUsage]) -- 使用信息。
id (str) -- 响应的唯一标识符。
meta (dict) -- 元信息。

class evofabric.core.typing.EmbedResponse(BaseModel)[源代码]#

定义嵌入客户端的响应格式。

参数:

embeddings (List[float]) -- 嵌入向量（浮点数列表）。
usage (Optional[EmbedUsage]) -- 使用信息。

class evofabric.core.typing.RerankResponse(BaseModel)[源代码]#

定义重排序客户端的响应格式。

参数:

scores (List[float]) -- 重排序得分。
texts (List[str]) -- 重排序文本。
usage (Optional[RerankUsage]) -- 使用信息。

class evofabric.core.typing.StateBaseMessage(BaseModel)[源代码]#

表示状态消息的基类。

参数:

content (Any) -- 消息内容。
msg_id (Optional[str]) -- 消息唯一 ID（由 append_message 策略自动添加）。
node_name (Optional[str]) -- 消息由哪个Node发出（由 GraphNodeSpec 在运行时注入）。

class evofabric.core.typing.SystemMessage(StateBaseMessage)[源代码]#

系统消息。

继承自 StateBaseMessage

参数:

content (Any) -- 消息内容。
msg_id (Optional[str]) -- 消息唯一 ID。
role (Literal['system']) -- 消息角色，固定为 system。

class evofabric.core.typing.UserMessage(StateBaseMessage)[源代码]#

用户消息。

继承自 StateBaseMessage

参数:

content (Any) -- 消息内容。
msg_id (Optional[str]) -- 消息唯一 ID。
role (Literal['user']) -- 消息角色，固定为 user。

class evofabric.core.typing.AssistantMessage(StateBaseMessage)[源代码]#

助手消息。

继承自 StateBaseMessage

参数:

content (Any) -- 消息内容。
msg_id (Optional[str]) -- 消息唯一 ID。
role (Literal['assistant']) -- 消息角色，固定为 assistant。
reasoning_content (Optional[str]) -- 推理内容。
tool_calls (Optional[List[ToolCall]]) -- 工具调用列表。
usage (Optional[ChatUsage]) -- 使用信息。

class evofabric.core.typing.ToolMessage(StateBaseMessage)[源代码]#

工具消息。

继承自 StateBaseMessage

参数:

content (Any) -- 消息内容。
msg_id (Optional[str]) -- 消息唯一 ID。
tool_call_id (str) -- 工具调用 ID。
role (Literal['tool']) -- 消息角色，固定为 tool。

class evofabric.core.typing.ToolCallResult(BaseModel)[源代码]#

定义工具调用的执行结果。

参数:

tool_call_id (str) -- 工具调用 ID。
success (bool) -- 工具是否成功返回结果。
content (Any) -- 工具返回的具体信息。若成功，为预期结果；若失败，为错误简述。
traceback (str) -- 完整错误回溯。

evofabric.core.typing.cast_state_message(msg) → StateMessage[源代码]#

将输入消息转换为验证后的 StateMessage 实例。

返回:: 一个经过验证的 StateMessage 实例。
返回类型:: StateMessage

class evofabric.core.typing.StateMessage#

聊天消息类型别名，表示对话历史中可能出现的任意消息类型。包括 UserMessage、ToolMessage、AssistantMessage、SystemMessage 或 StateBaseMessage。

定义为：

StateMessage = Union[UserMessage, ToolMessage, AssistantMessage, SystemMessage, StateBaseMessage]

Tool#

class evofabric.core.typing.CodeExecDockerConfig(BaseModel)[源代码]#

代码沙盒初始化配置类。

参数:

image (str) -- Docker 镜像名称，默认为 "python:3-slim"
auto_remove (bool) -- 容器运行结束后是否自动删除，默认为 True
working_dir (str) -- 容器内的工作目录，默认为 "/tmp"
tty (bool) -- 是否分配伪终端，默认为 True
detach (bool) -- 是否在后台运行容器，默认为 True
mem_limit (str) -- 内存限制，默认为 "4096m"
cpu_quota (int) -- CPU 配额，默认为 50000
entrypoint (str) -- 容器入口点，默认为 "/bin/sh"
command (Union[str, List[str]]) -- 容器启动时执行的命令，默认为 None
name (str) -- 容器名称，默认为 "evofabric_sandbox"
network (str) -- 容器使用的网络模式，默认为 "host"
volumes (dict) -- 挂载卷映射，默认为 None

class evofabric.core.typing.PromptRequest(BaseModel)[源代码]#

PromptRequest 用于定义提示请求。

参数:

server_name (str) -- 服务器名称。
prompt_name (str) -- 提示模板名称。
arguments (Dict[str, str]) -- 参数字典。

class evofabric.core.typing.ResourceRequest(BaseModel)[源代码]#

ResourceRequest 用于定义资源请求。

参数:

server_name (str) -- 服务器名称。
url (str) -- 资源 URL 地址。

class evofabric.core.typing.StdioLink(StdioServerParameters)[源代码]#

定义 MCP 服务器的标准输入输出（Stdio）链接类型配置。

继承自 StdioServerParameters

参数:

type (Literal["StdioLink"]) -- 类型标识符，固定为 "StdioLink"。
read_time_out (float) -- 读取超时时间，默认为 10.0 秒。
command (str) -- 启动服务器的执行方式（继承自父类）。
args (List[str]) -- 启动服务器的指令参数（继承自父类）。

class evofabric.core.typing.SseLink(BaseModel)[源代码]#

定义 MCP 服务器的 SSE（Server-Sent Events）链接类型配置。

参数:

type (Literal["SseLink"]) -- 类型标识符，固定为 "SseLink"。
url (str) -- SSE 服务器地址。
headers (Dict[str, Any]) -- 请求头，默认为 None。
timeout (float) -- 请求超时时间，默认为 30.0 秒。
sse_read_timeout (float) -- SSE 流读取超时时间，默认为 300.0 秒。

class evofabric.core.typing.StreamableHttpLink(BaseModel)[源代码]#

定义 MCP 服务器的可流式传输 HTTP 链接配置。

参数:

type (Literal["StreamableHttpLink"]) -- 类型标识符，固定为 "StreamableHttpLink"。
url (str) -- HTTP 服务器地址。
headers (Dict[str, Any]) -- 请求头，默认为 None。
timeout (float) -- 请求超时时间，默认为 30.0 秒。
sse_read_timeout (float) -- SSE 流读取超时时间，默认为 300.0 秒。
terminate_on_close (bool) -- 连接关闭时是否终止，默认为 True。

class evofabric.core.typing.MCPConfig(BaseModel)[源代码]#

MCPConfig 是 MCP 的配置类。

参数:: url (str) -- SSE/HTTP 传输的 URL。当使用 stdio 传输时，可以填写 MCP 服务器的绝对路径。

class evofabric.core.typing.ToolInnerState(BaseModel)[源代码]#

定义单个工具的内部状态。

参数:

type (Literal["ToolInnerState"]) -- 工具内部状态类型，固定为 "ToolInnerState"。
state (Dict[str, Any]) -- 工具状态内容，格式为 {state_name: state_content}。
meta_state (Dict[str, Any]) -- 工具元状态内容，格式为 {state_name: state_content}。

class evofabric.core.typing.ToolManagerState(BaseModel)[源代码]#

定义工具管理器的整体状态。

参数:

type (Literal["ToolManagerState"]) -- 状态类型，固定为 "ToolManagerState"。
state (Dict[str, ToolInnerState]) -- 工具管理状态。键为工具名称，值为对应的 ToolInnerState。

class evofabric.core.typing.McpServerLink#

MCP 服务器连接类型别名，用于表示不同类型的 MCP 服务器通信方式。

支持的连接类型：

StdioLink
SseLink
StreamableHttpLink

使用 Annotated 和 Pydantic 的 Field 设置了类型区分：

McpServerLink = Annotated[
    Union[StdioLink, SseLink, StreamableHttpLink],
    Field(discriminator="type")
]

说明：: 该类型用于 MCP 工具管理系统中，便于在不同通信方式的服务器间进行统一管理和调用。

VectorStore#

class evofabric.core.typing.DBItem[源代码]#

用于向量存储和检索的基础数据库项目。

参数:

text (str) -- 需要向量化的文本内容
ids (Optional[str]) -- 数据库中的唯一标识符
metadata (Optional[dict]) -- 附加的元数据字典

使用示例：

from evofabric.core.typing import DBItem

# Create a database item
item = DBItem(
    text="This is a sample document text",
    ids="doc_001",
    metadata={"category": "article", "author": "John Doe"}
)

# Access item attributes
print(item.text)      # "This is a sample document text"
print(item.ids)       # "doc_001"
print(item.metadata)  # {"category": "article", "author": "John Doe"}

class evofabric.core.typing.SearchResult[源代码]#

相似性搜索返回的结果数据结构。

参数:

text (str) -- 文档文本
metadata (Optional[Dict[str, Any]]) -- 关联的元数据
score (Optional[float]) -- 相似性分数（0-1，值越高表示越相似）
id (str) -- 文档唯一标识符

使用示例：

from evofabric.core.typing import SearchResult

# Create a search result
result = SearchResult(
    text="This is a matching document",
    metadata={"category": "article", "source": "web"},
    score=0.95,
    id="result_001"
)

# Access result attributes
print(result.text)      # "This is a matching document"
print(result.score)     # 0.95
print(result.metadata)  # {"category": "article", "source": "web"}

evofabric.core.typing

目录

evofabric.core.typing#

General#

Graph Related#

Messages#

Tool#

VectorStore#