ahvn.agent.base module

class ahvn.agent.base.BaseAgentSpec(tools=None, llm_args=None, max_steps=None, **kwargs)

基类：ABC

参数:

• tools (List[ToolSpec] | None)

• llm_args (Dict | None)

• max_steps (int | None)

__init__(tools=None, llm_args=None, max_steps=None, **kwargs)

参数:

• tools (List[ToolSpec] | None)

• llm_args (Dict | None)

• max_steps (int | None)

abstractmethod encode(**inputs)

Convert input arguments into Messages for the agent.

参数:

**inputs -- Arbitrary input arguments.

返回:

The encoded messages for the agent.

返回类型:

Messages

step(messages, include=None)

Execute a single LLM call with streaming.

参数:

• messages (Union[str, Dict[str, Any], Any, List[Union[str, Dict[str, Any], Any]]]) -- Current conversation messages.

• include (Optional[List[Literal['text', 'think', 'tool_calls', 'content', 'message', 'structured', 'tool_messages', 'tool_results', 'delta_messages', 'messages']]]) -- Fields to include in the stream chunks.

生成器:

Stream chunks from the LLM.

返回类型:

Generator[Dict[str, Any], None, None]

abstractmethod is_done(messages, delta_messages)

返回类型:

Tuple[bool, Dict[str, Any]]

参数:

• messages (str | Dict[str, Any] | Any | List[str | Dict[str, Any] | Any])

• delta_messages (List[Dict[str, Any]])

user_proxy(messages, delta_messages, finish_state=None)

Add a user proxy message to prompt the agent to continue.

This is called when the agent is not done after a step, to encourage it to keep going.

参数:

• messages (Union[str, Dict[str, Any], Any, List[Union[str, Dict[str, Any], Any]]]) -- Current conversation messages.

• delta_messages (List[Dict[str, Any]]) -- Delta messages from the last step.

• finish_state (Dict[str, Any]) -- Optional finish state from the last step.

返回:

A list of messages to append.

返回类型:

Messages

abstractmethod decode(messages, finish_state=None)

返回类型:

Any

参数:

• messages (str | Dict[str, Any] | Any | List[str | Dict[str, Any] | Any])

• finish_state (Dict[str, Any])

stream(messages, include=None)

Stream the agent execution, yielding chunks as they are generated.

This is the core streaming interface. Each chunk contains: - Standard LLM fields: text, think, tool_calls, tool_messages, etc. - Agent control fields: step, done, finish_state, messages

参数:

• messages (Union[str, Dict[str, Any], Any, List[Union[str, Dict[str, Any], Any]]]) -- Initial messages to start the agent.

• include (Optional[List[Literal['text', 'think', 'tool_calls', 'content', 'message', 'structured', 'tool_messages', 'tool_results', 'delta_messages', 'messages']]]) -- Fields to include in the stream. Defaults to common fields.

生成器:

AgentStreamChunk -- Stream chunks with LLM output and agent state.

返回类型:

Generator[Dict[str, Any], None, None]

run(messages, include=None)

Run the agent to completion, collecting all stream output.

This is a convenience wrapper around stream() that blocks until completion.

参数:

• messages (Union[str, Dict[str, Any], Any, List[Union[str, Dict[str, Any], Any]]]) -- Initial messages to start the agent.

• include (Optional[List[Literal['text', 'think', 'tool_calls', 'content', 'message', 'structured', 'tool_messages', 'tool_results', 'delta_messages', 'messages']]]) -- Fields to include (passed to stream).

返回类型:

Tuple[Union[str, Dict[str, Any], Any, List[Union[str, Dict[str, Any], Any]]], Dict[str, Any]]

返回:

Tuple of (final_messages, finish_state).

__call__(**inputs)

Convenience method to encode, run, and decode in one call.

参数:

**inputs -- Input arguments passed to encode().

返回类型:

Any

返回:

Decoded output from the agent.

class ahvn.agent.base.BasePromptAgentSpec(prompt, tools=None, llm_args=None, max_steps=None, **kwargs)

基类：BaseAgentSpec

参数:

• prompt (PromptUKFT)

• tools (List[ToolSpec] | None)

• llm_args (Dict | None)

• max_steps (int | None)

__init__(prompt, tools=None, llm_args=None, max_steps=None, **kwargs)

参数:

• prompt (PromptUKFT)

• tools (List[ToolSpec] | None)

• llm_args (Dict | None)

• max_steps (int | None)

encode(**inputs)

Convert input arguments into Messages for the agent.

参数:

**inputs -- Arbitrary input arguments.

返回:

The encoded messages for the agent.

返回类型:

Messages