-
Notifications
You must be signed in to change notification settings - Fork 103
Expand file tree
/
Copy pathagent_without_subagent.py
More file actions
181 lines (149 loc) · 7.69 KB
/
agent_without_subagent.py
File metadata and controls
181 lines (149 loc) · 7.69 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
"""
Context bloat without sub-agents.
When an agent uses tools that return large outputs (file contents, search
results, database rows), all that raw data accumulates in the agent's
context window. Over multiple tool calls, this bloats the context,
increasing cost and potentially degrading performance.
This example demonstrates the problem: a single agent reads and searches
source files directly. Every line of source code flows into the agent's
context window alongside the conversation.
Diagram:
agent.run("user question")
│
▼
┌──────────────────────────────────────────────────┐
│ Code Research Agent │
│ │
│ 1. Calls list_project_files() → full listing │
│ 2. Calls read_project_file() → entire file │
│ contents added to context (repeated N times) │
│ 3. Calls search_project_files() → all matching │
│ lines added to context │
│ 4. Generates answer from bloated context │
└──────────────────────────────────────────────────┘
│
▼
response (agent saw ALL raw file contents)
Compare with agent_with_subagent.py to see how sub-agents solve this.
"""
import asyncio
import glob
import logging
import os
import sys
from typing import Annotated
from agent_framework import Agent, tool
from agent_framework.openai import OpenAIChatClient
from azure.identity.aio import DefaultAzureCredential, get_bearer_token_provider
from dotenv import load_dotenv
from pydantic import Field
from rich import print
from rich.logging import RichHandler
# ── Logging ──────────────────────────────────────────────────────────
handler = RichHandler(show_path=False, rich_tracebacks=True, show_level=False)
logging.basicConfig(level=logging.WARNING, handlers=[handler], force=True, format="%(message)s")
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
# ── OpenAI client ────────────────────────────────────────────────────
load_dotenv(override=True)
API_HOST = os.getenv("API_HOST", "github")
async_credential = None
if API_HOST == "azure":
async_credential = DefaultAzureCredential()
token_provider = get_bearer_token_provider(async_credential, "https://cognitiveservices.azure.com/.default")
client = OpenAIChatClient(
base_url=f"{os.environ['AZURE_OPENAI_ENDPOINT']}/openai/v1/",
api_key=token_provider,
model_id=os.environ["AZURE_OPENAI_CHAT_DEPLOYMENT"],
)
elif API_HOST == "github":
client = OpenAIChatClient(
base_url="https://models.github.ai/inference",
api_key=os.environ["GITHUB_TOKEN"],
model_id=os.getenv("GITHUB_MODEL", "openai/gpt-4.1-mini"),
)
else:
client = OpenAIChatClient(api_key=os.environ["OPENAI_API_KEY"], model_id=os.environ.get("OPENAI_MODEL", "gpt-4o"))
# ── Project root for file tools ──────────────────────────────────────
PROJECT_DIR = os.path.join(os.path.dirname(__file__))
# ── Tools ────────────────────────────────────────────────────────────
@tool
def list_project_files(
directory: Annotated[str, Field(description="Relative directory path within the examples folder, e.g. '.' or 'spanish'.")],
) -> str:
"""List all files in the given directory under the examples folder."""
logger.info("[📂 Tool] list_project_files('%s')", directory)
target = os.path.join(PROJECT_DIR, directory)
if not os.path.isdir(target):
return f"Error: directory '{directory}' not found."
entries = sorted(os.listdir(target))
return "\n".join(entries)
@tool
def read_project_file(
filepath: Annotated[str, Field(description="Relative file path within the examples folder, e.g. 'agent_middleware.py'.")],
) -> str:
"""Read and return the full contents of a file in the examples folder."""
logger.info("[📄 Tool] read_project_file('%s')", filepath)
target = os.path.join(PROJECT_DIR, filepath)
if not os.path.isfile(target):
return f"Error: file '{filepath}' not found."
with open(target) as f:
return f.read()
@tool
def search_project_files(
query: Annotated[str, Field(description="Text to search for (case-insensitive) across all .py files in the examples folder.")],
) -> str:
"""Search all .py files in the examples folder for lines containing the query string."""
logger.info("[🔍 Tool] search_project_files('%s')", query)
query_lower = query.lower()
results: list[str] = []
for fpath in sorted(glob.glob(os.path.join(PROJECT_DIR, "*.py"))):
relpath = os.path.relpath(fpath, PROJECT_DIR)
with open(fpath) as f:
for lineno, line in enumerate(f, 1):
if query_lower in line.lower():
results.append(f"{relpath}:{lineno}: {line.rstrip()}")
if not results:
return f"No matches found for '{query}'."
if len(results) > 50:
return "\n".join(results[:50]) + f"\n... ({len(results) - 50} more matches truncated)"
return "\n".join(results)
# ── Agent ────────────────────────────────────────────────────────────
agent = Agent(
name="coding-assistant",
client=client,
instructions=(
"You are a helpful coding assistant. You answer questions about "
"codebases, explain patterns, and help developers understand code. "
"Use the available tools to list, read, and search Python source "
"files in the project, then provide a clear, well-organized answer."
),
tools=[list_project_files, read_project_file, search_project_files],
)
# ── Query ────────────────────────────────────────────────────────────
USER_QUERY = "What different middleware patterns are used across this project? Read the relevant files to find out."
async def main() -> None:
"""Run a single query and log token usage."""
print("\n[bold]=== Code Research WITHOUT Sub-Agents ===[/bold]")
print("[dim]All file contents flow directly into the agent's context window.[/dim]\n")
print(f"[blue]User:[/blue] {USER_QUERY}")
response = await agent.run(USER_QUERY)
print(f"[green]Assistant:[/green] {response.text}\n")
# Token usage
usage = response.usage_details or {}
input_t = usage.get("input_token_count", 0) or 0
output_t = usage.get("output_token_count", 0) or 0
total_t = usage.get("total_token_count", 0) or 0
print("[bold]── Token Usage ──[/bold]")
print(f"[yellow] Assistant tokens:[/yellow] input={input_t:,} output={output_t:,} total={total_t:,}")
print()
print("[dim]All raw file contents were in the agent's context window.[/dim]")
print("[dim]Compare with agent_with_subagent.py to see context isolation in action.[/dim]\n")
if async_credential:
await async_credential.close()
if __name__ == "__main__":
if "--devui" in sys.argv:
from agent_framework.devui import serve
serve(entities=[agent], auto_open=True)
else:
asyncio.run(main())