agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Remove outdated design doc (#320)

Browse Source
This commit is contained in:
Eric Zhu
2025-08-05 09:39:24 -05:00
committed by GitHub
Unverified
parent 9faa27b8eb
commit 98cf962b72
15 changed files with 0 additions and 2011 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/design/README.md
-103
View File
@@ -1,103 +0,0 @@
# Agent Framework Design Doc
What values does the framework provide?
- A set of configurable, extensible and high-quality components (e.g., model clients, tools, MCP servers and memory).
- An easy path for deploying, securing and scaling applications, both locally and in the cloud.
- Integration with tools for monitoring, debugging, evaluation and optimization, both locally and in the cloud.
- A community of developers and users for support, ideas, and contributions, benefiting everyone in the ecosystem.
What is this document?
- An overview of the new framework.
- Defining the major elements of the framework and their relationships.
- Detailed design of each element and its implementation will be in a separate document.
## Core Data Types
To unify the interaction between components, we define a set of core
data types that are used throughout the framework.
See [Core Data Types](types.md) for more details.
## Components
A component is a class that provides a specific functionality and can be used
independently by applications.
There are two types of components in the framework: agent components and agents. Agent components are the building blocks of agents, while agents are
the higher-level components, and can be composed from agent components
and other agents (as in workflows).
The framework defines the following components. Follow the links to
find the design details of each component:
- Agent Components:
- [Model Client](models.md)
- [Vector Store and Embedding Client](vector-stores.md)
- [Tool](tools.md)
- [MCP Server](mcp-servers.md)
- [Context Provider (memory, RAG, etc.)](context_providers.md)
- [Thread](threads.md)
- [Guardrail](guardrails.md)
- Agent and Workflow:
- [Agent](agents.md)
- [Workflow](workflows.md)
### Composition
Components can be composed to create complex components. For example,
an agent can be composed from model clients, tools and memory,
and a tool can be composed from an agent or a workflow.
It is the responsibility of the framework to validate components
and their composition.
### Configuration
A component can be created from a set of serializable configuration parameters,
with the help of dependency injection to resolve non-serializable dependencies.
### Relationships
The following diagram shows the component relationship of the framework:
```mermaid
graph TD
Component[Component] --> |extends| Agent[Agent]
Agent --> |extends| Workflow[Workflow]
Component --> |extends| ModelClient[Model Client]
Component --> |extends| VectorStore[Vector Store]
Component --> |extends| EmbeddingClient[Embedding Client]
Component --> |extends| Tool[Tool]
Component --> |extends| MCPServer[MCP Server]
Component --> |extends| ContextProvider[Context Provider]
Component --> |extends| Thread[Thread]
Component --> |extends| Guardrail[Guardrail]
Agent --> |uses| uses1[Model Client]
Agent --> |uses| uses2[Thread]
Agent --> |uses| uses3[Tools and MCP Servers]
Agent --> |uses| uses4[Context Provider]
Agent --> |uses| uses5[Guardrail]
Workflow --> |contains| contains[Child Agents]
ContextProvider --> |uses| uses5[Vector Store]
VectorStore --> |uses| uses6[Embedding Client]
```
## Deployment and Scaling
[Deployment](deployment.md).
## Observability and Monitoring
[Observability](observability.md).
## Evaluation
[Evaluation](evaluation.md).
## Optimization
[Optimization](optimization.md).
docs/design/agents.md
-512
View File
@@ -1,512 +0,0 @@
# Agents
An agent is a component that processes messages in a thread and returns a result.
During its handling of messages, an agent may:
- Use model client to process messages,
- Use thread to keep track of the interaction with the model,
- Invoke tools or MCP servers, and
- Retrieve and store data through memory.
It is up to the implementation of the agent class to decide how these components are used.
__An important design goal of the framework is to ensure the developer experience
of creating custom agent is as easy as possible.__ Existing frameworks
have made "kitchen-sink" agents that are hard to understand and maintain.
An agent might not use the components provided by the framework to implement
the agent interface.
Azure AI Agent is an example of such agent: its implementation is
backed by the Azure AI Agent Service.
The framework provides a set of pre-built agents:
- `ChatCompletionAgent`: an agent that uses a chat-completion model to process messages
and use thread, memory, tools and MCP servers in a configurable way. __If we can make
custom agents easy to implement, we can remove this agent.__
- `AzureAIAgent`: an agent that is backed by Azure AI Agent Service.
- `ResponsesAgent`: an agent that is backed by OpenAI's Responses API.
- `A2AAgent`: an agent that is backed by the [A2A Protocol](https://google.github.io/A2A/documentation/).
## `Agent` base class
```python
TInThread = TypeVar("TInThread", bound="AgentThread", contravariant=True)
TNewThread = TypeVar("TOutThread", bound="AgentThread", covariant=True)
class Agent(ABC, Generic[TInThread, TNewThread]):
"""The base class for all agents in the framework."""
@abstractmethod
async def run(
self,
messages: list[Message],
thread: TInThread,
context: RunContext,
) -> Result:
"""The method to run the agent on a thread of messages, and return the result.
Args:
messages: The list of new messages to process that have not been added
to the thread yet. The agent may use these messages and append
new messages to the thread as part of its processing.
thread: The thread of messages to process: it may be a local thread
or a stub thread that is backed by a remote service.
context: The context for the current invocation of the agent, providing
access to the event channel, and human-in-the-loop (HITL) features.
Returns:
The result of running the agent, which includes the final response.
"""
...
@classmethod
@abstractmethod
async def create_thread(self) -> TNewThread:
"""Create a new thread for the agent to use.
Returns:
A new thread that is compatible with the agent.
"""
...
@dataclass
class RunContext:
"""The context for the current invocation of the agent."""
event_handler: EventHandler | Callable[[Message], Awaitable[None]]
"""The event consumer for handling events emitted by the agent. Could be
a callable that takes a message and returns an awaitable, or an instance of
`EventHandler` that handles events emitted by the agent."""
user_input_source: UserInputSource
"""The user input source for requesting for user input during the agent run."""
... # Other fields, could be extended to include more for application-specific needs.
@dataclass
class Result:
"""The result of running an agent."""
final_response: Message
... # Other fields, could be extended to include more for application-specific needs.
```
## `ToolCallingAgent` example
Here is an example of a custom agent that calls a tool and returns the result.
The `ToolCallingAgent` implements the `Agent` base class and
it implements the `run` method to process incoming messages and call tools if needed.
```python
TInThread = TypeVar("TInThread", bound="ChatMessageThread", contravariant=True)
TNewThread = TypeVar("TOuthread", bound="ChatMessageThread", covariant=True)
class ToolCallingAgent(Agent[TInThread, TNewThread]):
def __init__(
self,
model_client: ModelClient,
tools: list[Tool],
) -> None:
self.model_client = model_client
self.tools = tools
async def run(self, messages: list[Message], thread: TInThread, context: RunContext) -> Result:
# Apply the messages to the thread.
await thread.on_new_messages(messages)
# Create a response using the model client, passing the thread and context.
create_result = await self.model_client.create(thread.messages, context, tools=self.tools)
# Emit the event to notify the workflow consumer of a model response.
await context.emit(ModelResponseEvent(create_result))
if create_result.is_tool_call():
# Get user approval for the tool call through the context.
approval = await context.get_user_approval(create_result.tool_calls)
if not approval:
# ... return a canned response.
# Call the tools with the tool calls in the response.
tools = ... # Find the tool by name in the tools list.
tool_results = ... # Call the tool with the tool call arguments.
# Emit the event to notify the workflow consumer of a tool call.
await context.emit(ToolCallEvent(tool_result))
# Update the thread with the tool result.
await thread.on_new_messages(tool_result.to_messages())
# Return the tool result as the response.
return Result(
final_response=tool_result,
)
else:
# Return the response as the result.
return Result(
final_response=create_result,
)
@classmethod
async def create_thread(self) -> TNewThread:
"""Create a new thread for the agent to use.
NOTE: this could be part of a new base class for this type of agent.
"""
return await ChatMessageThread.create()
```
Things to note in the implementation of the `run` method:
- Orchestration of tools and model is completly customizable.
- Components such as `thread` and `model_client` interacts smoothly with little boilerplate code.
- The `context` parameter provides convenient access to the workflow run fixtures such as event channel.
In practice, the developer likely will inherit from `ChatAgent` to
customize the `run` method, so they don't need to implement the boilerplate code
for creating a thread.
An agent doesn't need to use components provided by the framework to implement the agent interface.
For example, in a multi-agent workflow, we may need a verification agent in a using deterministic
logic to critic another agent's response.
```python
class CriticAgent(ChatAgent):
def __init__(self) -> None:
self.verification_logic = ... # Some verification logic, e.g. a set of rules.
async def run(self, messages: list[Message], thread: ChatMessageThread, context: RunContext) -> Result:
# Use the verification logic to verify the messages.
is_verified = self.verification_logic.verify(messages)
if is_verified:
final_response = Message("The response is verified.")
else:
final_response = Message("The response is not verified.")
return Result(
final_response=final_response,
)
```
## Run
A _run_ is a single invocation of the agent or a workflow given a thread of messages.
## Run agent
Developer can instantiate a subclass of `Agent` directly using it's constructor,
and run it by calling the `run` method.
```python
@FunctionTool
def my_tool(input: str) -> str:
return f"Tool result for {input}"
model_client = OpenAIChatCompletionClient("gpt-4.1")
agent = ToolCallingAgent(
model_client=model_client,
tools=[my_tool],
)
# Create a thread for the current task.
thread = await ChatMessageThread.create()
# Create a context that uses a handler that prints emitted events to the console,
# and a user input source that reads from the console.
context = RunContext(event_handler=ConsoleEventHandler(), user_input_source=ConsoleUserInputSource())
# Run the agent with the thread and context.
result = await agent.run([Message("Can you find the file 'foo.txt' for me?")], thread, context)
```
## User session
A user session is a logical concept which involves a sequence of messages exchanged between the user and the agent.
Consider the following examples:
- A chat session in ChatGPT.
- A delegation of task to a workflow agent from a user, with data exchanged between the user
and the workflow such as occassional feedbacks from the user and status updates from the workflow.
A user session may involve multiple runs.
## User session state
Rather than classifying agents as stateless or stateful, we focus on how state is managed during a user session.
There are several states that an application may maintain during a user session:
- **Conversation or workflow state**. This is the conversation history or execution
history in a workflow. This state is typically owned and managed by the thread object.
- **Long-term memory**. This can be information relevant to the user,
such as user preferences, past interactions, or other relevant data.
This can also be information relevant to the task, such as past trajectories,
past results, or other task-related data. These states are typically
owned and managed by a memory object.
The thread is always passed through the agent's `run` method.
Memory may be attached to the thread or passed to the agent's constructor.
See the [Context Providers](context_providers.md) design document for more details on how memory
and other context providers like RAG are used in the framework.
It is up to the application to decide whether to reuse state across different
user sessions. The framework should provide the necessary methods and storage layer integration
for persisting and retrieving state, but the application should decide how to use them.
## Run agent concurrently
If the agent just call models and tools that are stateless,
we can run the same instance of the agent concurrently.
```python
# Create threads for concurrent tasks.
thread1 = ChatMessageThread.create()
thread2 = ChatMessageThread.create()
# Run the agent concurrently on multiple threads.
results = await asyncio.gather(
agent.run([Message(...)], thread1, context),
agent.run([Message(...)], thread2, context),
)
# The `context`'s event handlers will emit events from both runs.
```
This is not always the right way to run concurrent agents, as some tools
or memory associated with the agent may not be concurrent-safe.
It is up the application to decide if an agent can run concurrently,
or multiple instances should be created for each thread.
## Using Foundry Agent Service
The framework offers a built-in agent class for users of the Foundry Agent Service.
The agent class essentially acts as a proxy to the agent hosted by the Foundry Agent Service.
```python
agent = FoundryAgent(
name="my_foundry_agent",
project_client="ProjectClient",
agent_id="my_agent_id", # If not provided, a new agent will be created.
deployment_name="my_deployment",
instruction="my_instruction",
... # Other parameters for the agent creation.
)
# Create a thread that is backed by the Foundry Agent Service.
thread = FoundryThread(thread_id="my_thread_id")
# Run the agent on the thread and an new context that emits events to the console.
result = await agent.run([Message(...)], thread, RunRunContext(event_channel="console"))
```
## Alternative agent abstractions
There are two alternatives:
1. **Agent with private conversation state**: The agent manages its own conversation state,
either by using a thread or other custom logics. The conversation state is
not shared with other agents or workflows. It is up to the agent to decide how
to manage the conversation state.
2. **Agent without conversation state**: The conversation state is externalized
and managed by a thread abstraction. The agent is invoked with a thread on
every run. While it can still use the thread to append messages etc., it loses
control over the conversation state the moment the run method returns.
### Protocol comparison
For agent with private conversation state, agent is invoked with new messages
and the agent is responsible for managing the conversation state while exposing
public methods for the orchestration code to manipulate its conversation state
indirectly.
```python
class Agent(ABC):
async def run(
self,
messages: list[Message],
context: RunContext,
) -> Result:
"""The method to run the agent and return the result.
Args:
messages: The list of new messages to process.
context: The context for the current invocation of the agent, providing
access to the event channel, and human-in-the-loop (HITL) features.
Returns:
The result of running the agent, which includes the final response.
"""
...
async def reset() -> None:
"""Reset the conversation state of the agent."""
...
# And other methods for managing the conversation state.
```
For agent without conversation state, the agent is invoked with a thread
and the agent is responsible for processing the messages in the thread.
```python
class Agent(ABC, Generic[TThread]):
async def run(
self,
messages: list[Message],
thread: TThread,
context: RunContext,
) -> Result:
"""The method to run the agent on a thread of messages, and return the result.
Args:
messages: The list of new messages to process.
thread: The current conversation state.
context: The context for the current invocation of the agent, providing
access to the event channel, and human-in-the-loop (HITL) features.
Returns:
The result of running the agent, which includes the final response.
"""
...
```
### Constructor comparison
For agent with private conversation state, the agent is initialized with
the a state in addition to components like model client and tools, which could be a thread passed to the constructor,
or a custom state object that the agent uses to manage its conversation state.
```python
class CustomAgent(Agent[ChatMessageThread]):
def __init__(self,
model_client: ModelClient,
tools: list[Tool],
state: CustomState, # Could be a thread or a custom state object, or nothing at all.
) -> None:
self.model_client = model_client
self.tools = tools
self.state = state # Could be created by the agent within the constructor.
```
For agent without conversation state, the agent is initialized with
the components it needs to process messages, such as a model client and tools.
```python
class CustomAgent(Agent[ChatMessageThread]):
def __init__(
self,
model_client: ModelClient,
tools: list[Tool],
) -> None:
self.model_client = model_client
self.tools = tools
```
### Thread-Agent compatibility considerations
For agent with private conversation state, compatibility with thread is not a concern,
as this is completely managed by the agent itself.
For agent without conversation state, the thread must be compatible with the agent's
`run` method. For example, a `FoundryAgent` must work with a `FoundryThread`
because the thread is backed by the Foundry Agent Service, and the implementation
requires the thread to be compatible with the service's API.
Compatibility constraints:
- `FoundryAgent` must work with `FoundryThread`.
- `OpenAIAssistantAgent` must work with `OpenAIAssistantThread`.
- `ResponsesAgent` must work with `ResponsesThread`, when using the stateful mode of the Responses API.
### Workflow-Agent compatibility considerations
For agent with private conversation state, the orchestration code cannot directly
modifies the conversation state of every agent in the workflow.
This means that for resetting the conversation state, branching a conversation,
or other orchestration logic, the agent must provides public
methods for the orchestration code to manipulate its conversation state.
Potential methods (just initial ideas):
- `reset()` to reset the conversation state.
- `branch()` to create a new branch of the conversation state from an existing state.
Example: AutoGen's MagenticOne orchestration requires the agents to be able to
reset their conversation states during re-planning. It is reasonable to expect
other types of orchestration logic will require behavior like branching
or backtracking.
For agent without conversation state, the orchestration code can directly
manipulate the thread that is passed to the agent's `run` method. So the orchestration code
can clone, fork, or reset the thread as needed.
This also means that the agent's converstion state must be abstracted as a thread.
### Extensibility considerations
For agent with private conversation state, the management of the conversation state
is completely up to the agent implementation. This means that custom agents can
be created with different conversation state management strategies, such as:
- Using a custom thread implementation that provides additional features.
- Using a custom state object that provides additional features.
When using a custom state object, the developer must also implement
methods for exporting and importing the state.
For agent without conversation state, the thread abstraction is required to
encapsulate the conversation state and ensure that the agent's `run` method
can use it without any issues. This puts a constraint on the agent implementation,
and also what can be represented as state in the thread.
Though, if the thread abstraction is designed well, it relieves the developer
from implementing the conversation state management logic themselves.
The developer only needs to come up with custom thread when the built-in thread
abstraction does not work with their custom agent.
### Discussion
- Either agent or thread must manage the conversation state.
- The class that manages the conversation state must provide a way to manipulate
it for orchestration purposes.
- Isolate thread as a separate required abstraction may introduce compatibility
issues.
- A thread abstraction with methods for manipulating the conversation state
should always be provided by the framework, whether it is exposed again
through the agent or not.
In a scenario with built-in agents and built-in threads, the developer experience
is nearly identical except for agent without conversation state the developer
must ensure the thread is compatible with the agent's `run` method.
In a scenario with custom agents and built-in threads, the developer experience
is simpler for agent without conversation state, as the thread abstraction
is already provided by the framework and the agent can use it directly. Plus,
the developer doesn't need to implement the conversation state management logic
through the agent's other methods, which will mostly likely be boilerplate code.
In a scenario with built-in agents and custom threads, the developer experience
is nearly identical, as in either case the developer must ensure
the agent's `run` method is compatible with the thread or general state object.
In a scenario with custom agents and custom threads, the developer experience
is nearly identical, as in either case the developer must ensure
the agent's `run` method is compatible with the thread or general state object,
and that the state management logic is implemented in the agent or the thread.
| Scenario | Agent with Conversation State | Agent without Conversation State |
|----------|------------------------------------------|---------------------------------------------|
| Built-in Agents, Built-in Threads | Simpler -- it should just work as there is no compatibility issue at runtime | Developer must ensure thread compatibility with agent's `run` method at runtime |
| Custom Agents, Built-in Threads | Developer must implement state management methods on the agent. | Simpler, as thread abstraction is provided by the framework and agent can use it directly |
| Built-in Agents, Custom Threads | Developer must ensure compatibility of the custom thread or state with agent's `run` method | Developer must ensure compatibility of the custom thread with agent's `run` method |
| Custom Agents, Custom Threads | Developer is fully responsible for implementing state management. | Developer is fully responsible for implementing state management. |
Overall, the agent without conversation state abstraction
provides a simpler and more consistent developer experience, as it relies on
the thread abstraction provided by the framework. The downside is that
developer must ensure the thread used is compatible with the agent's `run` method
-- this can be mitigated by enforcing strong types and validation, as well as
built-in factory methods for creating new threads given the agent type.
Another factor to consider is that Semantic Kernel already has agent abstraction
that passes a thread per invocation, so it is easier for us to migrate to the
new interface.
**Decision**: We will use the agent abstraction without conversation state
as the interface for agents in the framework.
> **We should continue to question this decision as we implement more agents and workflows, and revisit the design.**
docs/design/context_providers.md
-75
View File
@@ -1,75 +0,0 @@
# Memory, RAG, and other Context Providers
Prior to calling a model client, it's often necessary to add information to the client's context window gathered from various sources.
Two prime examples are long-term memory and retrieval-augmented generation (RAG) systems.
The ContextProvider class supports such scenarios through a unified interface for storing and retrieving context data.
## `ContextProvider` base class
```python
class ContextProvider(ABC):
"""
The base class for context providers like Memory and RAG.
Subclasses will typically have extra methods and constructor parameters for specific functionality,
such as clearing memory contents, or adding files to a RAG provider.
"""
@abstractmethod
async def get_relevant_context(self, messages: list["Message"]) -> ProvidedContext | None:
"""Searches for and returns any information relevant to the messages."""
...
@abstractmethod
async def on_new_messages(self, messages: list["Message"]) -> None:
"""Stores any information derived from the messages that may be useful to retrieve later."""
...
# To close, delete and release any runtime resources, each subclass should override the built-in Python `del` method.
```
## Usage
As an example, consider the following scenario involving long-term memory as a context provider.
Suppose that an app defines a subclass of `ContextProvider` called `Mem0Wrapper` which implements all required methods.
At runtime the app instantiates the memory provider, passing any necessary parameters to its constructor.
```python
mem = Mem0Wrapper(<params>)
```
In this example, the app then clears memory to ensure that it starts empty.
```python
mem.clear()
```
Then the app creates an agent and passes the memory provider to it through the constructor or some other method.
```python
agent.add_context_provider(mem)
```
After creating the agent, the app calls `agent.run(message, thread, run_config)` as usual,
where the user message assigns a task that requires knowledge the agent doesn't have.
`agent.run()` calls `get_relevant_context(message)` on each of the agent's context providers,
but `None` is returned since memory is empty.
Then `agent.run()` calls the model client as usual, but the LLM can't solve the task.
It may realize the information is missing, and ask the user for it.
The original user message (which assigned the task) is then added to the thread's message history as usual,
which automatically calls the `Agent.on_new_messages(messages)`,
which calls `on_new_messages` on each context provider.
In this case the memory provider fails to find any useful information in the user's message to store.
Suppose then that the user responds by supplying the missing information.
This time the agent will succeed, since the LLM's context window now contains the relevant information.
More importantly for our example, when the second user message is added to the thread's message history,
`mem.on_new_messages()` will extract and store the relevant information for later retrieval.
For this example, suppose the user then initiates a new chat (clearing the message history),
and assigns the original task again, but without providing the missing information.
This time when `mem.get_relevant_context(message)` is called, the memory provider finds the relevant information stored from the previous chat.
Then `agent.run()` attaches the retrieved information to the context window before calling the model client,
which allows the agent to succeed at the task without the user needing to repeat the missing information.
For more advanced memory implementations that have the ability to learn from their own experience
(instead of only from the user), `mem.get_relevant_context(message)` may return useful context
that was not previously extracted by `mem.on_new_messages(messages)`.
docs/design/deployment.md
View File
docs/design/evaluation.md
-129
View File
@@ -1,129 +0,0 @@
## Evaluation
The goal of Evaluation is to enable developers measure both the quality of agent responses and the efficiency of their decision-making processes.
### Core Evaluation Concepts
To enable effective evaluation (mindful of the fact that agents may be implemented with different approaches or even frameworks), it is useful to focus on the following core concepts:
- **Standardized Trajectory Format**: A unified representation of agent interactions (messages, tool calls, events) enabling consistent evaluation across different agent implementations.
- **Trajectory and Outcome Evaluation**: Analyze both the path an agent takes and the final response it generates. This includes evaluating the sequence of tool calls, the order of operations, and the final output.
### Evaluation Components
The framework provides these key evaluation components:
- **Trajectory Converter**: Transforms agent runs from various frameworks into a standardized format for evaluation.
- **Metrics Library**:
- Computation-based metrics: Direct algorithms that calculate objective measures without requiring a model
- Model-based metrics: Evaluation criteria that require an AI model to assess subjective qualities
- **Judge**: For model-based metrics, a judge is the LLM responsible for applying evaluation criteria. Different judge models can be selected based on evaluation needs.
- **Evaluator**: Coordinates the evaluation process by running computation-based metrics directly and applying judges to model-based metrics.
- **Integration**: Connect with cloud evaluation services including Azure AI Evaluation.
### (Example) Metrics
Metrics may be pointwise (evaluating a single response on some criteria) or pairwise (evaluating two responses against each other e.g., where some ground truth is available).
#### Computation-based Metrics
- **Tool Match**: Measures tool call sequence matching in various ways:
- Exact Match: Perfect match with reference sequence
- In-Order Match: Required tools called in correct order (extra steps allowed)
- Any-Order Match: All required tools called regardless of order
- **Precision**: Proportion of agent's tool calls that match reference tool calls.
- **Recall**: Proportion of reference tool calls included in the agent's tool calls.
- **Single Tool Usage**: Checks if a specific tool was used during the trajectory.
- **Tool Call Errors**: Measures rate of tool call failures or errors.
- **Latency**: Time required for agent to complete its task.
#### Model-based Metrics
- **Task Adherence**: Evaluates how well the agent's response addresses the assigned task.
- **Coherence**: Assesses logical flow and internal consistency of the response.
- **Safety**: Detects potential harmful content in responses.
- **Follows Trajectory**: Evaluates if the response logically follows from the tools used.
- **Efficiency**: Measures if the agent took an optimal path to reach the solution.
This can build on the suite of metrics provided by [Azure AI evaluation](https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/develop/agent-evaluate-sdk).
### Sample Developer Experience
**Sample Developer Experience:**
1. **Run Agent**: Execute your agent on tasks to generate trajectories.
2. **Create Trajectory**: Structure task, run data, and optional reference.
3. **Configure Metrics**: Select pre-built or custom metrics for evaluation.
4. **Evaluate**: Run evaluator to get scores and detailed results.
5. **Analyze**: Review metrics to identify improvements.
```python
from azure.ai.evaluation import AzureOpenAIModelConfiguration
from agent_framework.evaluation import (
TrajectoryMatchMetric,
TaskAdherenceMetric,
Evaluator,
Trajectory
)
# Model configuration for judge
model_config = AzureOpenAIModelConfiguration(
azure_deployment="o3-mini",
api_version="2024-02-01",
temperature=0
)
# Run your agent
task = "What's the weather in Seattle?"
run = your_agent.run(task)
# Create trajectory object
trajectory = Trajectory(
task=task,
run=run,
reference=[ # Optional reference trajectory
{"type": "tool_call", "tool": "weather_api", "args": {"location": "Seattle"}},
{"type": "response", "content": "Weather information for Seattle"}
]
)
# Define metrics
trajectory_match = TrajectoryMatchMetric(match_type="exact")
task_adherence = TaskAdherenceMetric(
criteria={
"Task adherence": (
"Does the response address the user's request and incorporate "
"information from tool calls appropriately?"
)
},
rating_rubric={
"5": "Excellent - Fully addresses task with complete detail",
"4": "Good - Addresses most aspects effectively",
"3": "Adequate - Addresses core task, minor gaps",
"2": "Poor - Partial addressing with significant gaps",
"1": "Inadequate - Fails to address task properly"
}
)
# Create evaluator
evaluator = Evaluator(
metrics=[trajectory_match, task_adherence],
model_config=model_config,
trajectory=trajectory
)
# Run evaluation
result = evaluator.run()
# Results follow Azure format
print("Evaluation Results:")
for metric_name, score in result.items():
if isinstance(score, dict):
print(f"{metric_name}: {score.get('score', 'N/A')}")
print(f" Result: {score.get('result', 'N/A')}")
print(f" Reason: {score.get('reason', 'N/A')}")
else:
print(f"{metric_name}: {score}")
```
docs/design/guardrails.md
-50
View File
@@ -1,50 +0,0 @@
# Guardrails
The design goal is to provide a flexible and extensible way to implement guardrails
and a built-in set of guardrails that can be used for common use cases.
> NOTE: this is work in progress.
Guardrails can be template-based to adapt to different input data types, which
include:
- `Message` for agent messages.
- `ToolCall` for tool call requests.
- `ToolResult` for tool call results.
Guardrails are added to other components such as `ModelClient` and `MCPServer`
as hooks that are called before and after the main logic of the component.
For example, the `ModelClient` has methods to add input and output guardrails.
```python
model_client = ModelClient(...)
model_client.add_input_guardrails([
PIIGuardrail[Message](...),
SensitiveDataGuardrail[Message](...),
])
model_client.add_output_guardrails([
HarmfulContentGuardrail[Message](...),
])
```
Another example to show how to use a guardrail with an MCP server:
```python
guardrail = PIIGuardrail(
config={
"rules": [
{
"type": "email",
"action": "block"
},
{
"type": "phone",
"action": "block"
}
]
}
)
mcp_server = MCPServer(...)
mcp_server.add_output_guardrail(guardrail)
```
docs/design/mcp-servers.md
-71
View File
@@ -1,71 +0,0 @@
# MCP Servers
An MCP server is a component that wraps a session to an
[Model Context Protocol](https://modelcontextprotocol.io/) (MCP) server.
The tools provided by MCP server should match the tool interface to ensure
minimal boilerplate code when dealing with both tools and MCP servers.
Other features like sampling and resources, should be accessible through
the MCP server interface as well.
## MCP Server base class (draft)
```python
class MCPServer(ABC):
"""The base class for all MCP servers in the framework."""
@abstractmethod
async def list_tools(self, context: Context) -> list[ToolSchema]:
"""List all available tools in the MCP server.
Returns:
A list of tool schemas available in the MCP server.
"""
...
@abstractmethod
async def call_tool(
self,
call: ToolCall,
context: Context,
) -> ToolResult:
"""Call a tool with the given name and arguments.
Args:
tool_name: The name of the tool to call.
args: The arguments to pass to the tool.
context: The context for the current invocation of the MCP server.
Returns:
The result of calling the tool.
"""
...
def add_input_guardrails(
self,
guardrails: list[InputGuardrail[ToolCall]]
) -> None:
"""Add input guardrails to the MCP server.
Args:
guardrails: The list of input guardrails to add.
"""
...
def add_output_guardrails(
self,
guardrails: list[OutputGuardrail[ToolResult]]
) -> None:
"""Add output guardrails to the MCP server.
Args:
guardrails: The list of output guardrails to add.
"""
...
```
MCP specs have other APIs. We should consider adding them as well.
docs/design/models.md
-305
View File
@@ -1,305 +0,0 @@
# Model Clients
A model client is a component that implements a unified interface for
interacting with different language models. It exposes a standardized metadata
about the model it provides (e.g., model name, tool call and vision capabilities, etc.)
to support validation and composition with other components.
The framework provides a set of pre-built model clients:
- `OpenAIChatCompletionClient`
- `AzureOpenAIChatCompletionClient`
- `AzureOpenAIResponseClient`
- `AzureAIClient`
- `AnthropicClient`
- `GeminiClient`
- `HuggingFaceClient`
- `OllamaClient`
- `VLLMClient`
- `ONNXRuntimeClient`
- `BedrockClient`
- `NIMClient`
Prompt template is a component that is used by model clients to generate prompts
with parameters set based on some injected context.
This gets into the actual interface and implementation detail of model clients,
so we just mention it here.
The design goal is to provide integration with a wide range of model providers,
including both open-source and commercial models, while maintaining a consistent
interface for developers to use.
## `ModelClient` Protocol (draft)
```python
class ModelClient(Protocol):
"""A protocol for a model client that can generate chat responses."""
async def generate_response(
self,
messages: Sequence[ChatMessage],
**options, # kwargs?
) -> ChatResponse:
"""Sends chat messages and returns the response.
Args:
messages: The sequence of chat messages to send.
**options: Additional options for the chat request, such as model_id, temperature, etc.
See `ChatOptions` for more details.
Returns:
The response messages generated by the client.
Raises:
ValueError: If the input message sequence is `None`.
"""
...
async def generate_streaming_response(
self,
messages: Sequence[ChatMessage],
**options, # kwargs?
) -> AsyncIterable[ChatResponseUpdate]:
"""Sends chat messages and streams the response.
Args:
messages: The sequence of chat messages to send.
**options: Additional options for the chat request, such as model_id, temperature, etc.
See `ChatOptions` for more details.
Returns:
An async iterable of chat response updates containing the content of the response messages
generated by the client.
Raises:
ValueError: If the input message sequence is `None`.
"""
...
def add_input_guardrails(
self,
guardrails: list[InputGuardrail[ChatMessage]]
) -> None:
"""Add input guardrails to the model client.
Args:
guardrails: The list of input guardrails to add.
"""
...
def add_output_guardrails(
self,
guardrails: list[OutputGuardrail[ChatResponse | Sequence[ChatResponseUpdate]]]
) -> None:
"""Add output guardrails to the model client.
Args:
guardrails: The list of output guardrails to add.
"""
...
```
## `ChatMessage`/ and `ChatResponse`/`StructuredChatResponse`/`ChatResponseUpdate` Types
For the data types used to represent message and response contents, see `docs/design/types.md`.
```python
class ChatMessage(BaseModel):
author_name: str | None # The name of the author of the message
contents: list[AIContent] # The contents of the message, which can include text, images, function calls, etc.
message_id: str | None # The ID of the message
raw_representation: Any | None = None # The raw representation of the chat message from an underlying implementation
role: ChatRole # The role of the author of the message
additional_properties: dict[str, Any] | None = None
class ChatRole(BaseModel):
"""Describes the intended purpose of a message within a chat interaction."""
value: str
SYSTEM: ClassVar[Self] # The role that instructs or sets the behaviour of the AI system.
USER: ClassVar[Self] # The role that provides user input for chat interactions.
ASSISTANT: ClassVar[Self] # The role that provides responses to system-instructed, user-prompted input.
TOOL: ClassVar[Self] # The role that provides additional information and references in response to tool use requests.
```
`ChatRole` is an enum-like class that defines the roles of the author in a chat message. We use a class with class variables to reduce the coupling between versions of the framework and the capabilities available in the various providers. This allows us to work with providers that support extra roles without having to change the framework code. (See also `ChatFinishReason`.)
The response types are designed to support both structured and unstructured responses, allowing for streaming in the unstructured case.
```python
class ChatResponse(BaseModel):
"""Represents the response to a chat request."""
messages: list[ChatMessage] # The chat response messages.
additional_properties: dict[str, Any] | None = None # Any additional properties associated with the chat response.
conversation_id: str | None = None # An identifier for the state of the conversation.
created_at: CreatedAtT | None = None # A timestamp for the chat response. (TODO: Use a datetime type?)
finish_reason: ChatFinishReason | None = None # The reason for the chat response.
model_id: str | None = None #The model ID used in the creation of the chat response.
raw_representation: Any | None = None # The raw representation of the chat response from an underlying implementation.
response_id: str | None = None # The ID of the chat response.
usage_details: UsageDetails | None = None # The usage details for the chat response.
TValue = TypeVar("TValue", bound=pydantic.BaseModel) # The value type needs to be deserializable (so we rely on Pydantic)
class StructuredResponse(GenericModel, Generic[TValue], ChatResponse):
"""Represents a structured response to a chat request."""
value: TValue
class ChatFinishReason(BaseModel):
"""Represents the reason a chat response completed."""
value: str
CONTENT_FILTER: ClassVar[Self] # type: ignore[assignment]
"""A ChatFinishReason representing the model filtering content, whether for safety, prohibited content,
sensitive content, or other such issues."""
LENGTH: ClassVar[Self] # type: ignore[assignment]
"""A ChatFinishReason representing the model reaching the maximum length allowed for the request and/or
response (typically in terms of tokens)."""
STOP: ClassVar[Self] # type: ignore[assignment]
"""A ChatFinishReason representing the model encountering a natural stop point or provided stop sequence."""
TOOL_CALLS: ClassVar[Self] # type: ignore[assignment]
"""A ChatFinishReason representing the model requesting the use of a tool that was defined in the request."""
```
For ease of use, all of `ChatMessage`, `ChatResponse`, `StructuredResponse` (and `ChatResponseUpdate`) provide a helper method to extract the text content. Note that all it does is concatenate any `TextContent` instances found in the contents. It will ignore any other content types, including `TextReasoningContent`.
```python
@property
def text(self) -> str:
"""Returns the concatenated text of all messages in the response."""
return " ".join(content.text for content in self.contents if isinstance(content, TextContent))
```
The streaming response type is designed to represent a "differential" of a full response object, in that the relationship between the two is akin to: `Join(ChatResponseUpdate) === ChatResponse`.
```python
class ChatResponseUpdate(BaseModel):
"""Represents a single streaming response chunk from a `ModelClient`."""
contents: list[AIContent]
"""The chat response update content items."""
additional_properties: dict[str, Any] | None = None
"""Any additional properties associated with the chat response update."""
author_name: str | None = None
"""The name of the author of the response update."""
conversation_id: str | None = None
"""An identifier for the state of the conversation of which this update is a part."""
created_at: CreatedAtT | None = None # use a datetimeoffset type?
"""A timestamp for the chat response update."""
finish_reason: ChatFinishReason | None = None
"""The finish reason for the operation."""
message_id: str | None = None
"""The ID of the message of which this update is a part."""
model_id: str | None = None
"""The model ID associated with this response update."""
raw_representation: Any | None = None
"""The raw representation of the chat response update from an underlying implementation."""
response_id: str | None = None
"""The ID of the response of which this update is a part."""
role: ChatRole | None = None
"""The role of the author of the response update."""
```
`StreamingResponseUpdate` provides helpers to join a set of updates into a single `ChatResponse` object, as well as to create the next update in a stream given the content of the new update.
## `ModelOptions`: Configuring a Model Client request
Although the `ModelClient` protocol uses named arguments to pass options to the `generate_response` and `generate_streaming_response` methods, we also provide a `ModelOptions` class representing the total set of configurable options that the user could reasonably expect on a `ModelClient` implementation. It also provides a convenient way to document them in one place.
```python
class ChatOptions(TypedDict, total=False):
"""Represents the options for a chat request.
Remarks:
This class is here for the purposes of documentation and ease of use. Options should still
be passed as keyword arguments to the `ModelClient.generate_response` and
`ModelClient.generate_streaming_response` methods.
"""
allow_multiple_tool_calls: bool | None = None
"""Indicates whether a single response is allowed to include multiple tool calls. If `False`,
the `ModelClient` is asked to return a maximum of one tool call per request. If `True`, there is
no limit. If `None`, the provider may select its own default."""
conversation_id: str | None = None
"""An optional identifier used to associate a request with an existing conversation."""
frequency_penalty: float | None = None
"""A penalty for repeated tokens in chat responses proportional to how many times they've appeared."""
max_output_tokens: int | None = None
"""The maximum number of tokens in the generated chat response."""
model_id: str | None = None
"""The model ID for the chat request."""
presence_penalty: float | None = None
"""a value that influences the probability of generated tokens appearing based on their existing
presence in generated text."""
response_format: ChatResponseFormat | None = None
"""The response format for the chat request."""
seed: int | None = None
"""A seed value used by a service to control the reproducibility of results."""
stop_sequences: list[str] | None = None
"""The list of stop sequences."""
temperature: float | None = None
"""The temperature for generating chat responses."""
tool_mode: ChatToolMode | None = None
"""The tool mode for the chat request."""
tools: list[AITool] | None = None
"""The list of tools to include with a chat request."""
top_k: int | None = None
"""The number of most probable tokens that the model considers when generating the next part of
the text."""
top_p: float | None = None
"""The 'nucleus sampling' factor (or "top p") for generating chat responses."""
```
`ChatResponseFormat` and `ChatToolMode` are used to configure the output format (structured or not) and tool use mode.
```python
class ChatResponseFormatJson(BaseModel):
"""Represents a response format for structured JSON data."""
type: Literal["json"] = "json"
schema_name: str | None = None
"""The name of the schema."""
schema_description: str | None = None
"""The description of the schema."""
schema_: dict[str, Any] | None = Field(default=None, alias="schema")
"""The JSON schema associated with the response, or `None` if there is none."""
class ChatResponseFormatText(BaseModel):
"""Represents a response format with no constraints around the format."""
type: Literal["text"] = "text"
ChatResponseFormat = ChatResponseFormatJson | ChatResponseFormatText
class AutoChatToolMode(BaseModel):
"""Indicates that a `ModelClient` is free to select any of the available tools, or none at all."""
value: Literal["auto"] = "auto"
class NoneChatToolMode(BaseModel):
"""Indicates that a `ModelClient` should not request the invocation of any tools."""
value: Literal["none"] = "none"
class RequiredChatToolMode(BaseModel):
"""Represents a mode where a chat tool must be called.
This class can optionally nominate a specific function or indicate that any of the functions can be
selected.
"""
value: Literal["require_any", "require_specific"] = "require_any"
required_function_name: str | None = None
"""The name of a specific function that must be called."""
ChatToolMode = AutoChatToolMode | NoneChatToolMode | RequiredChatToolMode
```
docs/design/observability.md
-3
View File
@@ -1,3 +0,0 @@
# Observability and Monitoring
Traces should follow the [OTEL GenAI Conventions](https://opentelemetry.io/docs/specs/semconv/gen-ai/).
docs/design/optimization.md
-10
View File
@@ -1,10 +0,0 @@
# Optimization and Tuning
> For future consideration.
The framework should support optimization of agents and workflows
with task feedback, by tuning the various components such as
system prompts, model parameters, and tool configurations.
We should also consider fine-tuning of the models and embeddings
as part of the optimization process.
docs/design/threads.md
-124
View File
@@ -1,124 +0,0 @@
# Threads
Threads are stateful objects to manage the conversation context of an agent or a workflow.
They are meant to be shown to the user as part of a user interface.
They can be persisted to a database or a file system, and used to
resume a previous user session.
Thread should use message and content types as defined in [Core Data Types](types.md).
For workflows, a thread can contain sub-threads as a dictionary of threads.
This is to ensure agents in a workflow can run concurrently on different threads.
The default thread has the key `main` and the sub-threads having keys that are usually
corresponding to the agents in a workflow.
For workflows, a thread should also support the concept of execution state, which includes:
- The history of steps taken.
- The current step in the workflow.
- The next steps to be taken.
This is to ensure the workflow can be resumed from where it left off, without losing
the state of execution.
The framework should provides default implementations of a thread class that:
- Can be backed by a database (i.e., Redis) or a file system (i.e., JSON file).
- Can be backed by the Foundry Agent Service.
- Can be copied and forked.
- Can be serialized and deserialized to/from JSON.
- Can support checkpointing, rollback, and time travel, for both agent and workflow.
- Can automantically export truncated views to be used by model clients to keep the context size within limits.
## `AgentThread` base class
```python
class AgentThread(ABC):
"""The base class for all threads defining the minimum interface."""
# ---------- Message-handling ----------
@abstractmethod
async def on_new_messages(self, messages: list["Message"]) -> None:
"""Handle a new message added to the thread."""
...
# ---------- Lifecycle management ----------
@classmethod
@abstractmethod
async def create(self) -> "AgentThread":
"""Create a new thread of the same type."""
...
# For delete and release resources, subclass should override built-in Python `del` method.
```
## `ChatMessageThread` class
The most common thread type is going to be the `ChatMessageThread`, which is a thread that stores the messages in a list. This thread type works well with `ChatCompletionAgent` and its subclasses.
```python
class ChatMessageThread(AgentThread):
"""A thread that stores the messages in a list."""
def __init__(self):
# NOTE: We should have some way to prevent direct calling of the constructor from the base class
# and enforce using the `create` class method.
if ThreadCreationContext.is_active:
raise RuntimeError("Cannot instantiate ChatMessageThread directly. Use ChatMessageThread.create() instead.")
self._messages: list["Message"] = []
@property
def messages(self) -> list["Message"]:
"""Get the list of messages in the thread."""
return self._messages
async def on_new_messages(self, messages: list["Message"]) -> None:
"""Handle a list of new messages added to the thread."""
self._messages.extend(messages)
async def fork(self, message_id: str | None = None) -> "ChatMessageThread":
"""Create a fork of the thread starting from the given message ID.
NOTE: we may need to create a new base class / protocol for this behavior.
If no message ID is provided, the fork will start from the latest message."""
new_thread = ChatMessageThread()
if message_id is None:
new_thread._messages = self._messages.copy()
else:
index = next((i for i, msg in enumerate(self._messages) if msg.id == message_id), -1)
new_thread._messages = self._messages[index + 1:] if index != -1 else []
return new_thread
@classmethod
async def create(cls) -> "ChatMessageThread":
"""Create a new chat history thread."""
with ThreadCreationContext.activate():
return cls()
async def delete(self) -> None:
"""Delete the thread. It will not be recoverable."""
self._messages.clear()
```
## `WorkflowThread` class
The `WorkflowThread` is a specialized thread that manages the execution state of a workflow. It extends the base `Thread` class and provides additional functionality to handle the workflow's execution steps and sub-threads.
```python
class WorkflowThread(AgentThread):
"""A thread that manages the execution state of a workflow."""
# ----------- Execution state management -----------
# TBD
# ----------- Lifecycle management -----------
async def create_sub_thread(self, agent: Agent, key: str) -> "AgentThread":
"""Create a sub-thread for the given agent with the given key."""
pass
async def delete_sub_thread(self, key: str) -> None:
"""Delete the sub-thread with the given key."""
pass
async def get_sub_thread(self, key: str) -> "AgentThread":
"""Get the sub-thread with the given key."""
pass
docs/design/tools.md
-215
View File
@@ -1,215 +0,0 @@
# Tools
> The design goal is to make it easy to create new tools and integrate existing APIs and make them available to agents.
A tool is a component that can be used to invoke procedure code
and returns a well-defined result type to the caller.
The result type should indicate the success or failure of the invocation,
as well as the output of the invocation in terms of the core data types.
There may be other fields in the result type for things like
side effects, etc.. We should address this when designing the
tool interface.
A tool may have arguments for invocation.
The arguments must be defined using JSON schema that language model supports.
A tool may have dependencies such as tokens, credentials,
or output message channels that will be provided by through
a context variable passed to the tool when it is invoked.
A tool may also have guardrails that are used to ensure the
tool is invoked with proper arguments, or that the agent has the
right context such as human approval to invoke the tool.
The framework provides a set of pre-built tools:
- `FunctionTool`: a tool that wraps a function.
- `AzureAISearchTool`: a tool that is backed by Azure AI Search Service.
- `OpenAPITool`: a tool that is backed by a service that defines an OpenAPI spec.
- Other tools backed by Foundry.
## `Tool` base class
```python
@dataclass
class ToolResult:
"""The result of running a tool."""
is_error: bool
output: List[ImageContent | TextContent] # The content types are defined as part of the core data types.
... # Other fields, could be extended to include more for application-specific needs.
class Tool(ABC):
"""The base class for all tools in the framework."""
@property
def name(self) -> str:
"""The name of the tool, used to identify it in the system."""
...
@property
def description(self) -> str:
"""The description of the tool, used to provide information about its
functionality.
"""
...
@property
def schema(self) -> ToolSchema:
"""The schema of the tool, which defines the JSON schema of the input
arguments."""
...
@property
def strict(self) -> bool:
"""Whether the JSON schema is in strict mode. If true, no optional
arguments are allowed.
"""
...
async def __call__(
self,
call: ToolCall,
context: Context,
) -> ToolResult:
"""The method to call to run the tool with arguments and return the result.
Args:
call: The tool call containing the name and arguments to pass to the tool.
context: The context for the current invocation of the tool, providing
access to the event channel, and human-in-the-loop (HITL) features.
Returns:
The result of running the tool.
"""
try:
# Call the on_invoke method to allow for input guardrails to be applied
# to the arguments before the tool is run.
await self.on_invoke(args, context)
# Call the run method to actually run the tool.
result = await self.run(args, context)
# Call the on_output method to handle the output of the tool.
result = await self.on_output(result, context)
except Exception as e:
# If an error occurs, call the on_error method to handle it.
result = await self.on_error(e, context)
return result
@abstractmethod
async def run(
self,
calls: ToolCall,
context: Context,
) -> ToolResult:
"""The method called by the tool itself to run the tool with arguments and return the result."""
...
async def on_invoke(
self,
calls: ToolCall,
context: Context,
) -> None:
"""The method called by the tool when is invoked but before it is run.
This is useful for input guardrails to be applied to the arguments
before the tool is run.
"""
...
async def on_error(
self,
error: Exception,
context: Context,
) -> ToolResult:
"""The method called by the tool when an error is raised."""
...
async def on_output(
self,
output: ToolResult,
context: Context,
) -> ToolResult:
"""The method called by the tool when the output is ready.
This is where output guardrails can be applied to the result
before it is returned to the caller.
"""
...
def add_input_guardrails(
self,
guardrails: list[InputGuardrail[ToolCall]]
) -> None:
"""Add input guardrails to the tool.
Args:
guardrails: The list of input guardrails to add.
"""
...
def add_output_guardrails(
self,
guardrails: list[OutputGuardrail[ToolResult]]
) -> None:
"""Add output guardrails to the tool.
Args:
guardrails: The list of output guardrails to add.
"""
...
def add_on_error_func(
self,
on_error_func: Callable[[Exception, Context], Awaitable[ToolResult]]
) -> None:
"""Add a function to call when an error is raised during the call to `run`.
Args:
on_error_func: The function to call when an error is raised.
"""
...
```
## `FunctionTool`
The `FunctionTool` is a decorator that can be used to create a tool from a function.
```python
@FunctionTool
def web_search(
query: str,
num_results: int = 10,
) -> str:
"""A tool that performs a web search and returns the results."""
...
```
`FunctionTool` supports customization of the following:
- `name`
- `description`
- `on_error_func`: the function to call when an error is raised during the call to `run`.
- `strict`: whether the JSON schema is in strict mode. If true, no optional
arguments are allowed.
- `input_guardrails`: a list of input guardrails to apply to the arguments
before the tool is run.
- `output_guardrails`: a list of output guardrails to apply to the result
before it is returned.
## `AgentTool`
The `AgentTool` is a wrapper around an agent that can be used as a tool.
```python
agent = SomeAgent(...)
tool = AgentTool(
agent=agent,
name="SomeAgent",
description="Some description of this agent tool.",
output_extractor=..., # Optional, a function to extract a ToolResult from the agent's run Result.
on_error_func=..., # Optional, a function to call when an error is raised during the call to `run`.
)
```
The argument to the `AgentTool` is a single string.
> NOTE: Do we also need to support passing a thread to the agent tool?
docs/design/types.md
-175
View File
@@ -1,175 +0,0 @@
# Core Data Types
A design goal of the new framework to simplify the interaction between agent components
through a common set of data types, minimizing boilerplate code
in the application for transforming data between components.
For example, text, images, function calls, tool schema are
all examples of such data types.
These data types are used to interact with agent components (model clients, tools, MCP, threads, and memory),
forming the connective tissue between those components.
In AutoGen, these are the data types mostly defined in `autogen_core.models` module,
and others like `autogen_core.Image` and `autogen_core.FunctionCall`. This is just
an example as AutoGen has no formal definition of model context.
To start, we should follow [MEAI](https://learn.microsoft.com/en-us/dotnet/api/microsoft.extensions.ai?view=net-9.0-pp).
This document describes the data types from Python perspective,
while for .NET, we should directly use the MEAI data types.
In Python, we will use `pydantic` to define and operate with these data types.
## Content types
The core data type is `AIContent`, which represents content used by AI services. While the Agent Framework will primarily use the abstracted types, shared across implementations of the underlying Model Client, the `raw_representation` field allows for more tightly-coupled implementations where appropriate.
```python
class AIContent(BaseModel):
raw_representation: Any | None = None
additional_properties: dict[str, Any] | None = None
```
The basic data-bearing types (`TextContent`, `TextReasoningContent`, `DataContent`, and `UriContent`) represent an interface for developers to input data into the system, as well as for agents to output resulting data. In AutoGen, messages containing these content types were considered `ChatMesssage` objects, contrasting with `AgentEvent` objects that were primarily used for internal events (such as tool calls, etc.)
```python
class TextContent(AIContent):
text: str
class TextReasoningContent(AIContent):
text: str
AnyTextualContent = TextContent | TextReasoningContent | "UriContent"
class DataContent(AIContent):
data: bytes
media_type: str
@property
def base64_data(self) -> str:
"""Returns the data represented by this instance encoded as a Base64 string."""
...
@property
def uri(self) -> str:
"""Returns the data as a data URI."""
...
class UriContent(AIContent):
uri: str
media_type: str
LocatableContent = DataContent | UriContent # TODO: Consider other names that are more descriptive.
```
Note that `TextContent` and `TextReasoningContent` do not have an inheritance relationship, and similarly `DataContent` and `UriContent` do not inherit from one-another. In this case, we opt to remain consistent with the .NET-side M.E.AI design. To abstract over both pairs, we provide the `AnyTextualContent` and `LocatableContent` type aliases.
Function calls and results have dedicated content types to represent the ModelClient's request for tool use as well as the means to flow the results back to the agent.
```python
class FunctionCallContent(AIContent):
call_id: str
name: str
arguments: dict[str, Any | None] | None = None
exception: Exception | None = None
class FunctionResultContent(AIContent):
call_id: str
result: Any | None = None
exception: Exception | None = None
```
The types contain all necessary error reporting channels (`FunctionCallContext.exception` when the raw function call request cannot be successfully mapped to the input to the tool - likely due to argument mismatch - and `FunctionResultContent.exception` when the tool execution fails). `ErrorContent` is a more general-purpose error reporting type, but one that should be used for reporting non-fatal errors. The way to think about this is that this is akin to a "400" error or a transient "500" error in HTTP parlance.
```python
class ErrorContent(AIContent):
error_code: str | None = None
details: str | None = None
message: str | None
```
Communicating the usage details when invoking a model client is important, but it has traditionally not been well supported in various frameworks, typically ending up relying on the implementor of the assistant consuming the underlying LLM to handle this. Ideally, the Model Client would provide an a-prori means to compute the usage given a message:
```python
class TokenEstimator(Protocol):
"""Estimates the count of tokens in the provided message"""
def estimate(message: ChatMessage) -> int:
...
def estimate_all(message: Sequence[ChatMessage]) -> Generator[int, None, int]
...
@runtime_checkable
class ModelClientWithEstimator(Protocol, ModelClient):
@property
def token_estimator(self) -> TokenEstimator:
"""Returns the token estimator for this model client."""
...
```
As this is not available in the current MEAI design, not generally a capability of native clients of existing hosted models, we will rely on the `UsageDetails` pattern to let the Model Client communicate this data into the Agent consuming it. Custom agents will be responsible for proper aggregation; built-in ones should provide the patterns and ideally implementation helper types. The `UsageDetails` class will provide a canonical additional operator.
```python
class UsageContent(AIContent):
"""Usage content type."""
details: UsageDetails
class UsageDetails:
"""Provides usage details about a request/response."""
input_token_count: int | None = None
"""The number of tokens in the input."""
output_token_count: int | None = None
"""The number of tokens in the output."""
total_token_count: int | None = None
"""The total number of tokens used to produce the response."""
additional_counts: AdditionalCounts | None = None
"""A dictionary of additional usage counts."""
```
The `AdditionalCounts` class is a `TypedDict` that allows for us to define well-known keys for additional counts (e.g. `thought_token_count` and `image_token_count`), while also allowing for arbitrary keys to be used by the underlying AI service prvoider. We recommend that services prefix their keys with a service identifier (e.g. `openai.`) to avoid collisions, unles using a well-known key.
```python
class AdditionalCounts(TypedDict, total=False):
"""Represents well-known additional counts for usage. This is not an exhaustive list.
Remarks:
To make it possible to avoid colisions between similarly-named, but unrelated, additional counts
between different AI services, any keys not explicitly defined here should be prefixed with the
name of the AI service, e.g., "openai." or "azure.". The separator "." was chosen because it cannot
be a legal character in a JSON key.
Over time additional counts may be added to this class.
"""
thought_token_count: int
"""The number of tokens used for thought processing."""
image_token_count: int
"""The number of token equivalents used for image processing."""
```
## Tool types
Align with the [MEAI tool types](https://learn.microsoft.com/en-us/dotnet/api/microsoft.extensions.ai.aifunction?view=net-9.0-pp)
in terms of the core attributes and methods.
See [Tools](./tools.md) for more details.
## Model client and `ChatMessage` types
Align with the MEAI model client types in terms of the core attributes and methods.
See [Models](./models.md) for more details.
docs/design/vector-stores.md
-38
View File
@@ -1,38 +0,0 @@
# Vector Stores and Embedding Clients
A vector store is component that provides a unified interface for
interacting with different vector databases, similar to model clients.
It exposes indexing and querying methods, including vector, text-based
and hybrid queries.
The details can be filled in based on the existing vector abstraction
in Semantic Kernel.
The framework provides pre-built vector stores (already exist in
Semantic Kernel):
- Azure AI Search
- Cosmos DB
- Chroma
- Couchbase
- Elasticsearch
- Faiss
- In-memory
- JDBC
- MongoDB
- Pinecone
- Postgres
- Qdrant
- Redis
- SQL Server
- SQLite
- Volatile
- Weaviate
Many vector store implementations will require embedding clients
to function. An embedding client is a component that implements a unified interface
to interact with different embedding models.
The framework provides a set of pre-built embedding clients:
- TBD.
docs/design/workflows.md
-201
View File
@@ -1,201 +0,0 @@
# Workflow
The design goal is to create workflows that can be specified in a declarative
way to allow for easy creation and modification without needing to change the
underlying code.
## `Workflow` is Agent
A `Workflow` is an agent composed of other agents. It follows the same interface
as an agent. This allows for nested workflows, where a workflow can contain other
workflows.
## Agents in a `Workflow`
Each agent (or a `Workflow`) in a `Workflow` has a thread on which it will
always run. The thread may be privated, or shared among some or all of the agents.
When do agents share a `thread`?
- When an agent is called through handoff or as a tool by another agent, the caller
agent's thread may be shared with the callee agent.
When do agents not share a `thread`?
- When a set of worker agents are called through a "fan-out" and "fan-in" pattern, where the worker
agents are called in parallel and the results are combined by an aggregator agent.
Thread sharing can be configured through the `Workflow`'s constructor.
By default, each agent has its own private thread and no sharing.
See [Threads](threads.md) for more details on how threads work.
## `Workflow` from control flow graph
A `Workflow` can be created from a control flow graph of agents.
The graph is a directed graph where each node is an agent and each edge
is a transition between agents. The graph can contain loops
and conditional transitions.
The control flow graph specifies the order in which agents are called
and the conditions under which they are called.
```python
# Create agent instances.
agent1 = MCPAgent(
model_client="OpenAIChatCompletionClient",
mcp_server=["MCPServer1", "MCPServer2"],
)
agent2 = MCPAgent(
model_client="OpenAIChatCompletionClient",
mcp_server=["MCPServer3", "MCPServer4"],
)
agent3 = MCPAgent(
model_client="OpenAIChatCompletionClient",
mcp_server="MCPServer5",
)
# Create a directed graph of agents with conditional loops and transitions.
# The graph builder validates the graph.
graph = GraphBuilder() \
.add_agent(agent1) \
.add_agent(agent2) \
.add_agent(agent3) \
.add_loop(agent1, agent2, conditions=Any(...)) \
.add_transition(agent2, agent3, conditions=Any(..., All(...))]) \
.build()
# Create a workflow from the graph.
workflow = Workflow(graph=graph)
```
## `Workflow` from message router
By default, each message is delivered to an _inbox_ of every agent in a `Workflow`.
When an agent is called, the inbox is cleared and the messages are added
to the thread that is used by the agent.
If multiple agents share a thread, each message is added exactly once to the thread.
To customize the message flow, we can configure how each inbox behaves.
Each agent's inbox can be configured to only accept messages from a specific sender(s).
We can also configure the inbox batch size, time-to-live for messages in the inbox
and various other parameters that controls how the inbox is processed.
The configuration of agents' inboxes is done using a `Router` object,
which can be built using a `RouterBuilder` object.
```python
graph = ...
router = RouterBuilder() \
.add_route(source=agent1, target=agent2) \ # Agent2 will receive messages from agent1.
.add_route(source=[agent1, agent2], target=agent3, batch_size=10, ttl="1h") \ # Agent3 will receive messages from agent1 and agent2, with a batch size of 10 and a time-to-live of 1 hour.
.add_route(source=Router.ANY, target=agent4) \ # Agent4 will receive all messages.
).build()
# Create a workflow from the graph and router.
workflow = Workflow(graph=graph, router=router)
```
You can also skip the graph all together and just create a workflow from the router.
In this case, all agents will run concurrently to process the messages delivered
to their inboxes, according to the inbox rules.
```python
# Create a workflow from the router.
workflow = Workflow(router=router)
```
The validation of the router is done as part of the workflow creation, to ensure
that no gap exists in the routing, and warning for cascading routes.
## Run `Workflow`
It is the same as running an agent.
```python
# Create a message batch to send to the workflow.
# The run context is used to pass in the event channel and other context
# shared by the agents.
thread = [
Message("Hello"),
Message("Can you find the file 'foo.txt' for me?"),
]
context = RunContext(event_channel="console")
result = await workflow.run(thread, context=context)
```
## `Workflow` has a final response
A `Workflow` is expected to have a final response, which is the final response in the
result of the last agent in the workflow. The final response is returned as part of the
`Result` object returned by the `run` method.
This is to ensure the workflow can be used in the same way as an agent.
## Stopping `Workflow`
A `workflow` may run indefinitely, so it is important to have a way to stop it.
```python
# Use a stopping condition to stop the workflow when the condition is met.
# Detail design TBD.
condition = StopCondition(
condition=Any(...),
timeout="1h",
)
workflow = Workflow(graph=graph, stop_condition=condition)
```
TBD.
## `Workflow` can be stateless
The workflow state is kept in the thread object as input to the `run` method.
If not provided, the workflow will create new sub-threads for each agent
in the workflow for their private threads, otherwise, the workflow will
use the provided sub-thread.
```python
# Create a workflow with a graph and router.
workflow = Workflow(graph=graph, router=router, stop_condition=condition)
# Create a new thread.
thread = [
Message("Hello"),
Message("Can you find the file 'foo.txt' for me?"),
]
# Run the workflow.
result = await workflow.run(thread, context=context)
# Update the thread with new messages from the user.
thread = result.thread + [
Message("Can you find the file 'bar.txt' for me?"),
]
# Resume the workflow from where it left off.
result = await workflow.run(thread, context=context)
```
Read more about [Threads](threads.md) for more details on threads.
## Pre-defined workflows
The framework ships with a few pre-defined workflows for common orchestration
patterns. These workflows can be used as-is or as a starting point for
new developers, however, when using them, you should be aware of the underlying
implementation and move on to custom workflows when a limit is reached.
The pre-defined workflows are:
- `Sequential`: A sequential workflow that calls each agent in order,
its message flow can be configured separately.
- `MapReduce`: A map-reduce workflow that splits a task into smaller
tasks, runs them in parallel and then combines the results.
- `RoundRobinGroupChat`: agents are called in a round-robin fashion in a loop.
- `SelectorGroupChat`: agents are selected on each iteration by the workflow's built-in
LLM based selector.
- `Swarm`: use handoffs.
The predefined workflows are implemented as subclasses of the `Workflow` class.